mod block_io;
mod bti_lookup_memo;
mod chunk_source;
pub mod compaction_row;
mod component_loading;
mod compression;
pub(crate) mod crc;
mod data_access;
#[cfg(feature = "delta-scan")]
pub mod delta_scan;
mod header;
mod header_helpers;
mod integrity;
mod key_digest;
pub(crate) mod parsing; mod partition_lookup;
pub mod presence_verification; mod partition_locator;
mod partition_successor;
mod read_at;
#[cfg(test)]
mod read_at_point_tests;
mod prefetch_window;
#[cfg(feature = "state_machine")]
mod registry_schema;
#[cfg(not(feature = "scan-offload-probe"))]
pub(crate) mod scan_stream_windowed;
#[cfg(feature = "scan-offload-probe")]
pub mod scan_stream_windowed;
mod source;
mod summary_point; #[cfg(test)]
mod tests;
mod types;
#[cfg(not(feature = "scan-offload-probe"))]
pub(crate) mod window_cursor;
#[cfg(feature = "scan-offload-probe")]
pub mod window_cursor;
pub(crate) mod value_borrow;
pub use types::{
BlockMeta, IntegrityCheckResult, IntegrityStatus, SSTableReader, SSTableReaderConfig,
SSTableReaderHealthMetrics, SSTableReaderStats,
};
pub use data_access::ClusteringSlice;
pub use data_access::ScanTokenBound;
#[cfg(not(feature = "tombstones"))]
pub use data_access::SinglePartitionCompaction;
pub use compaction_row::{
CompactionRow, CompactionRowData, ComplexColumn, ComplexElement, SimpleCell,
};
#[doc(hidden)]
pub use parsing::PublicV5CompressedLegacyParser as V5CompressedLegacyParser;
#[doc(hidden)]
pub use compression::extract_sstable_base_name;
use header::{
calculate_actual_header_size, extract_generation_from_path, parse_header_with_version_detection,
};
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicI64, Ordering};
use std::sync::Arc;
use tokio::fs::File;
use tokio::io::{AsyncReadExt, AsyncSeekExt};
use tokio::sync::Mutex;
use source::{BlockSource, ScanSource};
use crate::{
config::{DiskAccessMode, PrefetchMode},
parser::{header::CassandraVersion, SSTableHeader, SSTableParser},
platform::Platform,
schema::TableSchema,
storage::sstable::{
compression::CompressionReader,
compression_info::CompressionInfo,
version_gate::{BigVersionGates, VersionGates},
},
Config, Error, Result, RowKey, ScanRow, Value,
};
use tracing::debug;
#[cfg(feature = "tombstones")]
use super::tombstone_merger::TombstoneMerger;
pub(crate) fn index_db_sibling(data_path: &Path) -> Option<PathBuf> {
let parent = data_path.parent()?;
let name = data_path.file_name().and_then(|n| n.to_str())?;
let base = name.strip_suffix("-Data.db")?;
Some(parent.join(format!("{base}-Index.db")))
}
fn mmap_enabled_via_env() -> bool {
std::env::var("CQLITE_USE_MMAP")
.ok()
.as_deref()
.map(parse_truthy_env)
.unwrap_or(false)
}
fn parse_truthy_env(value: &str) -> bool {
matches!(
value.trim().to_ascii_lowercase().as_str(),
"1" | "true" | "yes" | "on"
)
}
fn parse_disk_access_mode(value: &str) -> Option<DiskAccessMode> {
match value.trim().to_ascii_lowercase().as_str() {
"auto" => Some(DiskAccessMode::Auto),
"buffered" | "buffer" => Some(DiskAccessMode::Buffered),
"mmap" | "mapped" => Some(DiskAccessMode::Mmap),
"direct" | "directio" | "direct_io" | "o_direct" => Some(DiskAccessMode::Direct),
_ => None,
}
}
fn parse_prefetch_mode(value: &str) -> Option<PrefetchMode> {
match value.trim().to_ascii_lowercase().as_str() {
"off" | "none" | "no" => Some(PrefetchMode::Off),
"sequential" | "seq" => Some(PrefetchMode::Sequential),
"willneed" | "will_need" | "will-need" => Some(PrefetchMode::WillNeed),
"auto" => Some(PrefetchMode::Auto),
_ => None,
}
}
fn disk_access_mode_via_env() -> Option<DiskAccessMode> {
std::env::var("CQLITE_DISK_ACCESS_MODE")
.ok()
.as_deref()
.and_then(parse_disk_access_mode)
}
fn prefetch_mode_via_env() -> Option<PrefetchMode> {
std::env::var("CQLITE_PREFETCH")
.ok()
.as_deref()
.and_then(parse_prefetch_mode)
}
fn system_memory_bytes() -> Option<u64> {
#[cfg(unix)]
{
let pages = unsafe { libc::sysconf(libc::_SC_PHYS_PAGES) };
let page_size = unsafe { libc::sysconf(libc::_SC_PAGESIZE) };
if pages > 0 && page_size > 0 {
return Some((pages as u64).saturating_mul(page_size as u64));
}
None
}
#[cfg(not(unix))]
{
None
}
}
fn resolve_disk_access_mode(
configured: DiskAccessMode,
file_size: u64,
mmap_min_size_bytes: u64,
memory_fraction: f64,
system_memory: Option<u64>,
direct_io_available: bool,
) -> DiskAccessMode {
if file_size == 0 {
return DiskAccessMode::Buffered;
}
match configured {
DiskAccessMode::Buffered => DiskAccessMode::Buffered,
DiskAccessMode::Mmap => DiskAccessMode::Mmap,
DiskAccessMode::Direct => DiskAccessMode::Direct,
DiskAccessMode::Auto => {
if file_size < mmap_min_size_bytes {
return DiskAccessMode::Buffered;
}
let fraction = if memory_fraction.is_finite() && memory_fraction > 0.0 {
memory_fraction.min(1.0)
} else {
0.5
};
if direct_io_available {
if let Some(mem) = system_memory {
let threshold = (mem as f64 * fraction) as u64;
if file_size > threshold {
return DiskAccessMode::Direct;
}
}
}
DiskAccessMode::Mmap
}
}
}
const fn direct_io_available() -> bool {
cfg!(all(
unix,
any(
target_os = "linux",
target_os = "android",
target_os = "macos"
)
))
}
#[cfg(unix)]
fn mmap_advice_for(prefetch: PrefetchMode) -> Option<memmap2::Advice> {
match prefetch {
PrefetchMode::Off | PrefetchMode::Auto => None,
PrefetchMode::Sequential => Some(memmap2::Advice::Sequential),
PrefetchMode::WillNeed => Some(memmap2::Advice::WillNeed),
}
}
#[cfg(unix)]
const POINT_MMAP_MADV_RANDOM_MIN_BYTES: u64 = 8 * 1024 * 1024;
static SSTABLES_OPEN_COUNT_BIG: AtomicI64 = AtomicI64::new(0);
static SSTABLES_OPEN_COUNT_BTI: AtomicI64 = AtomicI64::new(0);
fn sstables_open_count_for(format: &str) -> &'static AtomicI64 {
match format {
"bti" => &SSTABLES_OPEN_COUNT_BTI,
_ => &SSTABLES_OPEN_COUNT_BIG,
}
}
impl SSTableReader {
pub async fn open(path: &Path, config: &Config, platform: Arc<Platform>) -> Result<Self> {
let cache = super::build_chunk_cache(config);
Self::open_with_cache(path, config, platform, cache).await
}
pub async fn open_cancellable(
path: &Path,
config: &Config,
platform: Arc<Platform>,
cancel: crate::storage::scan_cancel::ScanCancel,
) -> Result<Self> {
let cache = super::build_chunk_cache(config);
Self::open_with_cache_cancellable(path, config, platform, cache, cancel).await
}
pub async fn open_with_cache(
path: &Path,
config: &Config,
platform: Arc<Platform>,
cache: Arc<crate::storage::cache::DecompressedChunkCache>,
) -> Result<Self> {
Self::open_with_cache_cancellable(
path,
config,
platform,
cache,
crate::storage::scan_cancel::ScanCancel::default(),
)
.await
}
pub async fn open_with_cache_cancellable(
path: &Path,
config: &Config,
platform: Arc<Platform>,
cache: Arc<crate::storage::cache::DecompressedChunkCache>,
cancel: crate::storage::scan_cancel::ScanCancel,
) -> Result<Self> {
use crate::observability::{self as obs, catalog};
use tracing::Instrument as _;
let span = tracing::debug_span!(
"sstable.reader.open",
file_size = tracing::field::Empty,
sstable_format = tracing::field::Empty,
);
let result = Self::open_inner(path, config, platform, cache, cancel)
.instrument(span.clone())
.await;
match &result {
Ok(reader) => {
let format = reader.sstable_format_label();
span.record("file_size", reader.stats.file_size);
span.record("sstable_format", format);
let now = sstables_open_count_for(format).fetch_add(1, Ordering::Relaxed) + 1;
obs::record_gauge(
catalog::SSTABLES_OPEN,
now,
&[(catalog::attr::SSTABLE_FORMAT, format.into())],
);
}
Err(e) => {
span.in_scope(|| obs::record_error(e, "reader"));
}
}
result
}
async fn open_inner(
path: &Path,
config: &Config,
platform: Arc<Platform>,
chunk_cache: Arc<crate::storage::cache::DecompressedChunkCache>,
cancel: crate::storage::scan_cancel::ScanCancel,
) -> Result<Self> {
let open_config = config.clone();
let version_gates = Arc::new(match VersionGates::from_path(path) {
Ok(gates) => gates,
Err(e @ Error::UnsupportedVersion { .. }) => return Err(e),
Err(e) => {
tracing::debug!(
"SSTableReader::open: could not derive VersionGates from {:?} ({}); \
defaulting to nb-compatible BIG gates",
path,
e
);
VersionGates::Big(BigVersionGates::nb_fallback())
}
});
let mut reader_config = SSTableReaderConfig::default();
reader_config.use_mmap = config.storage.use_mmap || mmap_enabled_via_env();
reader_config.mmap_min_size_bytes = config.storage.mmap_min_size_bytes;
let file_size = tokio::fs::metadata(path).await?.len();
let configured_mode = disk_access_mode_via_env().unwrap_or(config.storage.disk_access_mode);
let configured_mode =
if reader_config.use_mmap && matches!(configured_mode, DiskAccessMode::Buffered) {
DiskAccessMode::Mmap
} else {
configured_mode
};
let resolved_mode = resolve_disk_access_mode(
configured_mode,
file_size,
reader_config.mmap_min_size_bytes as u64,
config.storage.direct_io_memory_fraction,
system_memory_bytes(),
direct_io_available(),
);
let prefetch = prefetch_mode_via_env().unwrap_or(config.storage.prefetch);
let (source, scan_source) = Self::build_block_sources(
path,
file_size,
resolved_mode,
prefetch,
config.storage.direct_io_prefetch_bytes,
)
.await?;
let file = Arc::new(Mutex::new(source));
let point_source: Arc<dyn read_at::ReadAt> = match &scan_source {
ScanSource::Mapped(mmap) => {
#[cfg(unix)]
let point_mmap =
Self::point_read_mmap(path, file_size, mmap, POINT_MMAP_MADV_RANDOM_MIN_BYTES);
#[cfg(not(unix))]
let point_mmap = mmap.clone();
Arc::new(read_at::MmapReadAt::new(point_mmap))
}
#[cfg(unix)]
ScanSource::Direct { .. } => match read_at::DirectReadAt::open(path, file_size) {
Ok(d) => Arc::new(d) as Arc<dyn read_at::ReadAt>,
Err(e) => {
tracing::warn!(
"Direct-I/O point source for {} failed ({}); using buffered pread",
path.display(),
e
);
Arc::new(read_at::PlainFileReadAt::open(path, file_size)?)
}
},
ScanSource::Buffered { .. } => {
Arc::new(read_at::PlainFileReadAt::open(path, file_size)?)
}
};
let header_size = std::cmp::min(4096, file_size as usize);
let mut header_buffer = vec![0u8; header_size];
{
let mut file_guard = file.lock().await;
let bytes_read = file_guard.read(&mut header_buffer).await?;
header_buffer.truncate(bytes_read);
}
let (bti_partitions_db, bti_rows_db) = if matches!(*version_gates, VersionGates::Bti(_)) {
let base = extract_sstable_base_name(path).ok_or_else(|| {
Error::unsupported_format(format!(
"BTI (da) SSTable '{}' has a non-standard filename; cannot derive the \
sibling Partitions.db name required for trie point lookup (#831).",
path.display()
))
})?;
let parent = path.parent().unwrap_or_else(|| Path::new("."));
let partitions_path = parent.join(format!("{}-Partitions.db", base));
let partitions_bytes = tokio::fs::read(&partitions_path).await.map_err(|e| {
Error::unsupported_format(format!(
"BTI (da) SSTable '{}' is missing its sibling Partitions.db trie \
(expected '{}'): {}. BTI read support requires Partitions.db for \
partition-key point lookup (#831).",
path.display(),
partitions_path.display(),
e
))
})?;
let rows_path = parent.join(format!("{}-Rows.db", base));
let rows_bytes = tokio::fs::read(&rows_path).await.map_err(|e| {
Error::unsupported_format(format!(
"BTI (da) SSTable '{}' is missing its sibling Rows.db row-index trie \
(expected '{}'): {}. BTI read support requires Rows.db to resolve \
wide-partition point lookups (#909/#910).",
path.display(),
rows_path.display(),
e
))
})?;
(Some(Arc::new(partitions_bytes)), Some(Arc::new(rows_bytes)))
} else {
let none: Option<Arc<Vec<u8>>> = None;
(none.clone(), none)
};
use tracing::Instrument;
let config = crate::cql::config::ParserConfig::default();
let parser = SSTableParser::new(config)?;
let header = parse_header_with_version_detection(&header_buffer, path, &version_gates)
.instrument(tracing::debug_span!("sstable.reader.open.header_parse"))
.await
.map_err(|e| {
Error::corruption(format!(
"Failed to parse SSTable header for file '{}': {}. This indicates either \
file corruption or an unsupported SSTable format. File size: {} bytes, \
header buffer size: {} bytes.",
path.display(),
e,
file_size,
header_buffer.len()
))
})?;
let header_size = calculate_actual_header_size(&header, &header_buffer)?;
{
let mut file_guard = file.lock().await;
file_guard
.seek(std::io::SeekFrom::Start(header_size as u64))
.await?;
}
let compression_info = Self::load_compression_info_metadata(path, &platform).await?;
let compression_reader = match &compression_info {
Some(info) => Some(CompressionReader::new(info.algorithm_enum()?)),
None => None,
};
let crc_reader = if compression_info.is_none() {
Self::load_crc_reader(path, &header, file_size).await?
} else {
None
};
let components = Self::detect_component_files(path).await?;
if !components.is_empty() {
let integrity_issues = Self::validate_component_integrity(path, &components).await?;
if !integrity_issues.is_empty() {
tracing::warn!(
"Component integrity issues detected but proceeding with loading: {:?}",
integrity_issues
);
}
}
let index = Self::load_index(&file, &header)
.instrument(tracing::debug_span!("sstable.reader.open.load_index"))
.await?;
let bloom_filter = Self::load_bloom_filter(&file, &header, &platform, path)
.instrument(tracing::debug_span!("sstable.reader.open.load_bloom"))
.await?;
let summary_reader = Self::load_summary_reader(path, &platform)
.instrument(tracing::debug_span!("sstable.reader.open.load_summary"))
.await;
let summary_usable = summary_reader
.as_ref()
.map(|s| !s.get_entries().is_empty())
.unwrap_or(false);
let (index_reader, index_present_but_unloadable) =
match Self::load_index_reader(path, &platform, &cancel, summary_usable)
.instrument(tracing::debug_span!(
"sstable.reader.open.load_index_reader"
))
.await?
{
component_loading::IndexLoadOutcome::Loaded(reader) => (Some(*reader), false),
component_loading::IndexLoadOutcome::Absent => (None, false),
component_loading::IndexLoadOutcome::PresentButUnloadable => (None, true),
};
let statistics_reader = Self::load_statistics_reader(path, &platform)
.instrument(tracing::debug_span!("sstable.reader.open.load_statistics"))
.await?;
let mut header = header; if let Some(ref stats_reader) = statistics_reader {
let statistics = stats_reader.statistics();
let partition_columns = &statistics.serialization_header_partition_keys;
let clustering_columns = &statistics.serialization_header_clustering_keys;
let regular_columns = &statistics.serialization_header_columns;
if !partition_columns.is_empty()
|| !clustering_columns.is_empty()
|| !regular_columns.is_empty()
{
tracing::debug!(
"Populating header columns from Statistics.db SerializationHeader: {} partition keys, {} clustering keys, {} regular columns",
partition_columns.len(),
clustering_columns.len(),
regular_columns.len()
);
let mut merged_columns = Vec::with_capacity(
partition_columns.len() + clustering_columns.len() + regular_columns.len(),
);
merged_columns.extend_from_slice(partition_columns);
merged_columns.extend_from_slice(clustering_columns);
merged_columns.extend_from_slice(regular_columns);
header.columns = merged_columns;
}
}
let schema = if matches!(
header.cassandra_version,
CassandraVersion::V5_0NewBig
| CassandraVersion::V5_0Bti
| CassandraVersion::V5_0DataFormat
| CassandraVersion::V5_0FormatC
| CassandraVersion::V5_0FormatD
| CassandraVersion::V5_0FormatE
| CassandraVersion::V5_0FormatF
| CassandraVersion::V5_0FormatG
) {
match TableSchema::from_sstable_header(&header) {
Ok(s) => {
tracing::debug!(
"Extracted schema from SSTable header: {}.{} ({} columns, {} partition keys, {} clustering keys)",
s.keyspace,
s.table,
s.columns.len(),
s.partition_keys.len(),
s.clustering_keys.len()
);
Some(Arc::new(s))
}
Err(e) => {
tracing::warn!(
"Failed to extract schema from SSTable header for {}: {}. Schema-aware parsing will not be available.",
path.display(),
e
);
None
}
}
} else {
None
};
let block_count = compression_info
.as_ref()
.map(|ci| ci.chunk_offsets.len() as u64)
.unwrap_or(0);
let stats = SSTableReaderStats {
file_size,
entry_count: header.stats.row_count,
table_count: 1, block_count,
index_size: 0, bloom_filter_size: 0, compression_ratio: header.stats.compression_ratio,
cache_hit_rate: 0.0,
};
let generation = extract_generation_from_path(path);
let chunk_cache_id = {
use std::hash::{Hash, Hasher};
let mut h = std::collections::hash_map::DefaultHasher::new();
path.hash(&mut h);
generation.hash(&mut h);
h.finish()
};
let key_offset_cache = super::build_key_offset_cache(&open_config);
let generation_identity =
crate::storage::cache::GenerationIdentity::resolve(path, generation);
let endpoint_tokens = summary_reader.as_ref().and_then(|s| {
let first = s.get_first_key();
let last = s.get_last_key();
(!first.is_empty() && !last.is_empty()).then(|| {
(
crate::util::cassandra_murmur3::cassandra_murmur3_token(first),
crate::util::cassandra_murmur3::cassandra_murmur3_token(last),
)
})
});
Ok(Self {
file_path: arc_swap::ArcSwap::from_pointee(path.to_path_buf()),
file,
scan_source,
point_source,
header,
parser,
index,
bloom_filter,
compression_reader,
config: reader_config,
open_config,
platform,
stats,
#[cfg(feature = "tombstones")]
tombstone_merger: TombstoneMerger::new(),
generation,
actual_header_size: header_size,
index_reader,
index_present_but_unloadable,
summary_reader,
endpoint_tokens,
statistics_reader,
schema_registry: None, schema,
#[cfg(feature = "state_machine")]
registry_schema: None, udt_registry: None, scan_cancel: crate::storage::scan_cancel::ScanCancel::default(), compression_info: compression_info.map(Arc::new),
crc_reader,
verified_uncompressed_chunks: std::sync::Mutex::new(std::collections::HashSet::new()),
version_gates,
bti_partitions_db,
bti_rows_db,
chunk_cache,
chunk_cache_id,
bti_lookup_memo: std::sync::Mutex::new(None),
key_offset_cache,
generation_identity,
})
}
#[cfg(test)]
pub(crate) async fn is_mmap_backed(&self) -> bool {
self.file.lock().await.is_mmap()
}
#[cfg(test)]
pub(crate) fn clone_point_source(&self) -> Arc<dyn read_at::ReadAt> {
self.point_source.clone()
}
#[cfg(test)]
pub(crate) fn set_point_source(&mut self, src: Arc<dyn read_at::ReadAt>) {
self.point_source = src;
}
#[cfg(all(test, unix))]
pub(crate) async fn is_direct_backed(&self) -> bool {
self.file.lock().await.is_direct()
}
async fn open_buffered_sources(
path: &Path,
file_size: u64,
) -> Result<(BlockSource, ScanSource)> {
crate::storage::sstable::read_work_counters::record_file_open();
Ok((
BlockSource::buffered_sized(File::open(path).await?, file_size),
ScanSource::Buffered {
file_len: file_size,
},
))
}
async fn build_block_sources(
path: &Path,
file_size: u64,
mode: DiskAccessMode,
prefetch: PrefetchMode,
prefetch_bytes: usize,
) -> Result<(BlockSource, ScanSource)> {
let direct_window = if matches!(prefetch, PrefetchMode::Off) {
1
} else {
prefetch_window::clamp_direct_prefetch_window(
prefetch_bytes,
prefetch_window::DEFAULT_COMPRESSION_CHUNK_BYTES,
)
};
match mode {
DiskAccessMode::Buffered => Self::open_buffered_sources(path, file_size).await,
DiskAccessMode::Mmap => match Self::map_file(path) {
Ok(mmap) => {
tracing::debug!(
"Opened {} via memory map ({} bytes)",
path.display(),
file_size
);
#[cfg(unix)]
if let Some(advice) = mmap_advice_for(prefetch) {
if let Err(e) = mmap.advise(advice) {
tracing::debug!(
"madvise({:?}) on {} failed: {}",
advice,
path.display(),
e
);
}
}
let mmap = Arc::new(mmap);
Ok((BlockSource::mapped(mmap.clone()), ScanSource::Mapped(mmap)))
}
Err(e) => {
tracing::warn!(
"Memory-mapping {} failed ({}); falling back to buffered I/O",
path.display(),
e
);
Self::open_buffered_sources(path, file_size).await
}
},
DiskAccessMode::Direct => {
#[cfg(unix)]
{
match source::DirectCursor::open(path, direct_window) {
Ok(cursor) => {
tracing::debug!(
"Opened {} via direct I/O ({} bytes, {}-byte window)",
path.display(),
file_size,
direct_window
);
Ok((
BlockSource::direct(cursor),
ScanSource::Direct {
window: direct_window,
file_len: file_size,
},
))
}
Err(e) => {
tracing::warn!(
"Direct I/O on {} failed ({}); falling back to buffered I/O",
path.display(),
e
);
Self::open_buffered_sources(path, file_size).await
}
}
}
#[cfg(not(unix))]
{
let _ = direct_window;
tracing::warn!(
"Direct I/O is unavailable on this platform; using buffered I/O for {}",
path.display()
);
Self::open_buffered_sources(path, file_size).await
}
}
DiskAccessMode::Auto => Self::open_buffered_sources(path, file_size).await,
}
}
fn map_file(path: &Path) -> Result<memmap2::Mmap> {
crate::storage::sstable::read_work_counters::record_file_open();
let std_file = std::fs::File::open(path)?;
let mmap = unsafe { memmap2::MmapOptions::new().map(&std_file)? };
Ok(mmap)
}
#[cfg(unix)]
fn point_read_mmap(
path: &Path,
file_size: u64,
scan_mmap: &Arc<memmap2::Mmap>,
min_random_bytes: u64,
) -> Arc<memmap2::Mmap> {
if file_size >= min_random_bytes {
if let Ok(std_file) = std::fs::File::open(path) {
match unsafe { memmap2::MmapOptions::new().map(&std_file) } {
Ok(point_mmap) => match point_mmap.advise(memmap2::Advice::Random) {
Ok(()) => {
tracing::debug!(
"Dedicated MADV_RANDOM point-read mapping for {} ({} bytes, #2210)",
path.display(),
file_size
);
return Arc::new(point_mmap);
}
Err(e) => tracing::debug!(
"madvise(RANDOM) on dedicated point map for {} failed ({}); \
sharing scan mapping",
path.display(),
e
),
},
Err(e) => tracing::debug!(
"Dedicated point map for {} failed ({}); sharing scan mapping",
path.display(),
e
),
}
}
}
scan_mmap.clone()
}
async fn load_compression_info_metadata(
path: &Path,
_platform: &Arc<Platform>,
) -> Result<Option<CompressionInfo>> {
use tokio::fs::File;
use tokio::io::AsyncReadExt;
let parent_dir = path.parent().unwrap_or(Path::new("."));
let base_name = crate::storage::sstable::version_gate::SsTableDescriptor::parse(path)
.ok()
.map(|d| format!("{}-{}-{}", d.version, d.sstable_id, d.format.as_str()));
if let Some(base) = base_name {
let compression_info_path = parent_dir.join(format!("{}-CompressionInfo.db", base));
if compression_info_path.exists() {
let mut file = File::open(&compression_info_path).await?;
let mut buffer = Vec::new();
file.read_to_end(&mut buffer).await?;
let info = CompressionInfo::parse(&buffer).map_err(|e| {
Error::UnsupportedFormat(format!(
"Failed to parse CompressionInfo.db at {}: {}",
compression_info_path.display(),
e
))
})?;
tracing::debug!(
"Loaded CompressionInfo: algorithm={}, chunk_length={}, chunks={}",
info.algorithm,
info.chunk_length,
info.chunk_offsets.len()
);
return Ok(Some(info));
}
}
Ok(None)
}
async fn load_crc_reader(
path: &Path,
header: &SSTableHeader,
data_len: u64,
) -> Result<Option<Arc<crc::CrcDb>>> {
if matches!(header.cassandra_version, CassandraVersion::V5_0Bti) {
return Ok(None);
}
let parent_dir = path.parent().unwrap_or(Path::new("."));
let Some(base) = compression::extract_sstable_base_name(path) else {
return Ok(None);
};
let crc_path = parent_dir.join(format!("{}-CRC.db", base));
if !crc_path.exists() {
tracing::warn!(
"CRC.db absent for uncompressed SSTable {} — proceeding without \
read-time per-chunk CRC verification (warn-and-proceed, issue #1396)",
path.display()
);
return Ok(None);
}
let crc = crc::CrcDb::open(&crc_path, data_len).await.map_err(|e| {
Error::corruption(format!(
"Failed to parse CRC.db at {}: {}",
crc_path.display(),
e
))
})?;
tracing::debug!(
"Loaded CRC.db for {}: chunk_size={}, chunks={}",
path.display(),
crc.chunk_size(),
crc.chunk_count()
);
Ok(Some(Arc::new(crc)))
}
#[cfg(feature = "state_machine")]
#[deprecated(
note = "attaches the registry but CANNOT pre-resolve the sync schema-fallback cache \
(would need a forbidden block_on, issue #1692); registry schemas will not be \
available to a subsequent sync `get_table_schema`. Use the async \
`attach_schema_registry` (which pre-resolves), or call `resolve_registry_schema` \
after this from an async context, for registry-schema-aware reads."
)]
pub fn set_schema_registry(
&mut self,
schema_registry: Arc<tokio::sync::RwLock<crate::schema::SchemaRegistry>>,
) {
self.schema_registry = Some(schema_registry);
self.registry_schema = None;
tracing::debug!(
"Schema registry set for {}.{} - enabling schema-driven digest computation",
self.header.keyspace,
self.header.table_name
);
}
#[cfg(feature = "state_machine")]
pub async fn attach_schema_registry(
&mut self,
schema_registry: Arc<tokio::sync::RwLock<crate::schema::SchemaRegistry>>,
) {
#[allow(deprecated)]
self.set_schema_registry(schema_registry);
self.resolve_registry_schema().await;
}
#[cfg(not(feature = "state_machine"))]
pub fn set_schema_registry(&mut self, schema_registry: Arc<crate::schema::SchemaRegistry>) {
self.schema_registry = Some(schema_registry);
tracing::debug!(
"Schema registry set for {}.{} - enabling schema-driven digest computation",
self.header.keyspace,
self.header.table_name
);
}
pub fn set_udt_registry(&mut self, registry: crate::schema::UdtRegistry) {
self.udt_registry = Some(registry);
tracing::debug!(
"UDT registry set for {}.{} - enabling UDT-aware collection parsing",
self.header.keyspace,
self.header.table_name
);
}
pub fn has_udt_registry(&self) -> bool {
self.udt_registry.is_some()
}
pub fn file_path(&self) -> PathBuf {
self.file_path.load().as_ref().clone()
}
pub fn file_size(&self) -> u64 {
self.stats.file_size
}
pub fn rebind_path(&self, new_path: &Path) {
self.file_path.store(Arc::new(new_path.to_path_buf()));
if let Some(index_reader) = self.index_reader.as_ref() {
if let Some(index_sibling) = index_db_sibling(new_path) {
index_reader.rebind_path(&index_sibling);
}
}
}
pub fn endpoint_tokens(&self) -> Option<(i64, i64)> {
self.endpoint_tokens
}
pub fn index_is_materialized(&self) -> bool {
self.index_reader
.as_ref()
.map(|ir| ir.is_materialized())
.unwrap_or(true)
}
pub fn set_scan_cancel(&mut self, cancel: crate::storage::scan_cancel::ScanCancel) {
self.scan_cancel = cancel;
}
pub async fn stats(&self) -> Result<&SSTableReaderStats> {
Ok(&self.stats)
}
pub async fn close(self) -> Result<()> {
debug!("Closing SSTable reader for {:?}", self.file_path());
Ok(())
}
pub fn calculate_header_size(&self) -> usize {
self.actual_header_size
}
pub fn cassandra_version(&self) -> CassandraVersion {
self.header.cassandra_version
}
pub(crate) fn sstable_format_label(&self) -> &'static str {
match *self.version_gates {
VersionGates::Bti(_) => "bti",
VersionGates::Big(_) => "big",
}
}
pub fn has_uint_deletion_time(&self) -> bool {
match *self.version_gates {
VersionGates::Big(ref g) => g.has_uint_deletion_time,
VersionGates::Bti(ref g) => g.has_uint_deletion_time,
}
}
pub fn format_version(&self) -> Result<String> {
let file_path = self.file_path();
let filename = file_path
.file_name()
.and_then(|f| f.to_str())
.ok_or_else(|| {
Error::InvalidPath(format!("Invalid SSTable filename: {:?}", file_path))
})?;
let parts: Vec<&str> = filename.split('-').collect();
if parts.is_empty() {
return Err(Error::InvalidFormat(format!(
"Cannot extract format version from filename: {}",
filename
)));
}
Ok(parts[0].to_string())
}
pub fn header(&self) -> &SSTableHeader {
&self.header
}
pub fn schema(&self) -> Option<&TableSchema> {
self.schema.as_deref()
}
pub fn effective_schema(&self) -> Option<TableSchema> {
self.get_table_schema(None)
}
pub fn extract_write_time_from_entry(&self, _key: &RowKey, row: &ScanRow) -> i64 {
use tracing::warn;
match row {
ScanRow::Marker(Value::Tombstone(info)) => info.deletion_time,
_ => std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_micros() as i64)
.unwrap_or_else(|e| {
warn!("Failed to get system time: {}; using fallback value 0", e);
0
}),
}
}
}
impl Drop for SSTableReader {
fn drop(&mut self) {
use crate::observability::{self as obs, catalog};
let format = self.sstable_format_label();
let now = sstables_open_count_for(format).fetch_sub(1, Ordering::Relaxed) - 1;
obs::record_gauge(
catalog::SSTABLES_OPEN,
now,
&[(catalog::attr::SSTABLE_FORMAT, format.into())],
);
}
}
impl std::fmt::Debug for SSTableReader {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("SSTableReader")
.field("file_path", &self.file_path())
.field("header", &self.header)
.field("has_index", &self.index.is_some())
.field("has_bloom_filter", &self.bloom_filter.is_some())
.field("compression", &self.header.compression.algorithm)
.field("stats", &self.stats)
.finish()
}
}
pub async fn open_sstable_reader(
path: &Path,
config: &Config,
platform: Arc<Platform>,
) -> Result<SSTableReader> {
SSTableReader::open(path, config, platform).await
}