cqlite_core/lib.rs
1//! CQLite Core Database Engine
2//!
3//! A high-performance, embeddable database engine with SSTable-based storage,
4//! supporting both native and WASM deployments.
5
6// Value-representation-v2 (D1, issue #1583): keep the public `Value` enum's
7// inline layout bounded. `large_enum_variant` fails the build if a future change
8// re-inlines a fat variant (e.g. un-boxing `Tombstone`/`Udt`/`Json`) instead of
9// boxing it, so the `size_of::<Value>() <= 40` pin in `types.rs` cannot silently
10// regress.
11#![deny(clippy::large_enum_variant)]
12
13pub mod config;
14pub mod cql;
15pub mod error;
16pub(crate) mod float_cmp;
17pub mod parser;
18// DISABLED FOR M1: Security and performance modules causing compilation errors
19// pub mod performance;
20// pub mod security; // Security framework for comprehensive protection
21pub mod types;
22pub mod util;
23pub mod version_hints;
24
25pub mod benchmarks;
26pub mod memory;
27// Observability foundation (epic #1031, issues #1032 + #1038). Always present;
28// the OpenTelemetry exporter wiring inside it is gated behind the optional
29// `observability` feature, and all helpers compile to no-ops when it is off.
30pub mod observability;
31pub mod platform;
32#[cfg(feature = "state_machine")]
33pub mod query;
34pub mod schema;
35pub mod storage;
36
37// Embeddable export writers (Epic #682). The module is always present; the
38// Parquet writer inside it is gated behind the optional `parquet` feature.
39pub mod export;
40
41// M5: Write engine and serialization modules (Issue #359)
42// Re-exported at crate level for convenience when write-support is enabled
43#[cfg(feature = "write-support")]
44pub use storage::serialization;
45#[cfg(feature = "write-support")]
46pub use storage::write_engine;
47
48// Ingestion module for one-shot schema & SSTable discovery (Issue #249: CLI-specific)
49#[cfg(feature = "cli-helpers")]
50pub mod ingestion;
51
52// Discovery module for SSTable scanning and coverage analysis
53#[cfg(feature = "state_machine")]
54pub mod discovery;
55
56// Testing utilities - hidden from public docs via #[doc(hidden)] but available for integration tests
57#[doc(hidden)]
58pub mod testing;
59
60// Fuzz-support surface (issue #1614). Only compiled under `--features fuzz`, and
61// `#[doc(hidden)]` even then, so the default public API and docs are unchanged.
62// Exposes thin `Result`-returning drivers over the internal decode entry points
63// for the external `fuzz/` cargo-fuzz crate. See `fuzz_support.rs`.
64#[cfg(feature = "fuzz")]
65#[doc(hidden)]
66pub mod fuzz_support;
67
68// NOTE: memory_safety_runner moved to tools/memory-safety-runner (Issue #245)
69// NOTE: the orphaned memory_safety_tests module (never compiled since MemTable
70// was removed in Issue #175) was deleted in Issue #1568 — it exercised the
71// now-deleted MemoryManager cache core.
72
73// Test-only heap-allocation probe (issue #1590, E8). Installed as the global
74// allocator for `cqlite-core`'s unit-test binary so a test can count the heap
75// allocations a specific code path performs (see the `cartesian_product`
76// allocation regression test in the SELECT executor's `lookup` module). It
77// delegates every operation to the system allocator and only bumps a per-thread
78// counter while a measurement is ACTIVE, so it is inert for every other test.
79// `not(dhat-heap)`: mutually exclusive with `DHAT_TEST_ALLOC` below (#1668) —
80// only one `#[global_allocator]` per binary.
81#[cfg(all(test, feature = "state_machine", not(feature = "dhat-heap")))] // sole `measure` caller lives in the state_machine `query` module; else `-D dead-code` under minimal (#1981)
82pub(crate) mod test_alloc_probe {
83 use std::alloc::{GlobalAlloc, Layout, System};
84 use std::cell::Cell;
85
86 thread_local! {
87 static COUNT: Cell<u64> = const { Cell::new(0) };
88 static ACTIVE: Cell<bool> = const { Cell::new(false) };
89 }
90
91 pub(crate) struct CountingAllocator;
92
93 // SAFETY: every storage operation is delegated verbatim to `System`; the only
94 // added work is bumping a `Copy` thread-local counter, which never allocates
95 // (the `thread_local!`s are `const`-initialized, so first access is
96 // alloc-free — no reentrancy into the allocator).
97 unsafe impl GlobalAlloc for CountingAllocator {
98 unsafe fn alloc(&self, layout: Layout) -> *mut u8 {
99 bump();
100 System.alloc(layout)
101 }
102 unsafe fn dealloc(&self, ptr: *mut u8, layout: Layout) {
103 System.dealloc(ptr, layout)
104 }
105 // Counting `realloc` as an allocation is deliberate: it distinguishes a
106 // `Vec::clone()` + `push()` (a clone-alloc then a grow-realloc) from a
107 // single sized `Vec::with_capacity()` fill.
108 unsafe fn realloc(&self, ptr: *mut u8, layout: Layout, new_size: usize) -> *mut u8 {
109 bump();
110 System.realloc(ptr, layout, new_size)
111 }
112 }
113
114 fn bump() {
115 let _ = ACTIVE.try_with(|a| {
116 if a.get() {
117 let _ = COUNT.try_with(|c| c.set(c.get() + 1));
118 }
119 });
120 }
121
122 /// Count the heap allocations `f` performs ON THE CURRENT THREAD (thread-local
123 /// counter, so concurrent tests do not pollute each other). Returns
124 /// `(allocations, f's result)`.
125 pub(crate) fn measure<R>(f: impl FnOnce() -> R) -> (u64, R) {
126 ACTIVE.with(|a| a.set(true));
127 COUNT.with(|c| c.set(0));
128 let r = f();
129 let n = COUNT.with(|c| c.get());
130 ACTIVE.with(|a| a.set(false));
131 (n, r)
132 }
133}
134
135#[cfg(all(test, feature = "state_machine", not(feature = "dhat-heap")))]
136#[global_allocator]
137static TEST_ALLOC: test_alloc_probe::CountingAllocator = test_alloc_probe::CountingAllocator;
138
139// Issue #1668: dhat allocator for `cqlite-core`'s unit-test binary, so an
140// in-tree `#[cfg(test)]` test can drive `pub(crate)` `StreamingMerger`
141// directly (see `merge::streaming::streaming_dhat_test`'s doc). Run with
142// `--no-default-features --features write-support,dhat-heap` (excludes
143// `state_machine`, whose own allocator this would otherwise collide with).
144#[cfg(all(test, feature = "dhat-heap"))]
145#[global_allocator]
146static DHAT_TEST_ALLOC: dhat::Alloc = dhat::Alloc;
147
148// Re-export main types for convenience
149pub use crate::{
150 config::Config,
151 error::{Error, Result},
152 platform::Platform,
153 types::*,
154};
155
156// Explicit SSTable directory refresh report (issue #1749). Not feature-gated —
157// `Database::refresh()` is available in the minimal build too.
158pub use storage::sstable::RefreshReport;
159
160// Re-export query types when state_machine feature is enabled
161#[cfg(feature = "state_machine")]
162pub use query::SchemaStatus;
163
164use std::path::Path;
165#[cfg(feature = "state_machine")]
166use std::path::PathBuf;
167use std::sync::Arc;
168
169use crate::{memory::MemoryManager, storage::StorageEngine};
170
171#[cfg(feature = "state_machine")]
172use crate::schema::SchemaManager;
173
174#[cfg(feature = "state_machine")]
175use crate::query::QueryEngine;
176
177/// Main database handle
178///
179/// This is the primary interface for interacting with a CQLite database.
180/// It coordinates between the storage engine, schema manager, and query engine.
181#[derive(Debug)]
182pub struct Database {
183 storage: Arc<StorageEngine>,
184 #[cfg(feature = "state_machine")]
185 query: Arc<QueryEngine>,
186 memory: Arc<MemoryManager>,
187 config: Config,
188}
189
190impl Database {
191 /// Open a database at the given path with the specified configuration
192 ///
193 /// # Arguments
194 ///
195 /// * `path` - The directory path where the database files will be stored
196 /// * `config` - Database configuration options
197 ///
198 /// # Errors
199 ///
200 /// Returns an error if:
201 /// - The path cannot be created or accessed
202 /// - Database files are corrupted
203 /// - Configuration is invalid
204 ///
205 /// # Examples
206 ///
207 /// ```rust,no_run
208 /// use cqlite_core::{Database, Config};
209 /// use std::path::{Path, PathBuf};
210 ///
211 /// # tokio_test::block_on(async {
212 /// let config = Config::default();
213 /// let db = Database::open(Path::new("./data"), config).await?;
214 /// # Ok::<(), Box<dyn std::error::Error>>(())
215 /// # });
216 /// ```
217 pub async fn open(path: &Path, config: Config) -> Result<Self> {
218 // Initialize platform abstraction layer
219 let platform = Arc::new(Platform::new(&config).await?);
220
221 // Initialize storage engine (no schema registry for simple open)
222 let storage = Arc::new(
223 StorageEngine::open(
224 path,
225 &config,
226 platform.clone(),
227 #[cfg(feature = "state_machine")]
228 None,
229 )
230 .await?,
231 );
232
233 // Initialize the memory-stats shell over the storage engine's live B1
234 // decompressed-chunk cache (issue #1568), so `stats()` reports real cache
235 // numbers rather than the deleted always-zero counters. When block caching
236 // is disabled (`block_cache.enabled == false`) there is no live cache, so
237 // the shell reports a structural zero.
238 let memory = Arc::new(match storage.chunk_cache() {
239 Some(cache) => MemoryManager::with_chunk_cache(cache),
240 None => MemoryManager::new(&config)?,
241 });
242
243 // Initialize schema manager
244 #[cfg(feature = "state_machine")]
245 let schema = Arc::new(SchemaManager::new_with_storage(storage.clone(), &config).await?);
246
247 // Initialize query engine (only when feature enabled)
248 #[cfg(feature = "state_machine")]
249 let query = Arc::new(QueryEngine::new(
250 storage.clone(),
251 schema.clone(),
252 memory.clone(),
253 &config,
254 )?);
255
256 Ok(Self {
257 storage,
258 #[cfg(feature = "state_machine")]
259 query,
260 memory,
261 config,
262 })
263 }
264
265 /// Open a database with pre-discovered SSTable table directories
266 ///
267 /// This method is used in the ingestion flow where SSTable discovery has been performed
268 /// externally (e.g., via `DiscoveryService`) and the database should be initialized with
269 /// specific SSTable files rather than scanning the storage directory.
270 ///
271 /// # Use Case
272 ///
273 /// This method is designed for the one-shot ingestion workflow:
274 /// 1. `DiscoveryService::discover()` scans external Cassandra data directories
275 /// 2. `SchemaManager` parses schema from discovered files
276 /// 3. `Database::open_with_discovered_sstables()` creates a queryable database instance
277 ///
278 /// # Arguments
279 ///
280 /// * `storage_path` - The directory path for database runtime files (WAL, manifest, memtable)
281 /// * `discovered_table_dirs` - Vector of table directory paths from DiscoveryService
282 /// (e.g., `/var/lib/cassandra/data/keyspace1/table1-abc123`)
283 /// * `config` - Database configuration options
284 ///
285 /// # Errors
286 ///
287 /// Returns an error if:
288 /// - The storage path cannot be created or accessed
289 /// - Any discovered table directory cannot be read
290 /// - Configuration is invalid
291 /// - Storage engine or query engine initialization fails
292 ///
293 /// # Feature Gates
294 ///
295 /// This method is only available when the `state_machine` feature is enabled (default in M2+).
296 ///
297 /// # Examples
298 ///
299 /// ```rust,no_run
300 /// use cqlite_core::{Database, Config};
301 /// use std::path::{Path, PathBuf};
302 ///
303 /// # tokio_test::block_on(async {
304 /// let config = Config::default();
305 /// let storage_path = Path::new("./runtime");
306 /// let discovered_dirs = vec![
307 /// PathBuf::from("/var/lib/cassandra/data/keyspace1/table1-abc123"),
308 /// PathBuf::from("/var/lib/cassandra/data/keyspace1/table2-def456"),
309 /// ];
310 ///
311 /// let db = Database::open_with_discovered_sstables(
312 /// storage_path,
313 /// discovered_dirs,
314 /// config
315 /// ).await?;
316 /// # Ok::<(), Box<dyn std::error::Error>>(())
317 /// # });
318 /// ```
319 #[cfg(feature = "state_machine")]
320 pub async fn open_with_discovered_sstables(
321 storage_path: &Path,
322 discovered_table_dirs: Vec<PathBuf>,
323 config: Config,
324 ) -> Result<Self> {
325 Self::open_with_discovered_sstables_and_registry(
326 storage_path,
327 discovered_table_dirs,
328 config,
329 None,
330 )
331 .await
332 }
333
334 /// Open a database with pre-discovered SSTable table directories and optional schema registry
335 ///
336 /// This is the internal implementation that supports passing a pre-loaded schema registry.
337 /// Public callers should use `open_with_discovered_sstables()` which calls this with None.
338 /// The ingestion module uses this directly to pass loaded schemas.
339 ///
340 /// # Arguments
341 ///
342 /// * `storage_path` - The directory path for database runtime files
343 /// * `discovered_table_dirs` - Vector of table directory paths from DiscoveryService
344 /// * `config` - Database configuration options
345 /// * `schema_registry` - Optional pre-loaded schema registry from ingestion
346 #[cfg(feature = "state_machine")]
347 pub(crate) async fn open_with_discovered_sstables_and_registry(
348 storage_path: &Path,
349 discovered_table_dirs: Vec<PathBuf>,
350 config: Config,
351 schema_registry: Option<Arc<tokio::sync::RwLock<schema::SchemaRegistry>>>,
352 ) -> Result<Self> {
353 // Initialize platform abstraction layer
354 let platform = Arc::new(Platform::new(&config).await?);
355
356 // Initialize storage engine with pre-discovered SSTables and schema registry
357 let storage = Arc::new(
358 StorageEngine::open_with_sstables(
359 storage_path,
360 discovered_table_dirs,
361 &config,
362 platform.clone(),
363 schema_registry.clone(),
364 )
365 .await?,
366 );
367
368 // Memory-stats shell over the storage engine's live B1 chunk cache (#1568).
369 // No live cache when block caching is disabled → structural-zero stats.
370 let memory = Arc::new(match storage.chunk_cache() {
371 Some(cache) => MemoryManager::with_chunk_cache(cache),
372 None => MemoryManager::new(&config)?,
373 });
374
375 // Initialize schema manager - use registry if provided, otherwise create empty
376 let schema = if let Some(registry_rwlock) = schema_registry {
377 Arc::new(
378 SchemaManager::new_with_registry(storage.clone(), registry_rwlock, &config).await?,
379 )
380 } else {
381 Arc::new(SchemaManager::new_with_storage(storage.clone(), &config).await?)
382 };
383
384 // Initialize query engine
385 let query = Arc::new(QueryEngine::new(
386 storage.clone(),
387 schema.clone(),
388 memory.clone(),
389 &config,
390 )?);
391
392 Ok(Self {
393 storage,
394 query,
395 memory,
396 config,
397 })
398 }
399
400 /// Re-scan the data directory and atomically apply added/removed SSTable
401 /// generations to this handle's reader set (issue #1749).
402 ///
403 /// # Freshness contract
404 ///
405 /// A `Database` is a **snapshot at [`open`](Self::open)**: it discovers the
406 /// SSTable generations once and never re-scans on its own. A Cassandra
407 /// flush/compaction (or a CQLite `--flush`) may add or remove generations
408 /// underneath a warm handle at any time; those changes become visible only
409 /// after an explicit `refresh()`. Re-runs the same TOC/filename-based
410 /// discovery `open` used — no content sniffing, no heuristics.
411 ///
412 /// - Newly present generations become queryable; removed generations stop
413 /// being queried; unchanged generations keep their warm parsed
414 /// Index/Statistics/bloom state (not rebuilt).
415 /// - **In-flight queries are never affected**: a scan already running holds
416 /// its own `Arc` reader clones and completes against the pre-refresh set. A
417 /// query issued after `refresh()` returns sees the post-refresh set.
418 /// - **Atomic and fail-closed**: if any newly discovered generation fails to
419 /// open (e.g. a corrupt `Statistics.db`, issue #1626), `refresh()` returns
420 /// the typed error and leaves the previously held reader set fully
421 /// unchanged — no partial view.
422 ///
423 /// Returns a [`RefreshReport`] describing what this call applied. Explicit
424 /// refresh only: there is no filesystem watching or per-query staleness check.
425 pub async fn refresh(&self) -> Result<RefreshReport> {
426 self.storage.refresh().await
427 }
428
429 /// Execute a SQL query and return the result
430 ///
431 /// # Arguments
432 ///
433 /// * `sql` - The SQL query string to execute
434 ///
435 /// # Errors
436 ///
437 /// Returns an error if:
438 /// - SQL syntax is invalid
439 /// - Referenced tables/columns don't exist
440 /// - Query execution fails
441 ///
442 /// # Examples
443 ///
444 /// ```rust,no_run
445 /// # use cqlite_core::{Database, Config};
446 /// # use std::path::{Path, PathBuf};
447 /// # tokio_test::block_on(async {
448 /// # let config = Config::default();
449 /// # let db = Database::open(Path::new("./data"), config).await?;
450 /// let result = db.execute("SELECT * FROM users WHERE id = 1").await?;
451 /// # Ok::<(), Box<dyn std::error::Error>>(())
452 /// # });
453 /// ```
454 #[cfg(feature = "state_machine")]
455 pub async fn execute(&self, sql: &str) -> Result<query::result::QueryResult> {
456 let result = self.query.execute(sql).await;
457
458 // Data-safety (issue #1694): log the SHAPE (rows affected), never the SQL
459 // text — a query string carries user data (WHERE-clause literals).
460 #[cfg(debug_assertions)]
461 if let Ok(ref query_result) = result {
462 tracing::debug!(
463 "Database::execute returning rows_affected: {}",
464 query_result.rows_affected
465 );
466 }
467
468 result
469 }
470
471 /// Execute a SQL query with streaming results (Issue #280)
472 ///
473 /// Returns a `QueryResultIterator` that yields rows incrementally via a bounded
474 /// channel, enabling memory-efficient processing of large result sets.
475 ///
476 /// This is the recommended method for exporting large tables, as it avoids
477 /// materializing all rows in memory at once.
478 ///
479 /// # Arguments
480 ///
481 /// * `sql` - The SQL query to execute (must be a SELECT statement)
482 /// * `config` - Streaming configuration (buffer size, chunk hints)
483 ///
484 /// # Errors
485 ///
486 /// Returns an error if:
487 /// - Query is not a SELECT statement
488 /// - SQL syntax is invalid
489 /// - Query execution fails
490 ///
491 /// # Examples
492 ///
493 /// ```rust,no_run
494 /// # use cqlite_core::{Database, Config};
495 /// # use cqlite_core::query::result::StreamingConfig;
496 /// # use std::path::Path;
497 /// # tokio_test::block_on(async {
498 /// # let db = Database::open(Path::new("./data"), Config::default()).await?;
499 /// let config = StreamingConfig::default();
500 /// let mut iter = db.execute_streaming(
501 /// "SELECT * FROM large_table",
502 /// config
503 /// ).await?;
504 ///
505 /// while let Some(row_result) = iter.next_async().await {
506 /// let row = row_result?;
507 /// // Process row incrementally
508 /// }
509 /// # Ok::<(), Box<dyn std::error::Error>>(())
510 /// # });
511 /// ```
512 #[cfg(feature = "state_machine")]
513 pub async fn execute_streaming(
514 &self,
515 sql: &str,
516 config: query::result::StreamingConfig,
517 ) -> Result<query::result::QueryResultIterator> {
518 self.query.execute_streaming(sql, config).await
519 }
520
521 /// Execute a SELECT with positional `?` parameters (Issue #961).
522 ///
523 /// The `params` are bound, in source order, into the statement's `?`
524 /// placeholders before planning, so they participate in partition-key
525 /// classification and encoding. A `WHERE pk = ?` therefore engages the same
526 /// partition-targeted fast path as the equivalent literal query.
527 ///
528 /// # Arguments
529 ///
530 /// * `sql` - A SELECT statement that may contain positional `?` placeholders
531 /// * `params` - Values bound positionally to the `?` placeholders
532 ///
533 /// # Errors
534 ///
535 /// Returns an error if the SQL is not a SELECT, the parameter count does not
536 /// match the number of `?` placeholders, or execution fails.
537 #[cfg(feature = "state_machine")]
538 pub async fn execute_with_params(
539 &self,
540 sql: &str,
541 params: &[Value],
542 ) -> Result<query::result::QueryResult> {
543 self.query.execute_with_params(sql, params).await
544 }
545
546 /// Prepare a SQL statement for repeated execution
547 ///
548 /// # Arguments
549 ///
550 /// * `sql` - The SQL statement to prepare
551 ///
552 /// # Errors
553 ///
554 /// Returns an error if SQL syntax is invalid or references non-existent objects
555 #[cfg(feature = "state_machine")]
556 pub async fn prepare(&self, sql: &str) -> Result<std::sync::Arc<query::PreparedQuery>> {
557 self.query.prepare(sql).await
558 }
559
560 /// Explain a SQL query without executing it
561 ///
562 /// # Arguments
563 ///
564 /// * `sql` - The SQL query to explain
565 ///
566 /// # Errors
567 ///
568 /// Returns an error if SQL syntax is invalid
569 #[cfg(feature = "state_machine")]
570 pub async fn explain(&self, sql: &str) -> Result<query::ExplainResult> {
571 self.query.explain(sql).await
572 }
573
574 /// Check if schema is available for a table
575 ///
576 /// This is a fast boolean check useful for pre-flight validation.
577 /// For detailed diagnostic information, use `schema_status()`.
578 ///
579 /// # Examples
580 ///
581 /// ```rust,no_run
582 /// # use cqlite_core::{Database, Config};
583 /// # tokio_test::block_on(async {
584 /// let db = Database::open(std::path::Path::new("./data"), Config::default()).await?;
585 ///
586 /// if !db.has_schema_for_table("users").await {
587 /// eprintln!("Warning: No schema found for 'users' table");
588 /// }
589 /// # Ok::<(), Box<dyn std::error::Error>>(())
590 /// # });
591 /// ```
592 #[cfg(feature = "state_machine")]
593 pub async fn has_schema_for_table(&self, table: &str) -> bool {
594 self.query.has_schema_for_table(table).await
595 }
596
597 /// Get detailed schema status for debugging
598 ///
599 /// Returns diagnostic information about schema availability including
600 /// reasons for missing schemas or extraction failures.
601 ///
602 /// # Examples
603 ///
604 /// ```rust,no_run
605 /// # use cqlite_core::{Database, Config};
606 /// # use cqlite_core::query::SchemaStatus;
607 /// # tokio_test::block_on(async {
608 /// let db = Database::open(std::path::Path::new("./data"), Config::default()).await?;
609 ///
610 /// match db.schema_status("users").await {
611 /// SchemaStatus::Available { .. } => println!("Schema ready"),
612 /// SchemaStatus::ExtractionFailed { cause, suggestion, .. } => {
613 /// eprintln!("Schema extraction failed: {}", cause);
614 /// eprintln!("Suggestion: {}", suggestion);
615 /// }
616 /// _ => {}
617 /// }
618 /// # Ok::<(), Box<dyn std::error::Error>>(())
619 /// # });
620 /// ```
621 #[cfg(feature = "state_machine")]
622 pub async fn schema_status(&self, table: &str) -> query::SchemaStatus {
623 self.query.schema_status(table).await
624 }
625
626 /// Get database statistics
627 pub async fn stats(&self) -> Result<DatabaseStats> {
628 // The chunk-cache-derived block-cache fields come from the memory manager's
629 // live handle; the per-reader B4 key caches are aggregated here (issue
630 // #1571, B5) — this async site owns `storage` and reads live readers, so
631 // the aggregate is always current rather than a stale captured handle.
632 let mut memory_stats = self.memory.stats()?;
633 let key_cache = self.storage.key_cache_stats().await;
634 memory_stats.key_cache_hits = key_cache.hits;
635 memory_stats.key_cache_misses = key_cache.misses;
636 memory_stats.key_cache_evictions = key_cache.evictions;
637 memory_stats.key_cache_invalidations = key_cache.invalidations;
638 memory_stats.key_cache_resident_bytes = key_cache.resident_bytes;
639 memory_stats.key_cache_capacity_bytes = key_cache.capacity_bytes;
640
641 Ok(DatabaseStats {
642 storage_stats: self.storage.stats().await?,
643 memory_stats,
644 #[cfg(feature = "state_machine")]
645 query_stats: self.query.stats(),
646 })
647 }
648
649 /// Flush all pending writes to disk
650 #[cfg(feature = "experimental")]
651 pub async fn flush(&self) -> Result<()> {
652 self.storage.flush().await
653 }
654
655 /// Perform manual compaction of storage files
656 #[cfg(feature = "experimental")]
657 pub async fn compact(&self) -> Result<()> {
658 self.storage.compact().await
659 }
660
661 /// Shutdown the database storage engine without consuming self.
662 ///
663 /// This is useful for language bindings where the Database is wrapped
664 /// in an Arc and cannot be consumed. The shutdown operation is idempotent.
665 ///
666 /// For consuming close that also drops the Database, use `close()`.
667 pub async fn shutdown(&self) -> Result<()> {
668 self.storage.shutdown().await
669 }
670
671 /// Close the database and release all resources
672 ///
673 /// This method ensures all pending operations are completed and
674 /// all resources are properly cleaned up.
675 ///
676 /// ## Durability contract
677 ///
678 /// Embedders MUST call `close().await` for a graceful shutdown. `Drop` is
679 /// NOT a flush — Tokio has no async drop, so dropping a handle cannot await
680 /// a flush and any un-flushed writer state is left to recovery (WAL replay)
681 /// rather than being persisted here. For the write path this maps onto
682 /// [`storage::write_engine::WriteEngine::close`], which is the actual
683 /// memtable-to-SSTable durability boundary (issue #1693).
684 pub async fn close(self) -> Result<()> {
685 // Stop background tasks. There is nothing to flush: the WAL/MemTable write
686 // path was removed in Issue #175, so `StorageEngine::flush` is an
687 // always-erroring stub. Calling it here made `close()` fail unconditionally
688 // under the `experimental` feature; the shutdown above is the full teardown.
689 self.storage.shutdown().await?;
690 Ok(())
691 }
692
693 /// Get the database configuration
694 pub fn config(&self) -> &Config {
695 &self.config
696 }
697}
698
699impl Clone for Database {
700 fn clone(&self) -> Self {
701 Self {
702 storage: self.storage.clone(),
703 #[cfg(feature = "state_machine")]
704 query: self.query.clone(),
705 memory: self.memory.clone(),
706 config: self.config.clone(),
707 }
708 }
709}
710
711/// Database statistics
712#[derive(Debug, Clone)]
713pub struct DatabaseStats {
714 /// Storage engine statistics
715 pub storage_stats: storage::StorageStats,
716 /// Memory manager statistics
717 pub memory_stats: memory::MemoryStats,
718 /// Query engine statistics
719 #[cfg(feature = "state_machine")]
720 pub query_stats: query::QueryStats,
721}
722
723/// A prepared SQL statement that can be executed multiple times
724#[cfg(feature = "state_machine")]
725#[derive(Debug)]
726pub struct PreparedStatement {
727 statement: query::PreparedQuery,
728}
729
730#[cfg(feature = "state_machine")]
731impl PreparedStatement {
732 /// Execute the prepared statement with the given parameters
733 pub async fn execute(&self, params: &[Value]) -> Result<query::result::QueryResult> {
734 self.statement.execute(params).await
735 }
736}
737
738// Re-export query result types for convenience
739#[cfg(feature = "state_machine")]
740pub use query::result::{QueryResult, QueryRow};
741
742#[cfg(test)]
743mod tests {
744 use super::*;
745 use tempfile::TempDir;
746
747 #[tokio::test]
748 async fn test_database_open_close() {
749 let temp_dir = TempDir::new().unwrap();
750 let config = Config::test_config();
751
752 let db = Database::open(temp_dir.path(), config).await.unwrap();
753 db.close().await.unwrap();
754 }
755
756 /// Documents that open_with_discovered_sstables_and_registry is crate-private.
757 /// This test exists to document the API contract - the function should NOT be
758 /// callable from integration tests or external crates.
759 #[cfg(feature = "state_machine")]
760 #[test]
761 fn test_open_with_discovered_sstables_and_registry_is_crate_private() {
762 // This test compiling proves the function exists and is accessible within the crate
763 // If we accidentally made it pub instead of pub(crate), integration tests could access it
764 // The function signature itself enforces this via pub(crate) keyword
765
766 // Note: We don't actually call the function here since it requires async setup
767 // The mere existence of this test documents the API boundary
768 assert!(
769 true,
770 "open_with_discovered_sstables_and_registry is correctly marked pub(crate)"
771 );
772 }
773
774 #[tokio::test]
775 #[cfg(feature = "state_machine")]
776 async fn test_database_open_with_discovered_sstables() {
777 let temp_dir = TempDir::new().unwrap();
778 let config = Config::test_config();
779
780 // Create an empty list of discovered table directories
781 let discovered_dirs = Vec::new();
782
783 let db = Database::open_with_discovered_sstables(temp_dir.path(), discovered_dirs, config)
784 .await
785 .unwrap();
786
787 // Verify database was created successfully
788 let stats = db.stats().await.unwrap();
789 assert_eq!(stats.storage_stats.sstables.sstable_count, 0);
790
791 db.close().await.unwrap();
792 }
793
794 // NOTE: `test_database_basic_operations` (CREATE TABLE → INSERT → SELECT) was
795 // removed in Issue #1880. It asserted the row-count of data inserted via the
796 // write path, which was deleted in Issue #175 (`execute` on an INSERT now
797 // returns `UnsupportedFormat`), so the test could only ever panic under
798 // `--all-features`. Read-path SELECT coverage lives in the real-SSTable
799 // integration/parity tests; open/close lifecycle is covered above.
800}