Skip to main content

cqlite_core/storage/
mod.rs

1//! Storage engine implementation for CQLite
2
3/// Shared, bytes-bounded, sharded decompressed-chunk cache (issue #1567).
4pub mod cache;
5/// Cooperative cancellation for long-running synchronous scans (issue #2264).
6pub mod scan_cancel;
7pub mod sstable;
8
9/// Cassandra 5.0 CommitLog segment reader (issue #2389) — a second on-disk
10/// Cassandra format, sibling of `sstable` and `write_engine`.
11pub mod commitlog;
12
13// Canonical partition-key (de)serialization, shared by the read (query) path
14// and the write engine so the two never drift (Issue #586). Always compiled —
15// the scan path needs it even without `write-support`.
16pub mod partition_key_codec;
17
18// M5: Write engine and serialization (Issue #359)
19#[cfg(feature = "write-support")]
20pub mod serialization;
21#[cfg(feature = "write-support")]
22pub mod write_engine;
23
24// REPL data access components (Issue #249: CLI-specific)
25#[cfg(feature = "cli-helpers")]
26pub mod repl_data_api;
27pub mod schema_discovery;
28pub mod sstable_data_manager;
29
30use std::path::{Path, PathBuf};
31use std::sync::Arc;
32#[cfg(feature = "state_machine")]
33use tokio::sync::RwLock;
34
35use crate::platform::Platform;
36use crate::{
37    types::{CellWriteMetadata, TableId},
38    Config, Result, RowKey, ScanRow,
39};
40// `Value` is only referenced by the experimental write API (`put` / `BatchOperation`);
41// gate the import so the default build does not flag it unused (issue #1334).
42#[cfg(feature = "experimental")]
43use crate::types::Value;
44
45// Test/feature-gated whole-table-scan invocation counter (issue #1691).
46//
47// Counts entries into the two whole-table scan initiators — `StorageEngine::scan`
48// (materializing) and `StorageEngine::scan_stream` (bounded streaming). It exists to
49// pin the retirement of `execute_parallel_table_scan`: a `TableScan` plan must issue
50// exactly ONE whole-table pass, not the 4× duplicate passes the retired multi-worker
51// path produced.
52//
53// It is thread-local, mirroring the "thread-local invocation counter reflects exactly
54// this future's calls" pattern (issue #831 / `access_path`). A test drives its
55// measured operation on a current-thread Tokio runtime, so both `scan` and
56// `scan_stream` increment on the *calling* thread (the count records the entry, not
57// the spawned producer's work). This isolates the count from the thousands of other
58// lib tests running on their own threads — a process-global atomic would be polluted
59// by any concurrent test that scans. Zero-overhead in release: the body compiles to a
60// no-op and the cell is not even linked.
61#[cfg(any(test, feature = "work-counters"))]
62thread_local! {
63    static TABLE_SCAN_CALLS: std::cell::Cell<u64> = const { std::cell::Cell::new(0) };
64}
65
66/// Record one whole-table scan initiation (`scan` / `scan_stream`) on the current
67/// thread. Unconditional at the call site; the body is a no-op in release builds
68/// (issue #1691).
69#[inline(always)]
70pub(crate) fn record_table_scan_call() {
71    #[cfg(any(test, feature = "work-counters"))]
72    TABLE_SCAN_CALLS.with(|c| c.set(c.get().saturating_add(1)));
73}
74
75/// Number of whole-table scan initiations on the current thread since the last
76/// [`reset_table_scan_calls`] (issue #1691; test/feature builds only).
77#[cfg(any(test, feature = "work-counters"))]
78pub fn table_scan_call_count() -> u64 {
79    TABLE_SCAN_CALLS.with(|c| c.get())
80}
81
82/// Clear the current thread's whole-table scan counter before a measured operation
83/// (issue #1691; test/feature builds only).
84#[cfg(any(test, feature = "work-counters"))]
85pub fn reset_table_scan_calls() {
86    TABLE_SCAN_CALLS.with(|c| c.set(0));
87}
88
89/// Main storage engine that coordinates all storage components
90///
91/// NOTE: Issue #176 removed write infrastructure (compaction, manifest).
92/// This is now a read-only storage layer focused on SSTable access.
93#[derive(Debug)]
94pub struct StorageEngine {
95    /// SSTable manager for persistent storage
96    sstables: Arc<sstable::SSTableManager>,
97
98    /// Platform abstraction
99    #[allow(dead_code)]
100    _platform: Arc<Platform>,
101
102    /// Storage configuration
103    #[allow(dead_code)]
104    config: Config,
105
106    /// Schema registry for schema-aware operations (feature-gated)
107    #[cfg(feature = "state_machine")]
108    schema_registry: Arc<RwLock<Option<Arc<RwLock<crate::schema::SchemaRegistry>>>>>,
109}
110
111impl StorageEngine {
112    /// Open a storage engine at the given path
113    ///
114    /// This method discovers SSTables by scanning the storage directory.
115    /// For pre-discovered SSTables, use `open_with_sstables` instead.
116    ///
117    /// NOTE: Issue #176 removed write infrastructure (compaction, manifest).
118    /// This is now a read-only storage layer focused on SSTable access.
119    // `skip_all` (not an explicit skip list): the `schema_registry` parameter is
120    // `#[cfg(feature = "state_machine")]`-gated, so naming it in `skip(...)` would
121    // reference a nonexistent binding in the minimal build. `skip_all` records no
122    // args as fields regardless of cfg; the `fields(...)` below are set in the body.
123    #[tracing::instrument(
124        name = "storage.engine.open",
125        level = "debug",
126        skip_all,
127        fields(sstables = tracing::field::Empty, bytes = tracing::field::Empty)
128    )]
129    pub async fn open(
130        path: &Path,
131        config: &Config,
132        platform: Arc<Platform>,
133        #[cfg(feature = "state_machine")] schema_registry: Option<
134            Arc<RwLock<crate::schema::SchemaRegistry>>,
135        >,
136    ) -> Result<Self> {
137        // Create storage directory if it doesn't exist
138        crate::observability::record_result("reader", platform.fs().create_dir_all(path).await)?;
139
140        // Initialize SSTable manager with schema registry
141        let sstables = Arc::new(crate::observability::record_result(
142            "reader",
143            sstable::SSTableManager::new(
144                path,
145                config,
146                platform.clone(),
147                #[cfg(feature = "state_machine")]
148                schema_registry.clone(),
149            )
150            .await,
151        )?);
152
153        Self::record_discovery_metrics(&sstables).await;
154
155        Ok(Self {
156            sstables,
157            _platform: platform,
158            config: config.clone(),
159            #[cfg(feature = "state_machine")]
160            schema_registry: Arc::new(RwLock::new(schema_registry)),
161        })
162    }
163
164    /// Emit SSTable-discovery telemetry (issue #1034) for a freshly built
165    /// manager: total SSTables discovered, their on-disk byte total, and the
166    /// number of logical tables. Best-effort — a stats failure is logged and the
167    /// open continues, since telemetry must never change open behaviour.
168    async fn record_discovery_metrics(sstables: &sstable::SSTableManager) {
169        use crate::observability::{self as obs, catalog};
170        match sstables.stats().await {
171            Ok(stats) => {
172                tracing::Span::current().record("sstables", stats.sstable_count as u64);
173                tracing::Span::current().record("bytes", stats.total_size);
174                obs::add_counter(
175                    catalog::STORAGE_OPEN_SSTABLES,
176                    stats.sstable_count as u64,
177                    &[],
178                );
179                obs::add_counter(catalog::STORAGE_OPEN_BYTES, stats.total_size, &[]);
180                obs::add_counter(catalog::STORAGE_OPEN_TABLES, stats.total_tables, &[]);
181            }
182            Err(e) => {
183                tracing::debug!("storage.engine.open: discovery metrics unavailable: {}", e);
184            }
185        }
186    }
187
188    /// Open a storage engine with pre-discovered SSTable table directories
189    ///
190    /// This method is used when SSTables have been discovered externally (e.g., by DiscoveryService)
191    /// and allows the storage engine to be initialized with specific table directories rather than
192    /// scanning the storage directory. Each table directory will be scanned for Data.db files.
193    ///
194    /// # Arguments
195    /// * `path` - Base storage path for manifest and SSTable operations
196    /// * `discovered_table_dirs` - Vector of table directory paths (each containing SSTable files)
197    /// * `config` - Storage configuration
198    /// * `platform` - Platform abstraction for I/O operations
199    ///
200    /// # Returns
201    /// A StorageEngine instance with all components initialized, including SSTable readers
202    /// for all Data.db files found in the discovered table directories.
203    ///
204    /// # Example
205    /// ```no_run
206    /// # use std::path::{Path, PathBuf};
207    /// # use std::sync::Arc;
208    /// # use cqlite_core::{Config, Platform, storage::StorageEngine};
209    /// # async fn example() -> cqlite_core::Result<()> {
210    /// let config = Config::default();
211    /// let platform = Arc::new(Platform::new(&config).await?);
212    /// let storage_path = Path::new("/var/lib/cqlite/storage");
213    /// let discovered_table_dirs = vec![
214    ///     PathBuf::from("/var/lib/cassandra/keyspace1/table1-abc123"),
215    ///     PathBuf::from("/var/lib/cassandra/keyspace1/table2-def456"),
216    /// ];
217    ///
218    /// let engine = StorageEngine::open_with_sstables(
219    ///     storage_path,
220    ///     discovered_table_dirs,
221    ///     &config,
222    ///     platform,
223    ///     #[cfg(feature = "state_machine")]
224    ///     None,
225    /// ).await?;
226    /// # Ok(())
227    /// # }
228    /// ```
229    // `skip_all`: see the note on `open` — the cfg-gated `schema_registry` cannot
230    // be named in an explicit `skip(...)` without breaking the minimal build.
231    #[tracing::instrument(
232        name = "storage.engine.open",
233        level = "debug",
234        skip_all,
235        fields(sstables = tracing::field::Empty, bytes = tracing::field::Empty)
236    )]
237    pub async fn open_with_sstables(
238        path: &Path,
239        discovered_table_dirs: Vec<PathBuf>,
240        config: &Config,
241        platform: Arc<Platform>,
242        #[cfg(feature = "state_machine")] schema_registry: Option<
243            Arc<RwLock<crate::schema::SchemaRegistry>>,
244        >,
245    ) -> Result<Self> {
246        // Create storage directory if it doesn't exist
247        crate::observability::record_result("reader", platform.fs().create_dir_all(path).await)?;
248
249        // Initialize SSTable manager with pre-discovered paths and schema registry
250        let sstables = Arc::new(crate::observability::record_result(
251            "reader",
252            sstable::SSTableManager::new_from_discovered_paths(
253                path,
254                discovered_table_dirs,
255                config,
256                platform.clone(),
257                #[cfg(feature = "state_machine")]
258                schema_registry.clone(),
259            )
260            .await,
261        )?);
262
263        Self::record_discovery_metrics(&sstables).await;
264
265        Ok(Self {
266            sstables,
267            _platform: platform,
268            config: config.clone(),
269            #[cfg(feature = "state_machine")]
270            schema_registry: Arc::new(RwLock::new(schema_registry)),
271        })
272    }
273
274    /// Insert a key-value pair
275    ///
276    /// NOTE: Write functionality removed in Issue #175 (WAL/MemTable infrastructure deleted).
277    /// This method is feature-gated behind 'experimental' but currently unimplemented.
278    #[cfg(feature = "experimental")]
279    pub async fn put(&self, _table_id: &TableId, _key: RowKey, _value: Value) -> Result<()> {
280        Err(crate::error::Error::UnsupportedFormat(
281            "Write operations (put) removed in Issue #175 - WAL and MemTable infrastructure deleted".to_string()
282        ))
283    }
284
285    /// Get a value by key
286    pub async fn get(&self, table_id: &TableId, key: &RowKey) -> Result<Option<ScanRow>> {
287        // Check SSTables
288        self.sstables.get(table_id, key).await
289    }
290
291    /// Re-scan the data directory and atomically apply added/removed SSTable
292    /// generations to the held reader set (issue #1749).
293    ///
294    /// # Freshness contract
295    ///
296    /// A `StorageEngine` (and the [`Database`](crate::Database) built on it)
297    /// snapshots the discovered SSTable generations **at open**; it does not
298    /// re-scan on its own. This is the ONLY way the reader set changes for a
299    /// long-lived handle. Re-runs the same TOC/filename-based discovery `open`
300    /// used — no content sniffing.
301    ///
302    /// - Added generations become queryable; removed generations stop being
303    ///   queried; unchanged generations keep their warm parsed state.
304    /// - **In-flight queries are unaffected**: a scan already running holds its
305    ///   own `Arc` reader clones and completes against the pre-refresh set;
306    ///   queries started after the refresh see the new set.
307    /// - **Atomic / fail-closed**: if any newly discovered generation fails to
308    ///   open (e.g. a corrupt `Statistics.db`, issue #1626), the typed error is
309    ///   returned and the previously held reader set is left fully unchanged.
310    pub async fn refresh(&self) -> Result<sstable::RefreshReport> {
311        self.sstables.refresh_tables().await
312    }
313
314    /// Delete a key
315    ///
316    /// NOTE: Write functionality removed in Issue #175 (WAL/MemTable infrastructure deleted).
317    /// This method is feature-gated behind 'experimental' but currently unimplemented.
318    #[cfg(feature = "experimental")]
319    pub async fn delete(&self, _table_id: &TableId, _key: RowKey) -> Result<()> {
320        Err(crate::error::Error::UnsupportedFormat(
321            "Write operations (delete) removed in Issue #175 - WAL and MemTable infrastructure deleted".to_string()
322        ))
323    }
324
325    /// Scan a range of keys
326    ///
327    /// # Arguments
328    /// * `table_id` - The table to scan
329    /// * `start_key` - Optional start key for range scan
330    /// * `end_key` - Optional end key for range scan
331    /// * `limit` - Optional limit on number of results
332    /// * `schema` - Optional table schema for schema-aware parsing. When provided,
333    ///   enables accurate type detection and avoids heuristic-based parsing.
334    ///   Strongly recommended for Cassandra 5.0+ formats.
335    pub async fn scan(
336        &self,
337        table_id: &TableId,
338        start_key: Option<&RowKey>,
339        end_key: Option<&RowKey>,
340        limit: Option<usize>,
341        schema: Option<&crate::schema::TableSchema>,
342    ) -> Result<Vec<(RowKey, ScanRow)>> {
343        record_table_scan_call();
344        // Scan SSTables directly
345        self.sstables
346            .scan(table_id, start_key, end_key, limit, schema)
347            .await
348    }
349
350    /// Partition-targeted scan for a fully-constrained `WHERE pk = ?` (Issue #949).
351    ///
352    /// Returns only the rows for the single partition identified by the raw
353    /// `partition_key` bytes, after pruning the SSTable set down to those whose
354    /// bloom filter / BTI trie admit the key — so unrelated SSTables are never
355    /// parsed. Output matches filtering the full [`scan`](Self::scan) result to the
356    /// partition. Delegates to [`SSTableManager::scan_partition`] (which has a
357    /// bloom-prune implementation for the default build and a scan-and-filter
358    /// fallback for the `tombstones` build, so callers need no cfg branching).
359    ///
360    /// Returns `(rows, engaged)`: `engaged` is `true` only when the call actually
361    /// pruned the SSTable set to partition candidates (the default build). The
362    /// `tombstones` build returns `false` because it full-scans and retains with no
363    /// prune, so the caller reports an honest fallback access path (Epic #951).
364    pub async fn scan_partition(
365        &self,
366        table_id: &TableId,
367        partition_key: &[u8],
368        schema: Option<&crate::schema::TableSchema>,
369    ) -> Result<(Vec<(RowKey, ScanRow)>, bool)> {
370        self.sstables
371            .scan_partition(table_id, partition_key, schema)
372            .await
373    }
374
375    /// Resolve the AUTHORITATIVE partition-key shape for `table_id` from the
376    /// SSTable Statistics.db SerializationHeader (issue #1750).
377    ///
378    /// Used by the SCHEMA-LESS point-read classifier to confirm — from authoritative
379    /// metadata, never a synthesised pk name — that a `col = <literal>` predicate
380    /// column really is the sole partition key before taking a by-key seek. Returns
381    /// `None` when no reader exposes a SerializationHeader; delegates to
382    /// [`SSTableManager::partition_key_shape`].
383    pub async fn partition_key_shape(
384        &self,
385        table_id: &TableId,
386    ) -> Option<sstable::PartitionKeyShape> {
387        self.sstables.partition_key_shape(table_id).await
388    }
389
390    /// Clustering-slice-aware partition-targeted scan (Issue #954, Epic #951).
391    ///
392    /// Like [`scan_partition`](Self::scan_partition) but pushes a single-column
393    /// clustering-key restriction (`ck </>/= ?` / two-bound range) down to a
394    /// within-partition seek when the candidate's authoritative row index supports
395    /// it, so a wide-partition slice decodes O(matched rows + index) rather than
396    /// the whole partition. Returns `(rows, clustering_seek_engaged)`: the rows are
397    /// the full partition (or a clustering-narrowed superset) so the caller's
398    /// post-scan filter yields byte-identical output, and the bool reports whether
399    /// the clustering narrowing actually engaged (for the `ClusteringSlice` access
400    /// path). Delegates to [`SSTableManager::scan_partition_clustering`].
401    #[cfg(not(feature = "tombstones"))]
402    pub async fn scan_partition_clustering(
403        &self,
404        table_id: &TableId,
405        partition_key: &[u8],
406        clustering: Option<&crate::storage::sstable::reader::ClusteringSlice>,
407        schema: Option<&crate::schema::TableSchema>,
408    ) -> Result<(Vec<(RowKey, ScanRow)>, bool)> {
409        self.sstables
410            .scan_partition_clustering(table_id, partition_key, clustering, schema)
411            .await
412    }
413
414    /// Reverse single-partition clustering scan for a BIG (`nb`) wide partition
415    /// (Issue #1184). Returns `Ok(Some(rows))` in DESCENDING clustering order when
416    /// the BIG promoted-index reverse iterator applied, or `Ok(None)` to tell the
417    /// caller to keep the in-memory `ORDER BY DESC` sort (small / BTI /
418    /// multi-generation cases). Delegates to
419    /// [`SSTableManager::scan_partition_clustering_reverse`].
420    #[cfg(not(feature = "tombstones"))]
421    pub async fn scan_partition_clustering_reverse(
422        &self,
423        table_id: &TableId,
424        partition_key: &[u8],
425        schema: Option<&crate::schema::TableSchema>,
426    ) -> Result<Option<Vec<(RowKey, ScanRow)>>> {
427        self.sstables
428            .scan_partition_clustering_reverse(table_id, partition_key, schema)
429            .await
430    }
431
432    /// Partition-targeted, metadata-carrying scan for a fully-constrained
433    /// `WHERE pk = ?` WRITETIME/TTL projection (Issue #962).
434    ///
435    /// The metadata sibling of [`scan_partition`](Self::scan_partition): returns
436    /// only the rows for the single partition identified by the raw
437    /// `partition_key` bytes, WITH per-cell write metadata, after pruning the
438    /// SSTable set down to the candidates whose bloom filter / BTI trie admit the
439    /// key — so a `SELECT WRITETIME(col) ... WHERE pk = ?` never opens all N
440    /// SSTables. Output matches filtering the full
441    /// [`scan_with_cell_metadata`](Self::scan_with_cell_metadata) result to the
442    /// partition; cross-generation reconciliation runs over the pruned candidates.
443    /// Delegates to [`SSTableManager::scan_partition_with_cell_metadata`].
444    ///
445    /// Returns `(rows, engaged)`: `engaged` is `true` only when the call pruned the
446    /// SSTable set to partition candidates (the default build). The `tombstones`
447    /// build returns `false` because it full-scans with metadata and retains with no
448    /// prune, so the caller reports an honest fallback access path (Epic #951).
449    pub async fn scan_partition_with_cell_metadata(
450        &self,
451        table_id: &TableId,
452        partition_key: &[u8],
453        schema: Option<&crate::schema::TableSchema>,
454    ) -> Result<(
455        Vec<(
456            RowKey,
457            ScanRow,
458            std::collections::HashMap<String, CellWriteMetadata>,
459        )>,
460        bool,
461    )> {
462        self.sstables
463            .scan_partition_with_cell_metadata(table_id, partition_key, schema)
464            .await
465    }
466
467    /// Scan a table and return per-cell write metadata alongside row values.
468    ///
469    /// Delegates to [`SSTableManager::scan_with_cell_metadata`].  Used when
470    /// `ProjectionFlags::include_cell_metadata` is set (issue #693).
471    pub async fn scan_with_cell_metadata(
472        &self,
473        table_id: &TableId,
474        start_key: Option<&RowKey>,
475        end_key: Option<&RowKey>,
476        limit: Option<usize>,
477        schema: Option<&crate::schema::TableSchema>,
478    ) -> Result<
479        Vec<(
480            RowKey,
481            ScanRow,
482            std::collections::HashMap<String, CellWriteMetadata>,
483        )>,
484    > {
485        self.sstables
486            .scan_with_cell_metadata(table_id, start_key, end_key, limit, schema)
487            .await
488    }
489
490    /// Streaming scan (issue #790): return a bounded channel that yields
491    /// `(RowKey, ScanRow)` entries lazily in key (token) order, instead of the
492    /// materializing [`scan`](Self::scan) that returns the whole `Vec`.
493    ///
494    /// Live heap is bounded by `buffer_size` rows rather than growing O(rows),
495    /// so streaming a large `SELECT *` no longer holds the entire result set in
496    /// memory at once. Delegates to [`SSTableManager::scan_stream`].
497    ///
498    /// [`SSTableManager::scan_stream`]: sstable::SSTableManager::scan_stream
499    pub async fn scan_stream(
500        &self,
501        table_id: &TableId,
502        start_key: Option<&RowKey>,
503        end_key: Option<&RowKey>,
504        schema: Option<&crate::schema::TableSchema>,
505        buffer_size: usize,
506    ) -> Result<tokio::sync::mpsc::Receiver<Result<(RowKey, ScanRow)>>> {
507        record_table_scan_call();
508        self.sstables
509            .scan_stream(table_id, start_key, end_key, schema, buffer_size)
510            .await
511    }
512
513    /// Batched streaming scan (issue #1592, Epic F/F2): additive companion to
514    /// [`scan_stream`](Self::scan_stream) that yields a `Vec` BATCH of
515    /// `(RowKey, ScanRow)` entries per channel item instead of one entry, so a
516    /// full-scan consumer is woken once per batch rather than once per row.
517    ///
518    /// Content and order are identical to [`scan_stream`](Self::scan_stream) —
519    /// flattening the batches reproduces the per-row stream exactly. Backpressure
520    /// is preserved (bounded channel). Delegates to
521    /// [`SSTableManager::scan_stream_batched`].
522    ///
523    /// [`SSTableManager::scan_stream_batched`]: sstable::SSTableManager::scan_stream_batched
524    pub async fn scan_stream_batched(
525        &self,
526        table_id: &TableId,
527        start_key: Option<&RowKey>,
528        end_key: Option<&RowKey>,
529        schema: Option<&crate::schema::TableSchema>,
530        buffer_size: usize,
531    ) -> Result<tokio::sync::mpsc::Receiver<Result<Vec<(RowKey, ScanRow)>>>> {
532        record_table_scan_call();
533        self.sstables
534            .scan_stream_batched(table_id, start_key, end_key, schema, buffer_size)
535            .await
536    }
537
538    /// Reports whether [`scan_stream`](Self::scan_stream) PRE-MATERIALIZES the
539    /// full reconciled result for this table before returning the channel, rather
540    /// than yielding rows lazily (issue #1577).
541    ///
542    /// A bounded LIMIT consumer uses this to decide its `QUERY_ROWS_SCANNED`
543    /// accounting: when it returns `true` the storage layer has already decoded the
544    /// whole table (no decode-stop is possible and per-received-row counting would
545    /// under-report), so the caller must charge the full decoded row count and take
546    /// a materializing path. Delegates to
547    /// [`SSTableManager::scan_stream_materializes`].
548    ///
549    /// [`SSTableManager::scan_stream_materializes`]: sstable::SSTableManager::scan_stream_materializes
550    pub async fn scan_stream_materializes(
551        &self,
552        table_id: &TableId,
553        schema: Option<&crate::schema::TableSchema>,
554    ) -> bool {
555        self.sstables
556            .scan_stream_materializes(table_id, schema)
557            .await
558    }
559
560    /// Flush MemTable to SSTable
561    ///
562    /// NOTE: Write functionality removed in Issue #175 (WAL/MemTable infrastructure deleted).
563    /// This method is feature-gated behind 'experimental' but currently unimplemented.
564    #[allow(dead_code)]
565    #[cfg(feature = "experimental")]
566    async fn flush_memtable(&self) -> Result<()> {
567        Err(crate::error::Error::UnsupportedFormat(
568            "Write operations (flush_memtable) removed in Issue #175 - WAL and MemTable infrastructure deleted".to_string()
569        ))
570    }
571
572    /// Force flush all pending writes
573    ///
574    /// NOTE: Write functionality removed in Issue #175 (WAL/MemTable infrastructure deleted).
575    /// This method is feature-gated behind 'experimental' but currently unimplemented.
576    #[cfg(feature = "experimental")]
577    pub async fn flush(&self) -> Result<()> {
578        Err(crate::error::Error::UnsupportedFormat(
579            "Write operations (flush) removed in Issue #175 - WAL and MemTable infrastructure deleted".to_string()
580        ))
581    }
582
583    /// Perform manual compaction
584    #[cfg(feature = "experimental")]
585    pub async fn compact(&self) -> Result<()> {
586        // TODO: Implement proper compaction logic
587        // This would need to identify candidates and call CompactionManager::run_compaction
588        Ok(())
589    }
590
591    /// The shared, bytes-bounded B1 decompressed-chunk cache owned by the
592    /// SSTable manager (issue #1567/#1568), or `None` when block caching is
593    /// disabled (`config.memory.block_cache.enabled == false`). Cloned (`Arc`) so
594    /// the memory-stats shell can report the live cache's real hit/miss/occupancy
595    /// numbers through `Database::stats().memory_stats`; when `None` the shell
596    /// reports a structural zero (the toggle genuinely disables caching).
597    pub(crate) fn chunk_cache(&self) -> Option<Arc<crate::storage::cache::DecompressedChunkCache>> {
598        self.sstables.stats_chunk_cache()
599    }
600
601    /// Snapshot of the process-global key→partition-offset cache (issue #2059).
602    /// Merged into `Database::stats().memory_stats` so the key cache's real
603    /// hits/misses/evictions/invalidations/occupancy/capacity are observable as one
604    /// consolidated envelope.
605    pub(crate) async fn key_cache_stats(&self) -> crate::storage::cache::GlobalKeyCacheSnapshot {
606        self.sstables.aggregate_key_cache_stats().await
607    }
608
609    /// Get storage statistics
610    ///
611    /// NOTE: Issue #176 removed compaction stats (compaction.rs deleted).
612    pub async fn stats(&self) -> Result<StorageStats> {
613        let sstable_stats = self.sstables.stats().await?;
614
615        Ok(StorageStats {
616            sstables: sstable_stats,
617        })
618    }
619
620    /// Batch write operations for better performance
621    ///
622    /// NOTE: Write functionality removed in Issue #175 (WAL/MemTable infrastructure deleted).
623    /// This method is feature-gated behind 'experimental' but currently unimplemented.
624    #[cfg(feature = "experimental")]
625    pub async fn batch_write(&mut self, _operations: Vec<BatchOperation>) -> Result<()> {
626        Err(crate::error::Error::UnsupportedFormat(
627            "Write operations (batch_write) removed in Issue #175 - WAL and MemTable infrastructure deleted".to_string()
628        ))
629    }
630
631    /// Explicit batch flush
632    ///
633    /// NOTE: Write functionality removed in Issue #175 (WAL/MemTable infrastructure deleted).
634    /// This method is feature-gated behind 'experimental' but currently unimplemented.
635    #[cfg(feature = "experimental")]
636    pub async fn flush_batch(&mut self) -> Result<()> {
637        Err(crate::error::Error::UnsupportedFormat(
638            "Write operations (flush_batch) removed in Issue #175 - WAL and MemTable infrastructure deleted".to_string()
639        ))
640    }
641
642    /// Get batch writer statistics
643    ///
644    /// NOTE: Write functionality removed in Issue #175 (WAL/MemTable infrastructure deleted).
645    /// This method is feature-gated behind 'experimental' but currently unimplemented.
646    #[cfg(feature = "experimental")]
647    pub fn batch_stats(&self) -> Option<()> {
648        None
649    }
650
651    /// Shutdown the storage engine
652    ///
653    /// NOTE: Issue #176 removed compaction shutdown (compaction.rs deleted).
654    /// Issue #175 removed flush operations (WAL/MemTable deleted).
655    pub async fn shutdown(&self) -> Result<()> {
656        // Nothing to shutdown - read-only storage layer
657        Ok(())
658    }
659
660    /// Set the schema registry for schema-aware operations
661    ///
662    /// This method propagates the schema registry to the SSTable manager,
663    /// which will apply it to all SSTable readers for schema-aware parsing.
664    #[cfg(feature = "state_machine")]
665    pub async fn set_schema_registry(
666        &self,
667        registry: Arc<RwLock<crate::schema::SchemaRegistry>>,
668    ) -> Result<()> {
669        // Store in our field
670        {
671            let mut schema_reg = self.schema_registry.write().await;
672            *schema_reg = Some(registry.clone());
673        }
674
675        // Propagate to SSTable manager
676        self.sstables.set_schema_registry(registry).await
677    }
678}
679
680/// Batch operation types
681#[cfg(feature = "experimental")]
682#[derive(Debug, Clone)]
683pub enum BatchOperation {
684    /// Put operation
685    Put {
686        table_id: TableId,
687        key: RowKey,
688        value: Value,
689    },
690    /// Delete operation
691    Delete { table_id: TableId, key: RowKey },
692    /// Merge operation
693    Merge {
694        table_id: TableId,
695        key: RowKey,
696        value: Value,
697    },
698}
699
700/// Storage engine statistics
701///
702/// NOTE: Issue #176 removed compaction statistics (compaction.rs deleted).
703#[derive(Debug, Clone)]
704pub struct StorageStats {
705    /// SSTable statistics
706    pub sstables: sstable::SSTableStats,
707}
708
709#[cfg(test)]
710mod tests {
711    use super::*;
712    use tempfile::TempDir;
713
714    #[tokio::test]
715    async fn test_storage_engine_creation() {
716        let temp_dir = TempDir::new().unwrap();
717        let config = Config::test_config();
718        let platform = Arc::new(Platform::new(&config).await.unwrap());
719
720        let storage = StorageEngine::open(
721            temp_dir.path(),
722            &config,
723            platform,
724            #[cfg(feature = "state_machine")]
725            None,
726        )
727        .await
728        .unwrap();
729        let stats = storage.stats().await.unwrap();
730
731        assert_eq!(stats.sstables.sstable_count, 0);
732        storage.shutdown().await.unwrap();
733    }
734
735    #[tokio::test]
736    async fn test_storage_engine_with_discovered_sstables() {
737        let temp_dir = TempDir::new().unwrap();
738        let config = Config::test_config();
739        let platform = Arc::new(Platform::new(&config).await.unwrap());
740
741        // Create an empty list of discovered SSTables for this test
742        let discovered_paths = Vec::new();
743
744        let storage = StorageEngine::open_with_sstables(
745            temp_dir.path(),
746            discovered_paths,
747            &config,
748            platform,
749            #[cfg(feature = "state_machine")]
750            None,
751        )
752        .await
753        .unwrap();
754
755        let stats = storage.stats().await.unwrap();
756
757        // Should have 0 SSTables since we provided an empty list
758        assert_eq!(stats.sstables.sstable_count, 0);
759        storage.shutdown().await.unwrap();
760    }
761
762    // NOTE: `test_batch_operations` and `test_batch_operations_fallback` were
763    // removed in Issue #1880. They drove `StorageEngine::batch_write`, whose
764    // WAL/MemTable implementation was deleted in Issue #175 — the method is now an
765    // always-erroring stub, so both tests could only ever panic under
766    // `--all-features`. There is no batch-write behavior left to assert.
767}