pub mod bloom;
pub mod bti;
pub mod bulletproof_reader;
pub mod chunk_decompressor;
pub mod chunk_reader;
pub mod compression;
pub mod compression_info;
pub mod directory;
pub mod directory_integration_tests;
pub mod format_detector;
#[cfg(feature = "write-support")]
mod generation_merge; pub mod header_spec;
pub mod index;
pub mod index_reader;
pub mod key_digest;
pub mod performance_benchmarks;
pub mod promoted_index_reader;
pub mod read_work_counters;
pub mod reader;
pub mod summary_reader;
pub mod version_gate;
pub mod work_counters;
#[cfg(feature = "zstd")]
pub mod zstd_frame;
pub use reader::SSTableReader;
pub mod refresh;
pub use refresh::RefreshReport;
mod reverse_scan; pub mod row_cell_state_machine;
mod scan_merge;
pub(crate) mod snapshot_path;
pub mod statistics_reader;
pub mod stream_merge_probe; #[cfg(feature = "tombstones")]
pub mod tombstone_merger;
pub mod validation;
pub mod verify; pub use verify::{
verify_sstable, verify_sstable_generation, VerifyErrorClass, VerifyFinding, VerifyMode,
VerifyReport,
};
#[cfg(feature = "write-support")]
pub mod writer;
#[cfg(test)]
mod issue_1591_scan_lock_test;
#[cfg(all(test, feature = "scan-offload-probe"))]
mod issue_1594_fanout_deadlock_test;
#[cfg(test)]
mod issue_653_version_gates_plumbing_test;
#[cfg(test)]
mod key_digest_integration_test;
#[cfg(test)]
mod key_digest_test;
#[cfg(all(test, feature = "experimental"))]
mod oa_format_compliance_test;
#[cfg(all(test, feature = "state_machine"))]
mod row_cell_state_machine_test;
#[cfg(test)]
mod s3_verification_test;
#[cfg(test)]
mod s4_verification_test;
use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use tokio::sync::{Mutex, RwLock};
#[cfg(feature = "tombstones")]
use self::tombstone_merger::{EntryMetadata, GenerationValue, TombstoneMerger};
use crate::platform::Platform;
use crate::types::CellWriteMetadata;
#[cfg(not(feature = "tombstones"))] use crate::util::cassandra_murmur3::cmp_partition_keys_by_token;
use crate::{types::TableId, Config, Result, RowKey, ScanRow};
pub(crate) const MAX_SSTABLE_SCAN_DEPTH: usize = 3;
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PartitionKeyShape {
pub partition_key_count: usize,
pub clustering_key_count: usize,
pub non_key_column_names: std::collections::HashSet<String>,
pub single_pk_component: Option<PartitionKeyComponent>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PartitionKeyComponent {
pub name: String,
pub cql_type: String,
}
fn partition_key_shape_from_headers<'a>(
headers: impl IntoIterator<Item = &'a [crate::parser::header::ColumnInfo]>,
) -> Option<PartitionKeyShape> {
struct AgreedKeyShape {
partition_key_count: usize,
clustering_key_count: usize,
single_pk_component: Option<PartitionKeyComponent>,
}
let mut agreed: Option<AgreedKeyShape> = None;
let mut non_key_column_names = std::collections::HashSet::new();
for columns in headers {
if columns.is_empty() {
return None;
}
let mut partition_key_count = 0usize;
let mut clustering_key_count = 0usize;
let mut first_pk_component: Option<PartitionKeyComponent> = None;
for col in columns {
match (col.is_primary_key, col.is_clustering) {
(true, false) => {
if first_pk_component.is_none() {
first_pk_component = Some(PartitionKeyComponent {
name: col.name.clone(),
cql_type: col.column_type.clone(),
});
}
partition_key_count += 1;
}
(true, true) => clustering_key_count += 1,
(false, _) => {
non_key_column_names.insert(col.name.clone());
}
}
}
if partition_key_count == 0 {
return None;
}
let single_pk_component = (partition_key_count == 1)
.then_some(first_pk_component)
.flatten();
match &agreed {
None => {
agreed = Some(AgreedKeyShape {
partition_key_count,
clustering_key_count,
single_pk_component,
});
}
Some(prev) => {
if prev.partition_key_count != partition_key_count
|| prev.clustering_key_count != clustering_key_count
|| prev.single_pk_component != single_pk_component
{
return None;
}
}
}
}
agreed.map(|shape| PartitionKeyShape {
partition_key_count: shape.partition_key_count,
clustering_key_count: shape.clustering_key_count,
non_key_column_names,
single_pk_component: shape.single_pk_component,
})
}
#[derive(Debug, Clone, PartialEq, Eq, Hash, serde::Serialize, serde::Deserialize)]
pub struct SSTableId(pub String);
impl Default for SSTableId {
fn default() -> Self {
Self::new()
}
}
impl SSTableId {
pub fn new() -> Self {
let timestamp = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_micros();
Self(format!("sstable-{}-big-Data.db", timestamp))
}
pub fn from_filename(filename: &str) -> Self {
Self(filename.to_string())
}
pub fn filename(&self) -> &str {
&self.0
}
}
pub(crate) fn extract_table_name(sstable_path: &Path) -> Option<String> {
snapshot_path::extract_table_name(sstable_path)
}
pub fn extract_keyspace_and_table_name(sstable_path: &Path) -> Option<(String, String)> {
let table_name = snapshot_path::extract_table_name(sstable_path)?;
let keyspace = snapshot_path::extract_keyspace(sstable_path)?;
Some((keyspace, table_name))
}
#[inline]
fn is_apple_double_sidecar(filename: &str) -> bool {
filename.starts_with("._")
}
#[cfg(test)]
pub(crate) mod scan_gate {
use std::sync::Arc;
use tokio::sync::Notify;
#[derive(Debug, Default)]
pub(crate) struct Gate {
pub reached: Notify,
pub release: Notify,
}
pub(crate) async fn wait_if_armed(gate: &Option<Arc<Gate>>) {
if let Some(gate) = gate {
gate.reached.notify_one();
gate.release.notified().await;
}
}
}
#[derive(Debug)]
pub struct SSTableManager {
base_path: PathBuf,
readers: Arc<RwLock<HashMap<SSTableId, Arc<reader::SSTableReader>>>>,
pub(crate) table_readers: Arc<RwLock<HashMap<String, Vec<Arc<reader::SSTableReader>>>>>,
platform: Arc<Platform>,
config: Config,
discovery_source: refresh::DiscoverySource,
refresh_lock: Arc<Mutex<()>>,
#[cfg(feature = "state_machine")]
schema_registry: Arc<RwLock<Option<Arc<RwLock<crate::schema::SchemaRegistry>>>>>,
pub(crate) chunk_cache: Arc<crate::storage::cache::DecompressedChunkCache>,
#[cfg(test)]
scan_gate: std::sync::Mutex<Option<Arc<scan_gate::Gate>>>,
}
fn build_chunk_cache(config: &Config) -> Arc<crate::storage::cache::DecompressedChunkCache> {
use crate::storage::cache::DecompressedChunkCache;
if config.memory.block_cache.enabled {
Arc::new(DecompressedChunkCache::with_budget_bytes(
config.memory.block_cache.max_size as usize,
))
} else {
Arc::new(DecompressedChunkCache::disabled())
}
}
pub(crate) fn build_key_offset_cache(
config: &Config,
) -> Arc<crate::storage::cache::GlobalKeyOffsetCache> {
use crate::storage::cache::GlobalKeyOffsetCache;
if config.memory.block_cache.enabled {
GlobalKeyOffsetCache::global()
} else {
Arc::new(GlobalKeyOffsetCache::disabled())
}
}
impl SSTableManager {
pub(crate) fn stats_chunk_cache(
&self,
) -> Option<Arc<crate::storage::cache::DecompressedChunkCache>> {
if self.config.memory.block_cache.enabled {
Some(Arc::clone(&self.chunk_cache))
} else {
None
}
}
pub(crate) async fn aggregate_key_cache_stats(
&self,
) -> crate::storage::cache::GlobalKeyCacheSnapshot {
if self.config.memory.block_cache.enabled {
crate::storage::cache::GlobalKeyOffsetCache::global().snapshot()
} else {
crate::storage::cache::GlobalKeyCacheSnapshot::default()
}
}
pub async fn new(
path: &Path,
config: &Config,
platform: Arc<Platform>,
#[cfg(feature = "state_machine")] schema_registry: Option<
Arc<RwLock<crate::schema::SchemaRegistry>>,
>,
) -> Result<Self> {
let base_path = path.to_path_buf();
let readers = Arc::new(RwLock::new(HashMap::new()));
let table_readers = Arc::new(RwLock::new(HashMap::new()));
let manager = Self {
base_path,
readers,
table_readers,
platform,
config: config.clone(),
discovery_source: refresh::DiscoverySource::BasePath,
refresh_lock: Arc::new(Mutex::new(())),
#[cfg(feature = "state_machine")]
schema_registry: Arc::new(RwLock::new(schema_registry)),
chunk_cache: build_chunk_cache(config),
#[cfg(test)]
scan_gate: std::sync::Mutex::new(None),
};
manager.load_existing_sstables().await?;
Ok(manager)
}
pub async fn new_from_discovered_paths(
storage_path: &Path,
table_dirs: Vec<PathBuf>,
config: &Config,
platform: Arc<Platform>,
#[cfg(feature = "state_machine")] schema_registry: Option<
Arc<RwLock<crate::schema::SchemaRegistry>>,
>,
) -> Result<Self> {
let base_path = storage_path.to_path_buf();
let readers = Arc::new(RwLock::new(HashMap::new()));
let table_readers = Arc::new(RwLock::new(HashMap::new()));
let manager = Self {
base_path,
readers,
table_readers,
platform: platform.clone(),
config: config.clone(),
discovery_source: refresh::DiscoverySource::TableDirs(table_dirs.clone()),
refresh_lock: Arc::new(Mutex::new(())),
#[cfg(feature = "state_machine")]
schema_registry: Arc::new(RwLock::new(schema_registry)),
chunk_cache: build_chunk_cache(config),
#[cfg(test)]
scan_gate: std::sync::Mutex::new(None),
};
manager.load_from_table_directories(table_dirs).await?;
Ok(manager)
}
async fn load_from_table_directories(&self, table_dirs: Vec<PathBuf>) -> Result<()> {
let mut readers = self.readers.write().await;
let mut table_readers = self.table_readers.write().await;
tracing::debug!(
"SSTableManager::load_from_table_directories: processing {} directories",
table_dirs.len()
);
for table_dir in table_dirs {
if !self.platform.fs().exists(&table_dir).await? {
tracing::warn!("Table directory does not exist: {:?}", table_dir);
continue;
}
tracing::debug!("SSTableManager scanning directory: {:?}", table_dir);
let mut dir_entries = match self.platform.fs().read_dir(&table_dir).await {
Ok(entries) => entries,
Err(e) => {
tracing::warn!("Cannot read table directory {:?}: {}", table_dir, e);
continue;
}
};
let mut files_found = 0;
while let Some(entry) = dir_entries.next_entry().await? {
let path = entry.path();
if let Some(filename) = path.file_name().and_then(|n| n.to_str()) {
if filename.ends_with("-Data.db") && !is_apple_double_sidecar(filename) {
files_found += 1;
tracing::debug!("SSTableManager found SSTable file: {:?}", path);
let sstable_id = SSTableId::from_filename(filename);
match self.open_reader_with_schema(&path).await {
Ok(reader_arc) => {
tracing::debug!(
"SSTableManager successfully loaded SSTable: {}",
sstable_id.0
);
readers.insert(sstable_id, reader_arc.clone());
if let Some(key) = refresh::table_dir_table_key(&path) {
tracing::debug!(
"SSTableManager mapping table '{}' to SSTable '{}'",
key,
path.display()
);
table_readers
.entry(key)
.or_insert_with(Vec::new)
.push(reader_arc);
} else {
tracing::warn!(
"SSTableManager could not extract table name from path: {}",
path.display()
);
}
}
Err(e) => {
tracing::warn!("Could not load SSTable file {:?}: {}", path, e);
}
}
}
}
}
tracing::debug!(
"SSTableManager directory scan complete: found {} Data.db files in {:?}",
files_found,
table_dir
);
}
tracing::debug!("SSTableManager total SSTables loaded: {}", readers.len());
tracing::debug!(
"SSTableManager tables discovered: {:?}",
table_readers.keys().collect::<Vec<_>>()
);
Ok(())
}
async fn load_existing_sstables(&self) -> Result<()> {
if !self.platform.fs().exists(&self.base_path).await? {
return Ok(()); }
let data_files: Vec<PathBuf> =
Self::find_data_files(&self.platform, &self.base_path, MAX_SSTABLE_SCAN_DEPTH).await?;
if data_files.is_empty() {
return Ok(());
}
let mut readers = self.readers.write().await;
let mut table_readers = self.table_readers.write().await;
let base_dir_name = self
.base_path
.file_name()
.and_then(|n| n.to_str())
.unwrap_or("")
.to_string();
for path in data_files {
let filename = match path.file_name().and_then(|n| n.to_str()) {
Some(f) => f.to_string(),
None => continue,
};
let sstable_id = SSTableId::from_filename(&filename);
match self.open_reader_with_schema(&path).await {
Ok(reader_arc) => {
readers.insert(sstable_id, reader_arc.clone());
let table_key = refresh::base_path_table_key(
&path,
&base_dir_name,
&reader_arc.header().table_name,
);
if let Some(key) = table_key {
tracing::debug!(
"SSTableManager mapping table '{}' to SSTable '{}'",
key,
path.display()
);
table_readers
.entry(key)
.or_insert_with(Vec::new)
.push(reader_arc);
} else {
tracing::warn!(
"SSTableManager could not determine table name for: {}",
path.display()
);
}
}
Err(_) => {
tracing::warn!("Could not load SSTable file: {:?}", path);
}
}
}
Ok(())
}
fn find_data_files<'a>(
platform: &'a Platform,
dir: &'a Path,
max_depth: usize,
) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<Vec<PathBuf>>> + Send + 'a>>
{
let dir = dir.to_path_buf();
Box::pin(async move {
let mut results = Vec::new();
let mut dir_entries = match platform.fs().read_dir(&dir).await {
Ok(entries) => entries,
Err(_) => return Ok(results),
};
while let Some(entry) = dir_entries.next_entry().await? {
let path = entry.path();
if let Some(filename) = path.file_name().and_then(|n| n.to_str()) {
if filename.ends_with("-Data.db") && !is_apple_double_sidecar(filename) {
results.push(path);
} else if max_depth > 0 {
if entry
.file_type()
.await
.map(|ft| ft.is_dir())
.unwrap_or(false)
{
let sub_results =
Self::find_data_files(platform, &path, max_depth - 1).await?;
results.extend(sub_results);
}
}
}
}
Ok(results)
})
}
#[cfg(feature = "experimental")]
pub async fn create_from_memtable(
&self,
_data: Vec<(TableId, RowKey, ScanRow)>,
) -> Result<SSTableId> {
Err(crate::error::Error::unsupported_format(
"SSTable writing removed in Issue #176 - writer.rs deleted",
))
}
#[cfg(not(feature = "experimental"))]
pub async fn create_from_memtable(
&self,
_data: Vec<(TableId, RowKey, ScanRow)>,
) -> Result<SSTableId> {
Err(crate::error::Error::unsupported_format(
"SSTable writing requires experimental feature",
))
}
#[cfg(feature = "tombstones")]
pub async fn get(&self, table_id: &TableId, key: &RowKey) -> Result<Option<ScanRow>> {
let (reader_list, fully_qualified_match) = self.resolve_reader_snapshot(table_id).await;
if reader_list.is_empty() {
return Ok(None);
}
let mut all_values = Vec::new();
for reader in &reader_list {
if let Some(value) = reader
.get_with_resolution(table_id, key, fully_qualified_match)
.await?
{
let generation = reader.generation;
let write_time = reader.extract_write_time_from_entry(key, &value);
let gen_value = GenerationValue {
value,
metadata: EntryMetadata {
write_time,
generation,
ttl: None, },
};
all_values.push(gen_value);
}
}
let merger = TombstoneMerger::new();
merger.merge_generations(all_values)
}
#[cfg(not(feature = "tombstones"))]
pub async fn get(&self, table_id: &TableId, key: &RowKey) -> Result<Option<ScanRow>> {
let (reader_list, fully_qualified_match) = self.resolve_reader_snapshot(table_id).await;
for reader in &reader_list {
if let Some(value) = reader
.get_with_resolution(table_id, key, fully_qualified_match)
.await?
{
return Ok(Some(value));
}
}
Ok(None)
}
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)>> {
tracing::debug!("SSTableManager::scan - Scanning table_id='{}'", table_id);
let (reader_list, _fully_qualified_match) = self.resolve_reader_snapshot(table_id).await;
if reader_list.is_empty() {
tracing::debug!(
"SSTableManager::scan - No readers found for table '{}'",
table_id
);
return Ok(Vec::new());
}
tracing::debug!(
"SSTableManager::scan - Found {} readers for table '{}'",
reader_list.len(),
table_id
);
#[cfg(feature = "write-support")]
if reader_list.len() > 1 {
if let Some(schema) = schema {
match generation_merge::merge_generations_for_read(
&reader_list,
schema,
start_key,
end_key,
limit,
None,
)
.await
{
Ok(merged) => {
tracing::debug!(
"SSTableManager::scan - cross-generation merge produced {} rows",
merged.len()
);
return Ok(merged);
}
Err(e) => {
tracing::warn!(
"SSTableManager::scan - cross-generation merge failed for '{}' ({}); \
falling back to per-reader concatenation",
table_id,
e
);
}
}
}
}
let mut per_reader = Vec::with_capacity(reader_list.len());
for reader in &reader_list {
#[cfg(test)]
{
let gate = self.scan_gate.lock().ok().and_then(|g| g.clone());
scan_gate::wait_if_armed(&gate).await;
}
let results = reader
.scan(table_id, start_key, end_key, None, schema)
.await?;
per_reader.push(results);
}
let all_results = scan_merge::kway_merge_token_order(per_reader, limit);
tracing::debug!(
"SSTableManager::scan - Returning {} final results (token-ordered k-way merge)",
all_results.len()
);
Ok(all_results)
}
#[cfg(not(feature = "tombstones"))]
pub async fn scan_partition(
&self,
table_id: &TableId,
partition_key: &[u8],
schema: Option<&crate::schema::TableSchema>,
) -> Result<(Vec<(RowKey, ScanRow)>, bool)> {
let (rows, _clustering_engaged) = self
.scan_partition_clustering(table_id, partition_key, None, schema)
.await?;
Ok((rows, true))
}
#[cfg(not(feature = "tombstones"))]
fn prune_candidates(
readers: &[Arc<reader::SSTableReader>],
partition_key: &[u8],
) -> (
Vec<Arc<reader::SSTableReader>>,
Vec<Arc<reader::SSTableReader>>,
) {
use crate::storage::sstable::bti::encode_partition_key_for_bti_trie;
let encoded = readers
.iter()
.any(|r| r.is_bti())
.then(|| encode_partition_key_for_bti_trie(partition_key));
readers.iter().cloned().partition(|r| {
match (r.is_bti(), &encoded) {
(true, Some(enc)) => r.might_contain_partition_encoded(partition_key, enc),
_ => r.might_contain_partition(partition_key),
}
})
}
#[cfg(not(feature = "tombstones"))]
async fn verify_pruned_candidates(
pruned: &[Arc<reader::SSTableReader>],
table_id: &TableId,
partition_key: &[u8],
) {
if !reader::presence_verification::enabled() {
return;
}
for r in pruned {
if let Err(e) = r
.verify_presence_oracle_negative(table_id, partition_key)
.await
{
tracing::error!(
error = %e,
sstable_format = r.sstable_format_label(),
"opt-in presence-oracle false-negative verification scan FAILED for a \
prune_candidates-excluded SSTable — the read itself is unaffected \
(fail-open), but this soundness check could not run for this SSTable and \
needs investigation"
);
crate::observability::record_error(&e, "reader");
}
}
}
#[cfg(not(feature = "tombstones"))]
pub async fn scan_partition_with_cell_metadata(
&self,
table_id: &TableId,
partition_key: &[u8],
schema: Option<&crate::schema::TableSchema>,
) -> Result<(
Vec<(RowKey, ScanRow, HashMap<String, CellWriteMetadata>)>,
bool,
)> {
let (reader_list, _fully_qualified_match) = self.resolve_reader_snapshot(table_id).await;
if reader_list.is_empty() {
return Ok((Vec::new(), true));
}
let (candidates, pruned) = Self::prune_candidates(&reader_list, partition_key);
Self::verify_pruned_candidates(&pruned, table_id, partition_key).await;
tracing::debug!(
"SSTableManager::scan_partition_with_cell_metadata - {}/{} SSTables admit partition \
key (len={}) for '{}'",
candidates.len(),
reader_list.len(),
partition_key.len(),
table_id
);
if candidates.is_empty() {
return Ok((Vec::new(), true));
}
let matches_key = |entry: &(RowKey, ScanRow, HashMap<String, CellWriteMetadata>)| {
entry.0.as_bytes() == partition_key
};
#[cfg(feature = "write-support")]
if candidates.len() > 1 {
if let Some(schema) = schema {
let target = RowKey::new(partition_key.to_vec());
match generation_merge::merge_generations_for_read_with_metadata(
&candidates,
schema,
None,
None,
None,
Some(&target),
)
.await
{
Ok(merged) => {
work_counters::add_sstables_scanned(candidates.len() as u64);
work_counters::add_partitions_parsed(merged.len() as u64);
return Ok((merged, true));
}
Err(e) => {
tracing::warn!(
"SSTableManager::scan_partition_with_cell_metadata - cross-generation \
metadata merge failed for '{}' ({}); falling back to per-reader \
concatenation",
table_id,
e
);
}
}
}
}
let mut all_results = Vec::new();
for reader in &candidates {
work_counters::add_sstables_scanned(1);
let mut results = reader
.scan_with_cell_metadata(table_id, None, None, None, schema)
.await?;
results.retain(matches_key);
all_results.append(&mut results);
}
if candidates.len() > 1 {
all_results.sort_by(|a, b| cmp_partition_keys_by_token(a.0.as_bytes(), b.0.as_bytes()));
}
work_counters::add_partitions_parsed(all_results.len() as u64);
Ok((all_results, true))
}
#[cfg(feature = "tombstones")]
pub async fn scan_partition_with_cell_metadata(
&self,
table_id: &TableId,
partition_key: &[u8],
schema: Option<&crate::schema::TableSchema>,
) -> Result<(
Vec<(
RowKey,
ScanRow,
HashMap<String, crate::types::CellWriteMetadata>,
)>,
bool,
)> {
let mut rows = self
.scan_with_cell_metadata(table_id, None, None, None, schema)
.await?;
rows.retain(|entry| entry.0.as_bytes() == partition_key);
Ok((rows, false))
}
#[cfg(not(feature = "tombstones"))]
pub async fn scan_partition_clustering(
&self,
table_id: &TableId,
partition_key: &[u8],
clustering: Option<&reader::ClusteringSlice>,
schema: Option<&crate::schema::TableSchema>,
) -> Result<(Vec<(RowKey, ScanRow)>, bool)> {
let (reader_list, fully_qualified_match) = self.resolve_reader_snapshot(table_id).await;
if reader_list.is_empty() {
return Ok((Vec::new(), false));
}
let (candidates, pruned) = Self::prune_candidates(&reader_list, partition_key);
Self::verify_pruned_candidates(&pruned, table_id, partition_key).await;
tracing::debug!(
"SSTableManager::scan_partition - {}/{} SSTables admit partition key (len={}) for '{}'",
candidates.len(),
reader_list.len(),
partition_key.len(),
table_id
);
if candidates.is_empty() {
return Ok((Vec::new(), false));
}
let matches_key = |entry: &(RowKey, ScanRow)| entry.0.as_bytes() == partition_key;
#[cfg(all(feature = "write-support", not(feature = "tombstones")))]
if candidates.len() > 1 {
if let Some(schema) = schema {
let target = RowKey::new(partition_key.to_vec());
match generation_merge::seek_merge_generations_for_read(
&candidates,
schema,
&target,
)
.await
{
Ok(merged) => {
work_counters::add_sstables_scanned(candidates.len() as u64);
work_counters::add_partitions_parsed(merged.len() as u64);
return Ok((merged, false));
}
Err(e) => {
tracing::warn!(
"SSTableManager::scan_partition - cross-generation merge failed for \
'{}' ({}); falling back to per-reader concatenation",
table_id,
e
);
}
}
}
}
let mut all_results = Vec::new();
let mut clustering_engaged = false;
for reader in &candidates {
work_counters::add_sstables_scanned(1);
let mut results = if candidates.len() == 1 {
match reader
.scan_single_partition_clustering(
table_id,
partition_key,
clustering,
fully_qualified_match,
schema,
)
.await
{
Ok(Some((rows, engaged))) => {
clustering_engaged = engaged;
rows
}
Ok(None) => {
let mut r = reader.scan(table_id, None, None, None, schema).await?;
r.retain(matches_key);
r
}
Err(e) => return Err(e),
}
} else {
let mut r = reader.scan(table_id, None, None, None, schema).await?;
r.retain(matches_key);
r
};
all_results.append(&mut results);
}
if candidates.len() > 1 {
all_results.sort_by(|a, b| cmp_partition_keys_by_token(a.0.as_bytes(), b.0.as_bytes()));
}
work_counters::add_partitions_parsed(all_results.len() as u64);
Ok((all_results, clustering_engaged))
}
#[cfg(feature = "tombstones")]
pub async fn scan_partition(
&self,
table_id: &TableId,
partition_key: &[u8],
schema: Option<&crate::schema::TableSchema>,
) -> Result<(Vec<(RowKey, ScanRow)>, bool)> {
let mut rows = self.scan(table_id, None, None, None, schema).await?;
rows.retain(|entry| entry.0.as_bytes() == partition_key);
Ok((rows, false))
}
pub(in crate::storage::sstable) fn resolve_reader_list<'a>(
table_readers: &'a HashMap<String, Vec<Arc<reader::SSTableReader>>>,
table_name: &str,
) -> Option<&'a Vec<Arc<reader::SSTableReader>>> {
if let Some(list) = table_readers.get(table_name) {
return Some(list);
}
let unqualified = table_name
.rfind('.')
.map_or(table_name, |dot| &table_name[dot + 1..]);
table_readers.get(unqualified)
}
pub(in crate::storage::sstable) fn fully_qualified_match(
table_readers: &HashMap<String, Vec<Arc<reader::SSTableReader>>>,
table_name: &str,
) -> bool {
!table_name.contains('.') || table_readers.contains_key(table_name)
}
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, HashMap<String, CellWriteMetadata>)>> {
let (reader_list, _fully_qualified_match) = self.resolve_reader_snapshot(table_id).await;
if reader_list.is_empty() {
return Ok(Vec::new());
}
#[cfg(feature = "write-support")]
if reader_list.len() > 1 {
if let Some(schema) = schema {
match generation_merge::merge_generations_for_read_with_metadata(
&reader_list,
schema,
start_key,
end_key,
limit,
None,
)
.await
{
Ok(merged) => return Ok(merged),
Err(e) => {
tracing::warn!(
"SSTableManager::scan_with_cell_metadata - cross-generation merge \
failed for '{}' ({}); falling back to per-reader concatenation",
table_id,
e
);
}
}
}
}
let mut per_reader = Vec::with_capacity(reader_list.len());
for reader in &reader_list {
let results = reader
.scan_with_cell_metadata(table_id, start_key, end_key, None, schema)
.await?;
per_reader.push(
results
.into_iter()
.map(|(k, v, m)| (k, (v, m)))
.collect::<Vec<_>>(),
);
}
let all_results = scan_merge::kway_merge_token_order(per_reader, limit)
.into_iter()
.map(|(k, (v, m))| (k, v, m))
.collect();
Ok(all_results)
}
async fn resolve_reader_snapshot(
&self,
table_id: &TableId,
) -> (Vec<Arc<reader::SSTableReader>>, bool) {
let table_readers = self.table_readers.read().await;
let table_name = table_id.name();
let fully_qualified_match = Self::fully_qualified_match(&table_readers, table_name);
let readers = Self::resolve_reader_list(&table_readers, table_name)
.cloned()
.unwrap_or_default();
(readers, fully_qualified_match)
}
pub async fn partition_key_shape(&self, table_id: &TableId) -> Option<PartitionKeyShape> {
let (readers, _) = self.resolve_reader_snapshot(table_id).await;
partition_key_shape_from_headers(readers.iter().map(|r| r.header().columns.as_slice()))
}
#[cfg(all(test, feature = "write-support", feature = "state_machine"))] fn arm_scan_gate(&self) -> Arc<scan_gate::Gate> {
let gate = Arc::new(scan_gate::Gate::default());
if let Ok(mut slot) = self.scan_gate.lock() {
*slot = Some(Arc::clone(&gate));
}
gate
}
#[cfg(not(feature = "tombstones"))]
async fn resolve_table_readers(&self, table_id: &TableId) -> Vec<Arc<reader::SSTableReader>> {
let table_readers = self.table_readers.read().await;
let table_name = table_id.name();
let list = if table_readers.contains_key(table_name) {
table_readers.get(table_name)
} else {
let unqualified_name = if let Some(dot_pos) = table_name.rfind('.') {
&table_name[dot_pos + 1..]
} else {
table_name
};
table_readers.get(unqualified_name)
};
list.cloned().unwrap_or_default()
}
#[cfg(feature = "tombstones")]
pub async fn scan_stream_materializes(
&self,
_table_id: &TableId,
_schema: Option<&crate::schema::TableSchema>,
) -> bool {
true
}
#[cfg(not(feature = "tombstones"))]
pub async fn scan_stream_materializes(
&self,
table_id: &TableId,
_schema: Option<&crate::schema::TableSchema>,
) -> bool {
let readers = self.resolve_table_readers(table_id).await;
if readers.iter().any(|r| r.is_bti()) {
return true;
}
false
}
#[cfg(not(feature = "tombstones"))]
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)>>> {
let readers = self.resolve_table_readers(table_id).await;
#[cfg(feature = "write-support")]
if readers.len() > 1 {
if let Some(schema) = schema {
match generation_merge::stream_generations_for_read(
&readers,
schema,
start_key,
end_key,
buffer_size,
)
.await
{
Ok(rx) => {
tracing::debug!(
"SSTableManager::scan_stream - cross-generation merge streaming \
(O(window), not materialized)"
);
return Ok(rx);
}
Err(e) => {
tracing::warn!(
"SSTableManager::scan_stream - cross-generation merge failed for '{}' ({}); \
falling back to lazy per-reader streaming merge",
table_id,
e
);
}
}
}
}
let (out_tx, out_rx) = tokio::sync::mpsc::channel(buffer_size.max(1));
let table_id = table_id.clone();
let start_key = start_key.cloned();
let end_key = end_key.cloned();
let schema = schema.cloned();
tokio::spawn(async move {
let _admission =
crate::storage::sstable::reader::scan_stream_windowed::scan_admission::admit()
.await;
let mut streams: Vec<tokio::sync::mpsc::Receiver<Result<(RowKey, ScanRow)>>> = readers
.into_iter()
.map(|reader| {
reader.scan_stream_admitted(
table_id.clone(),
start_key.clone(),
end_key.clone(),
schema.clone(),
buffer_size,
crate::storage::sstable::reader::scan_stream_windowed::scan_admission::ScanAdmission::Exempt,
)
})
.collect();
let token_of = |key: &RowKey| {
crate::util::cassandra_murmur3::cassandra_murmur3_token(key.as_bytes())
};
let mut heads: Vec<Option<(i64, RowKey, ScanRow)>> = Vec::with_capacity(streams.len());
for stream in streams.iter_mut() {
match stream.recv().await {
Some(Ok((key, row))) => heads.push(Some((token_of(&key), key, row))),
Some(Err(e)) => {
let _ = out_tx.send(Err(e)).await;
return;
}
None => heads.push(None),
}
}
loop {
let mut min_idx: Option<usize> = None;
for (i, head) in heads.iter().enumerate() {
if let Some((ref token, ref key, _)) = head {
match min_idx {
None => min_idx = Some(i),
Some(m) => {
if let Some((ref min_token, ref min_key, _)) = heads[m] {
if (token, key) < (min_token, min_key) {
min_idx = Some(i);
}
}
}
}
}
}
let idx = match min_idx {
Some(idx) => idx,
None => break, };
let entry = match heads[idx].take() {
Some((_, key, row)) => (key, row),
None => break, };
match streams[idx].recv().await {
Some(Ok((key, row))) => heads[idx] = Some((token_of(&key), key, row)),
Some(Err(e)) => {
let _ = out_tx.send(Err(e)).await;
return;
}
None => {} }
if out_tx.send(Ok(entry)).await.is_err() {
return; }
}
});
Ok(out_rx)
}
#[cfg(not(feature = "tombstones"))]
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)>>>> {
let readers = self.resolve_table_readers(table_id).await;
if readers.len() == 1 {
if let Some(reader) = readers.into_iter().next() {
return Ok(reader.scan_stream_batched(
table_id.clone(),
start_key.cloned(),
end_key.cloned(),
schema.cloned(),
buffer_size,
));
}
}
let per_row = self
.scan_stream(table_id, start_key, end_key, schema, buffer_size)
.await?;
Ok(Self::rechunk_into_batches(per_row, buffer_size))
}
fn rechunk_into_batches(
mut per_row: tokio::sync::mpsc::Receiver<Result<(RowKey, ScanRow)>>,
buffer_size: usize,
) -> tokio::sync::mpsc::Receiver<Result<Vec<(RowKey, ScanRow)>>> {
use reader::scan_stream_windowed::BATCH_EMIT_ROWS;
let cap = buffer_size.div_ceil(BATCH_EMIT_ROWS).max(1);
let (tx, rx) = tokio::sync::mpsc::channel(cap);
tokio::spawn(async move {
let mut batch: Vec<(RowKey, ScanRow)> = Vec::with_capacity(BATCH_EMIT_ROWS);
while let Some(item) = per_row.recv().await {
match item {
Ok(entry) => {
batch.push(entry);
if batch.len() >= BATCH_EMIT_ROWS {
if tx.send(Ok(std::mem::take(&mut batch))).await.is_err() {
return; }
batch.reserve(BATCH_EMIT_ROWS);
}
}
Err(e) => {
if !batch.is_empty() {
let _ = tx.send(Ok(std::mem::take(&mut batch))).await;
}
let _ = tx.send(Err(e)).await;
return;
}
}
}
if !batch.is_empty() {
let _ = tx.send(Ok(batch)).await;
}
});
rx
}
#[cfg(feature = "tombstones")]
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)>>> {
let results = self
.scan(table_id, start_key, end_key, None, schema)
.await?;
let (tx, rx) = tokio::sync::mpsc::channel(buffer_size.max(1));
tokio::spawn(async move {
for entry in results {
if tx.send(Ok(entry)).await.is_err() {
break; }
}
});
Ok(rx)
}
#[cfg(feature = "tombstones")]
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)>>>> {
let per_row = self
.scan_stream(table_id, start_key, end_key, schema, buffer_size)
.await?;
Ok(Self::rechunk_into_batches(per_row, buffer_size))
}
pub async fn list_sstables(&self) -> Vec<SSTableId> {
let readers = self.readers.read().await;
readers.keys().cloned().collect()
}
pub async fn remove_sstable(&self, sstable_id: &SSTableId) -> Result<()> {
{
let mut readers = self.readers.write().await;
if let Some(removed) = readers.remove(sstable_id) {
removed.invalidate_key_cache_entries();
}
}
let file_path = self.base_path.join(sstable_id.filename());
if self.platform.fs().exists(&file_path).await? {
self.platform.fs().remove_file(&file_path).await?;
}
Ok(())
}
pub async fn stats(&self) -> Result<SSTableStats> {
let readers = self.readers.read().await;
let mut total_size = 0u64;
let mut total_entries = 0u64;
let mut total_tables = 0u64;
let sstable_count = readers.len();
for reader in readers.values() {
let reader_stats = reader.stats().await?;
total_size += reader_stats.file_size;
total_entries += reader_stats.entry_count;
total_tables += reader_stats.table_count;
}
Ok(SSTableStats {
sstable_count,
total_size,
total_entries,
total_tables,
average_size: if sstable_count > 0 {
total_size / sstable_count as u64
} else {
0
},
})
}
#[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());
}
Ok(())
}
#[cfg(feature = "experimental")]
pub async fn merge_sstables(
&self,
_source_ids: Vec<SSTableId>,
_target_id: SSTableId,
) -> Result<()> {
Err(crate::error::Error::unsupported_format(
"SSTable merging removed in Issue #176 - writer.rs deleted",
))
}
#[cfg(not(feature = "experimental"))]
pub async fn merge_sstables(
&self,
_source_ids: Vec<SSTableId>,
_target_id: SSTableId,
) -> Result<()> {
Err(crate::error::Error::unsupported_format(
"SSTable merging requires experimental feature",
))
}
}
#[derive(Debug, Clone)]
pub struct SSTableStats {
pub sstable_count: usize,
pub total_size: u64,
pub total_entries: u64,
pub total_tables: u64,
pub average_size: u64,
}
#[cfg(test)]
mod tests {
use super::*;
use crate::platform::Platform;
use tempfile::TempDir;
#[tokio::test]
async fn test_sstable_manager_creation() {
let temp_dir = TempDir::new().unwrap();
let config = Config::default();
let platform = Arc::new(Platform::new(&config).await.unwrap());
let manager = SSTableManager::new(
temp_dir.path(),
&config,
platform,
#[cfg(feature = "state_machine")]
None,
)
.await
.unwrap();
let stats = manager.stats().await.unwrap();
assert_eq!(stats.sstable_count, 0);
assert_eq!(stats.total_size, 0);
}
#[tokio::test]
async fn test_sstable_manager_from_discovered_paths_empty() {
let temp_dir = TempDir::new().unwrap();
let config = Config::default();
let platform = Arc::new(Platform::new(&config).await.unwrap());
let discovered_paths = Vec::new();
let manager = SSTableManager::new_from_discovered_paths(
temp_dir.path(),
discovered_paths,
&config,
platform,
#[cfg(feature = "state_machine")]
None,
)
.await
.unwrap();
let stats = manager.stats().await.unwrap();
assert_eq!(stats.sstable_count, 0);
assert_eq!(stats.total_size, 0);
}
#[tokio::test]
async fn test_sstable_manager_from_discovered_paths_with_directories() {
use std::fs;
let temp_dir = TempDir::new().unwrap();
let config = Config::default();
let platform = Arc::new(Platform::new(&config).await.unwrap());
let keyspace_dir = temp_dir.path().join("test_ks");
fs::create_dir(&keyspace_dir).unwrap();
let table1_dir = keyspace_dir.join("users-abc123");
fs::create_dir(&table1_dir).unwrap();
fs::write(table1_dir.join("na-1-big-Data.db"), b"mock_data").unwrap();
let table2_dir = keyspace_dir.join("posts-def456");
fs::create_dir(&table2_dir).unwrap();
fs::write(table2_dir.join("na-2-big-Data.db"), b"mock_data").unwrap();
fs::write(table2_dir.join("na-3-big-Data.db"), b"mock_data").unwrap();
let table_dirs = vec![table1_dir.clone(), table2_dir.clone()];
let manager = SSTableManager::new_from_discovered_paths(
temp_dir.path(),
table_dirs,
&config,
platform,
#[cfg(feature = "state_machine")]
None,
)
.await
.unwrap();
let stats = manager.stats().await.unwrap();
let _ = stats.sstable_count; }
#[tokio::test]
#[ignore = "M3+ feature; gated for M1"]
async fn test_sstable_id_generation() {
let id1 = SSTableId::new();
let id2 = SSTableId::new();
assert_ne!(id1.filename(), id2.filename());
assert!(id1.filename().starts_with("sstable_"));
assert!(id1.filename().ends_with(".sst"));
}
#[tokio::test]
async fn test_find_data_files_excludes_apple_double_sidecar() {
use std::fs;
let temp_dir = tempfile::TempDir::new().unwrap();
let config = Config::default();
let platform = Arc::new(Platform::new(&config).await.unwrap());
let real_file = temp_dir.path().join("nb-1-big-Data.db");
let sidecar = temp_dir.path().join("._nb-1-big-Data.db");
fs::write(&real_file, b"\x00").unwrap();
fs::write(&sidecar, b"\x00\x00").unwrap();
let results = SSTableManager::find_data_files(&platform, temp_dir.path(), 0)
.await
.unwrap();
assert_eq!(
results.len(),
1,
"expected exactly 1 result but got {}: {:?}",
results.len(),
results
);
assert_eq!(results[0], real_file);
assert!(
!results.contains(&sidecar),
"AppleDouble sidecar must not appear in results"
);
}
#[test]
fn test_is_apple_double_sidecar() {
assert!(is_apple_double_sidecar("._nb-1-big-Data.db"));
assert!(is_apple_double_sidecar("._anything"));
assert!(is_apple_double_sidecar("._"));
assert!(!is_apple_double_sidecar("nb-1-big-Data.db"));
assert!(!is_apple_double_sidecar("na-2-big-Data.db"));
assert!(!is_apple_double_sidecar(""));
}
#[test]
fn test_extract_table_name() {
use std::path::PathBuf;
let path =
PathBuf::from("test-data/datasets/sstables/test_basic/simple_table-6aa08200a25111f0a3fef1a551383fb9/nb-1-big-Data.db");
assert_eq!(extract_table_name(&path), Some("simple_table".to_string()));
let path = PathBuf::from(
"test-data/datasets/sstables/test_basic/my-test-table-6aa08200a25111f0a3fef1a551383fb9/nb-1-big-Data.db",
);
assert_eq!(extract_table_name(&path), Some("my-test-table".to_string()));
let path = PathBuf::from(
"test-data/datasets/sstables/test_basic/multi_partition_table-6ac52100a25111f0a3fef1a551383fb9/nb-1-big-Data.db",
);
assert_eq!(
extract_table_name(&path),
Some("multi_partition_table".to_string())
);
let path = PathBuf::from(
"test-data/datasets/sstables/test_basic/compression_test_table-6ad6ad30a25111f0a3fef1a551383fb9/nb-1-big-Data.db",
);
assert_eq!(
extract_table_name(&path),
Some("compression_test_table".to_string())
);
let path =
PathBuf::from("test-data/datasets/sstables/test_basic/simple_table/nb-1-big-Data.db");
assert_eq!(extract_table_name(&path), Some("simple_table".to_string()));
let path = PathBuf::from("nb-1-big-Data.db");
assert_eq!(extract_table_name(&path), None);
}
#[test]
fn test_fully_qualified_match_signal_both_builds() {
let mut table_readers: HashMap<String, Vec<Arc<reader::SSTableReader>>> = HashMap::new();
table_readers.insert("ks_a.users".to_string(), Vec::new());
assert!(
SSTableManager::fully_qualified_match(&table_readers, "ks_a.users"),
"exact FQ map hit must signal an exact match (relax)"
);
assert!(
!SSTableManager::fully_qualified_match(&table_readers, "ks_b.users"),
"FQ query missing its exact key must signal a fallback (strict)"
);
assert!(
SSTableManager::fully_qualified_match(&table_readers, "users"),
"unqualified query must signal an exact match (relax)"
);
}
async fn open_dataset_reader(
keyspace: &str,
table: &str,
) -> Option<Arc<reader::SSTableReader>> {
let datasets_root = std::env::var("CQLITE_DATASETS_ROOT").ok()?;
let keyspace_dir = PathBuf::from(datasets_root).join("sstables").join(keyspace);
let table_prefix = format!("{}-", table);
for entry in std::fs::read_dir(&keyspace_dir).ok()?.flatten() {
let path = entry.path();
let file_name = path.file_name()?.to_str()?.to_string();
if file_name.starts_with(&table_prefix) {
let data_file = std::fs::read_dir(&path)
.ok()?
.flatten()
.find(|e| {
e.file_name()
.to_str()
.map(|s| s.ends_with("-Data.db"))
.unwrap_or(false)
})?
.path();
let config = Config::default();
let platform = Arc::new(Platform::new(&config).await.ok()?);
return reader::SSTableReader::open(&data_file, &config, platform)
.await
.ok()
.map(Arc::new);
}
}
None
}
#[tokio::test]
async fn test_tombstones_get_resolves_only_target_keyspace_readers() {
let Some(reader_a) = open_dataset_reader("test_basic", "simple_table").await else {
eprintln!("skipping: CQLITE_DATASETS_ROOT / test_basic.simple_table absent");
return;
};
let Some(reader_b) = open_dataset_reader("test_basic", "counters").await else {
eprintln!("skipping: CQLITE_DATASETS_ROOT / test_basic.counters absent");
return;
};
assert!(
!Arc::ptr_eq(&reader_a, &reader_b),
"the two stand-in keyspace readers must be distinct Arcs"
);
let mut table_readers: HashMap<String, Vec<Arc<reader::SSTableReader>>> = HashMap::new();
table_readers.insert("ks_a.users".to_string(), vec![Arc::clone(&reader_a)]);
table_readers.insert("ks_b.users".to_string(), vec![Arc::clone(&reader_b)]);
let resolved = SSTableManager::resolve_reader_list(&table_readers, "ks_a.users")
.expect("ks_a.users resolves");
assert_eq!(
resolved.len(),
1,
"merge set must be exactly ks_a's readers"
);
assert!(
Arc::ptr_eq(&resolved[0], &reader_a),
"resolved merge set must contain the ks_a reader"
);
assert!(
!resolved.iter().any(|r| Arc::ptr_eq(r, &reader_b)),
"Issue #1321: a ks_a.users query must NOT merge the same-named ks_b.users reader"
);
assert!(
SSTableManager::fully_qualified_match(&table_readers, "ks_a.users"),
"exact FQ key present → relaxed guard, applied only to the resolved (ks_a) set"
);
}
fn dataset_table_dir(keyspace: &str, table: &str) -> Option<PathBuf> {
let datasets_root = std::env::var("CQLITE_DATASETS_ROOT").ok()?;
let keyspace_dir = PathBuf::from(datasets_root).join("sstables").join(keyspace);
let table_prefix = format!("{}-", table);
for entry in std::fs::read_dir(&keyspace_dir).ok()?.flatten() {
let path = entry.path();
let file_name = path.file_name()?.to_str()?.to_string();
if path.is_dir() && file_name.starts_with(&table_prefix) && dir_has_data_db(&path) {
return Some(path);
}
}
None
}
fn dir_has_data_db(dir: &Path) -> bool {
let Ok(entries) = std::fs::read_dir(dir) else {
return false;
};
entries
.flatten()
.any(|f| f.file_name().to_string_lossy().ends_with("-Data.db"))
}
#[tokio::test]
async fn test_global_key_cache_capacity_is_reader_count_independent() {
let Some(dir_a) = dataset_table_dir("test_basic", "simple_table") else {
eprintln!("skipping: CQLITE_DATASETS_ROOT / test_basic.simple_table absent");
return;
};
let Some(dir_b) = dataset_table_dir("test_basic", "counters") else {
eprintln!("skipping: CQLITE_DATASETS_ROOT / test_basic.counters absent");
return;
};
let config = Config::default();
let platform = Arc::new(Platform::new(&config).await.unwrap());
let storage_path = dir_a.parent().map(PathBuf::from).unwrap_or(dir_a.clone());
let manager = SSTableManager::new_from_discovered_paths(
&storage_path,
vec![dir_a, dir_b],
&config,
platform,
#[cfg(feature = "state_machine")]
None,
)
.await
.unwrap();
let distinct = {
let readers = manager.readers.read().await;
let table_readers = manager.table_readers.read().await;
assert!(!readers.is_empty(), "fixture present but no readers by-id");
let mut seen: std::collections::HashSet<*const reader::SSTableReader> =
std::collections::HashSet::new();
for r in readers.values().chain(table_readers.values().flatten()) {
seen.insert(Arc::as_ptr(r));
}
seen.len()
};
let agg = manager.aggregate_key_cache_stats().await;
assert_eq!(
agg.capacity_bytes,
crate::storage::cache::DEFAULT_GLOBAL_KEY_CACHE_BYTES,
"global cache reports its single fixed budget regardless of reader count \
({distinct} readers open)"
);
let mut disabled_config = Config::default();
disabled_config.memory.block_cache.enabled = false;
let dplatform = Arc::new(Platform::new(&disabled_config).await.unwrap());
let dmanager = SSTableManager::new(
&storage_path,
&disabled_config,
dplatform,
#[cfg(feature = "state_machine")]
None,
)
.await
.unwrap();
let dagg = dmanager.aggregate_key_cache_stats().await;
assert_eq!(
dagg.capacity_bytes, 0,
"disabled manager reports zero capacity"
);
assert_eq!(dagg.hits, 0);
assert_eq!(dagg.misses, 0);
}
#[tokio::test]
async fn rechunk_flushes_pending_rows_before_midstream_error() {
use reader::scan_stream_windowed::BATCH_EMIT_ROWS;
let pending = 3usize;
assert!(pending < BATCH_EMIT_ROWS);
let (in_tx, in_rx) = tokio::sync::mpsc::channel::<Result<(RowKey, ScanRow)>>(16);
for i in 0..pending {
let key = RowKey::new(vec![i as u8]);
let row = ScanRow::RawRow(vec![i as u8]);
in_tx.send(Ok((key, row))).await.unwrap();
}
in_tx
.send(Err(crate::Error::Corruption("boom".to_string())))
.await
.unwrap();
drop(in_tx);
let mut out = SSTableManager::rechunk_into_batches(in_rx, 64);
let first = out
.recv()
.await
.expect("expected a pending batch before the error");
let batch = first.expect("pending batch must be Ok, not the error");
assert_eq!(
batch.len(),
pending,
"all confirmed rows must survive the error"
);
assert!(
batch.len() <= BATCH_EMIT_ROWS,
"batch must respect the BATCH_EMIT_ROWS bound"
);
for (i, (key, _row)) in batch.iter().enumerate() {
assert_eq!(key.as_bytes(), [i as u8], "rows must arrive in order");
}
let second = out.recv().await.expect("expected the terminal error item");
assert!(second.is_err(), "second item must be the forwarded error");
assert!(
out.recv().await.is_none(),
"no items after the terminal error"
);
}
fn col(
name: &str,
cql_type: &str,
is_primary_key: bool,
is_clustering: bool,
) -> crate::parser::header::ColumnInfo {
crate::parser::header::ColumnInfo {
name: name.to_string(),
column_type: cql_type.to_string(),
is_primary_key,
key_position: None,
is_static: false,
is_clustering,
clustering_reversed: false,
}
}
#[test]
fn partition_key_shape_unions_later_generation_regular_column() {
let gen1 = vec![
col("partition_key", "int", true, false),
col("v1", "text", false, false),
];
let gen2 = vec![
col("partition_key", "int", true, false),
col("v1", "text", false, false),
col("v2", "text", false, false),
];
let shape = partition_key_shape_from_headers([gen1.as_slice(), gen2.as_slice()])
.expect("consistent headers must resolve a shape");
assert_eq!(shape.partition_key_count, 1);
assert_eq!(shape.clustering_key_count, 0);
assert!(
shape.non_key_column_names.contains("v2"),
"later-gen regular column must be in the unioned non-key set (roborev 3786)",
);
assert!(shape.non_key_column_names.contains("v1"));
assert_eq!(
shape
.single_pk_component
.as_ref()
.map(|c| c.cql_type.as_str()),
Some("int"),
);
let gen1_only =
partition_key_shape_from_headers([gen1.as_slice()]).expect("single header resolves");
assert!(
!gen1_only.non_key_column_names.contains("v2"),
"gen-1 alone MISSES v2 — this is exactly the multi-gen bug the union fixes",
);
}
#[test]
fn partition_key_shape_declines_inconsistent_readers() {
let single = vec![
col("pk", "int", true, false),
col("v", "text", false, false),
];
let composite = vec![
col("pk_a", "int", true, false),
col("pk_b", "int", true, false),
col("v", "text", false, false),
];
assert_eq!(
partition_key_shape_from_headers([single.as_slice(), composite.as_slice()]),
None,
"a pk-count disagreement across readers must decline (→ full-scan)",
);
let clustered = vec![
col("pk", "int", true, false),
col("ck", "int", true, true),
col("v", "text", false, false),
];
assert_eq!(
partition_key_shape_from_headers([single.as_slice(), clustered.as_slice()]),
None,
"a clustering-count disagreement across readers must decline",
);
let single_bigint = vec![
col("pk", "bigint", true, false),
col("v", "text", false, false),
];
assert_eq!(
partition_key_shape_from_headers([single.as_slice(), single_bigint.as_slice()]),
None,
"a single-pk type disagreement across readers must decline",
);
}
#[test]
fn partition_key_shape_none_without_usable_header() {
assert_eq!(partition_key_shape_from_headers(std::iter::empty()), None);
let empty: Vec<crate::parser::header::ColumnInfo> = vec![];
assert_eq!(partition_key_shape_from_headers([empty.as_slice()]), None);
let no_pk = vec![col("v", "text", false, false)];
assert_eq!(partition_key_shape_from_headers([no_pk.as_slice()]), None);
}
#[test]
fn partition_key_shape_declines_when_any_reader_lacks_usable_header() {
let usable = vec![
col("partition_key", "int", true, false),
col("v1", "text", false, false),
];
let headerless: Vec<crate::parser::header::ColumnInfo> = vec![];
assert_eq!(
partition_key_shape_from_headers([headerless.as_slice(), usable.as_slice()]),
None,
"a leading headerless reader must decline the whole shape (→ full-scan)",
);
assert_eq!(
partition_key_shape_from_headers([usable.as_slice(), headerless.as_slice()]),
None,
"a trailing headerless reader must decline the whole shape (→ full-scan)",
);
let usable2 = vec![
col("partition_key", "int", true, false),
col("v1", "text", false, false),
col("v2", "text", false, false),
];
assert_eq!(
partition_key_shape_from_headers([
usable.as_slice(),
headerless.as_slice(),
usable2.as_slice(),
]),
None,
"any headerless reader among usable ones declines the shape",
);
assert!(
partition_key_shape_from_headers([usable.as_slice(), usable2.as_slice()]).is_some(),
"usable readers alone must still resolve (the fix only declines on a gap)",
);
}
}