pub mod cache;
pub mod scan_cancel;
pub mod sstable;
pub mod partition_key_codec;
#[cfg(feature = "write-support")]
pub mod serialization;
#[cfg(feature = "write-support")]
pub mod write_engine;
#[cfg(feature = "cli-helpers")]
pub mod repl_data_api;
pub mod schema_discovery;
pub mod sstable_data_manager;
use std::path::{Path, PathBuf};
use std::sync::Arc;
#[cfg(feature = "state_machine")]
use tokio::sync::RwLock;
use crate::platform::Platform;
use crate::{
types::{CellWriteMetadata, TableId},
Config, Result, RowKey, ScanRow,
};
#[cfg(feature = "experimental")]
use crate::types::Value;
#[cfg(any(test, feature = "work-counters"))]
thread_local! {
static TABLE_SCAN_CALLS: std::cell::Cell<u64> = const { std::cell::Cell::new(0) };
}
#[inline(always)]
pub(crate) fn record_table_scan_call() {
#[cfg(any(test, feature = "work-counters"))]
TABLE_SCAN_CALLS.with(|c| c.set(c.get().saturating_add(1)));
}
#[cfg(any(test, feature = "work-counters"))]
pub fn table_scan_call_count() -> u64 {
TABLE_SCAN_CALLS.with(|c| c.get())
}
#[cfg(any(test, feature = "work-counters"))]
pub fn reset_table_scan_calls() {
TABLE_SCAN_CALLS.with(|c| c.set(0));
}
#[derive(Debug)]
pub struct StorageEngine {
sstables: Arc<sstable::SSTableManager>,
#[allow(dead_code)]
_platform: Arc<Platform>,
#[allow(dead_code)]
config: Config,
#[cfg(feature = "state_machine")]
schema_registry: Arc<RwLock<Option<Arc<RwLock<crate::schema::SchemaRegistry>>>>>,
}
impl StorageEngine {
#[tracing::instrument(
name = "storage.engine.open",
level = "debug",
skip_all,
fields(sstables = tracing::field::Empty, bytes = tracing::field::Empty)
)]
pub async fn open(
path: &Path,
config: &Config,
platform: Arc<Platform>,
#[cfg(feature = "state_machine")] schema_registry: Option<
Arc<RwLock<crate::schema::SchemaRegistry>>,
>,
) -> Result<Self> {
crate::observability::record_result("reader", platform.fs().create_dir_all(path).await)?;
let sstables = Arc::new(crate::observability::record_result(
"reader",
sstable::SSTableManager::new(
path,
config,
platform.clone(),
#[cfg(feature = "state_machine")]
schema_registry.clone(),
)
.await,
)?);
Self::record_discovery_metrics(&sstables).await;
Ok(Self {
sstables,
_platform: platform,
config: config.clone(),
#[cfg(feature = "state_machine")]
schema_registry: Arc::new(RwLock::new(schema_registry)),
})
}
async fn record_discovery_metrics(sstables: &sstable::SSTableManager) {
use crate::observability::{self as obs, catalog};
match sstables.stats().await {
Ok(stats) => {
tracing::Span::current().record("sstables", stats.sstable_count as u64);
tracing::Span::current().record("bytes", stats.total_size);
obs::add_counter(
catalog::STORAGE_OPEN_SSTABLES,
stats.sstable_count as u64,
&[],
);
obs::add_counter(catalog::STORAGE_OPEN_BYTES, stats.total_size, &[]);
obs::add_counter(catalog::STORAGE_OPEN_TABLES, stats.total_tables, &[]);
}
Err(e) => {
tracing::debug!("storage.engine.open: discovery metrics unavailable: {}", e);
}
}
}
#[tracing::instrument(
name = "storage.engine.open",
level = "debug",
skip_all,
fields(sstables = tracing::field::Empty, bytes = tracing::field::Empty)
)]
pub async fn open_with_sstables(
path: &Path,
discovered_table_dirs: Vec<PathBuf>,
config: &Config,
platform: Arc<Platform>,
#[cfg(feature = "state_machine")] schema_registry: Option<
Arc<RwLock<crate::schema::SchemaRegistry>>,
>,
) -> Result<Self> {
crate::observability::record_result("reader", platform.fs().create_dir_all(path).await)?;
let sstables = Arc::new(crate::observability::record_result(
"reader",
sstable::SSTableManager::new_from_discovered_paths(
path,
discovered_table_dirs,
config,
platform.clone(),
#[cfg(feature = "state_machine")]
schema_registry.clone(),
)
.await,
)?);
Self::record_discovery_metrics(&sstables).await;
Ok(Self {
sstables,
_platform: platform,
config: config.clone(),
#[cfg(feature = "state_machine")]
schema_registry: Arc::new(RwLock::new(schema_registry)),
})
}
#[cfg(feature = "experimental")]
pub async fn put(&self, _table_id: &TableId, _key: RowKey, _value: Value) -> Result<()> {
Err(crate::error::Error::UnsupportedFormat(
"Write operations (put) removed in Issue #175 - WAL and MemTable infrastructure deleted".to_string()
))
}
pub async fn get(&self, table_id: &TableId, key: &RowKey) -> Result<Option<ScanRow>> {
self.sstables.get(table_id, key).await
}
pub async fn refresh(&self) -> Result<sstable::RefreshReport> {
self.sstables.refresh_tables().await
}
#[cfg(feature = "experimental")]
pub async fn delete(&self, _table_id: &TableId, _key: RowKey) -> Result<()> {
Err(crate::error::Error::UnsupportedFormat(
"Write operations (delete) removed in Issue #175 - WAL and MemTable infrastructure deleted".to_string()
))
}
pub async fn scan(
&self,
table_id: &TableId,
start_key: Option<&RowKey>,
end_key: Option<&RowKey>,
limit: Option<usize>,
schema: Option<&crate::schema::TableSchema>,
) -> Result<Vec<(RowKey, ScanRow)>> {
record_table_scan_call();
self.sstables
.scan(table_id, start_key, end_key, limit, schema)
.await
}
pub async fn scan_partition(
&self,
table_id: &TableId,
partition_key: &[u8],
schema: Option<&crate::schema::TableSchema>,
) -> Result<(Vec<(RowKey, ScanRow)>, bool)> {
self.sstables
.scan_partition(table_id, partition_key, schema)
.await
}
pub async fn partition_key_shape(
&self,
table_id: &TableId,
) -> Option<sstable::PartitionKeyShape> {
self.sstables.partition_key_shape(table_id).await
}
#[cfg(not(feature = "tombstones"))]
pub async fn scan_partition_clustering(
&self,
table_id: &TableId,
partition_key: &[u8],
clustering: Option<&crate::storage::sstable::reader::ClusteringSlice>,
schema: Option<&crate::schema::TableSchema>,
) -> Result<(Vec<(RowKey, ScanRow)>, bool)> {
self.sstables
.scan_partition_clustering(table_id, partition_key, clustering, schema)
.await
}
#[cfg(not(feature = "tombstones"))]
pub async fn scan_partition_clustering_reverse(
&self,
table_id: &TableId,
partition_key: &[u8],
schema: Option<&crate::schema::TableSchema>,
) -> Result<Option<Vec<(RowKey, ScanRow)>>> {
self.sstables
.scan_partition_clustering_reverse(table_id, partition_key, schema)
.await
}
pub async fn scan_partition_with_cell_metadata(
&self,
table_id: &TableId,
partition_key: &[u8],
schema: Option<&crate::schema::TableSchema>,
) -> Result<(
Vec<(
RowKey,
ScanRow,
std::collections::HashMap<String, CellWriteMetadata>,
)>,
bool,
)> {
self.sstables
.scan_partition_with_cell_metadata(table_id, partition_key, schema)
.await
}
pub async fn scan_with_cell_metadata(
&self,
table_id: &TableId,
start_key: Option<&RowKey>,
end_key: Option<&RowKey>,
limit: Option<usize>,
schema: Option<&crate::schema::TableSchema>,
) -> Result<
Vec<(
RowKey,
ScanRow,
std::collections::HashMap<String, CellWriteMetadata>,
)>,
> {
self.sstables
.scan_with_cell_metadata(table_id, start_key, end_key, limit, schema)
.await
}
pub async fn scan_stream(
&self,
table_id: &TableId,
start_key: Option<&RowKey>,
end_key: Option<&RowKey>,
schema: Option<&crate::schema::TableSchema>,
buffer_size: usize,
) -> Result<tokio::sync::mpsc::Receiver<Result<(RowKey, ScanRow)>>> {
record_table_scan_call();
self.sstables
.scan_stream(table_id, start_key, end_key, schema, buffer_size)
.await
}
pub async fn scan_stream_batched(
&self,
table_id: &TableId,
start_key: Option<&RowKey>,
end_key: Option<&RowKey>,
schema: Option<&crate::schema::TableSchema>,
buffer_size: usize,
) -> Result<tokio::sync::mpsc::Receiver<Result<Vec<(RowKey, ScanRow)>>>> {
record_table_scan_call();
self.sstables
.scan_stream_batched(table_id, start_key, end_key, schema, buffer_size)
.await
}
pub async fn scan_stream_materializes(
&self,
table_id: &TableId,
schema: Option<&crate::schema::TableSchema>,
) -> bool {
self.sstables
.scan_stream_materializes(table_id, schema)
.await
}
#[allow(dead_code)]
#[cfg(feature = "experimental")]
async fn flush_memtable(&self) -> Result<()> {
Err(crate::error::Error::UnsupportedFormat(
"Write operations (flush_memtable) removed in Issue #175 - WAL and MemTable infrastructure deleted".to_string()
))
}
#[cfg(feature = "experimental")]
pub async fn flush(&self) -> Result<()> {
Err(crate::error::Error::UnsupportedFormat(
"Write operations (flush) removed in Issue #175 - WAL and MemTable infrastructure deleted".to_string()
))
}
#[cfg(feature = "experimental")]
pub async fn compact(&self) -> Result<()> {
Ok(())
}
pub(crate) fn chunk_cache(&self) -> Option<Arc<crate::storage::cache::DecompressedChunkCache>> {
self.sstables.stats_chunk_cache()
}
pub(crate) async fn key_cache_stats(&self) -> crate::storage::cache::GlobalKeyCacheSnapshot {
self.sstables.aggregate_key_cache_stats().await
}
pub async fn stats(&self) -> Result<StorageStats> {
let sstable_stats = self.sstables.stats().await?;
Ok(StorageStats {
sstables: sstable_stats,
})
}
#[cfg(feature = "experimental")]
pub async fn batch_write(&mut self, _operations: Vec<BatchOperation>) -> Result<()> {
Err(crate::error::Error::UnsupportedFormat(
"Write operations (batch_write) removed in Issue #175 - WAL and MemTable infrastructure deleted".to_string()
))
}
#[cfg(feature = "experimental")]
pub async fn flush_batch(&mut self) -> Result<()> {
Err(crate::error::Error::UnsupportedFormat(
"Write operations (flush_batch) removed in Issue #175 - WAL and MemTable infrastructure deleted".to_string()
))
}
#[cfg(feature = "experimental")]
pub fn batch_stats(&self) -> Option<()> {
None
}
pub async fn shutdown(&self) -> Result<()> {
Ok(())
}
#[cfg(feature = "state_machine")]
pub async fn set_schema_registry(
&self,
registry: Arc<RwLock<crate::schema::SchemaRegistry>>,
) -> Result<()> {
{
let mut schema_reg = self.schema_registry.write().await;
*schema_reg = Some(registry.clone());
}
self.sstables.set_schema_registry(registry).await
}
}
#[cfg(feature = "experimental")]
#[derive(Debug, Clone)]
pub enum BatchOperation {
Put {
table_id: TableId,
key: RowKey,
value: Value,
},
Delete { table_id: TableId, key: RowKey },
Merge {
table_id: TableId,
key: RowKey,
value: Value,
},
}
#[derive(Debug, Clone)]
pub struct StorageStats {
pub sstables: sstable::SSTableStats,
}
#[cfg(test)]
mod tests {
use super::*;
use tempfile::TempDir;
#[tokio::test]
async fn test_storage_engine_creation() {
let temp_dir = TempDir::new().unwrap();
let config = Config::test_config();
let platform = Arc::new(Platform::new(&config).await.unwrap());
let storage = StorageEngine::open(
temp_dir.path(),
&config,
platform,
#[cfg(feature = "state_machine")]
None,
)
.await
.unwrap();
let stats = storage.stats().await.unwrap();
assert_eq!(stats.sstables.sstable_count, 0);
storage.shutdown().await.unwrap();
}
#[tokio::test]
async fn test_storage_engine_with_discovered_sstables() {
let temp_dir = TempDir::new().unwrap();
let config = Config::test_config();
let platform = Arc::new(Platform::new(&config).await.unwrap());
let discovered_paths = Vec::new();
let storage = StorageEngine::open_with_sstables(
temp_dir.path(),
discovered_paths,
&config,
platform,
#[cfg(feature = "state_machine")]
None,
)
.await
.unwrap();
let stats = storage.stats().await.unwrap();
assert_eq!(stats.sstables.sstable_count, 0);
storage.shutdown().await.unwrap();
}
}