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}