Skip to main content

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}