cqlite_core/storage/sstable/

mod.rs

//! SSTable (Sorted String Table) implementation

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; // Cross-generation read reconciliation (issues #883/#885/#957/#1579).
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;
/// Authoritative zstd frame-header parsing for dictionary detection (issue #1414).
///
/// Gated on the `zstd` feature: the only consumer is the zstd decode path in
/// `chunk_decompressor`, which is itself `#[cfg(feature = "zstd")]`. Without this
/// gate the module's items are unused under a zstd-off build (e.g. the
/// `Minimal Compression Build` CI lane `--features=lz4,snappy`) and fail
/// `-D warnings`.
#[cfg(feature = "zstd")]
pub mod zstd_frame;
pub use reader::SSTableReader;
/// Explicit directory refresh (issue #1749): re-scan + atomic diff-and-swap of
/// the held reader set. Not `state_machine`-gated — meaningful for minimal builds.
pub mod refresh;
pub use refresh::RefreshReport;
mod reverse_scan; // BIG reverse partition iteration (issue #1184); file is tombstones-gated.
pub mod row_cell_state_machine;
/// Cross-SSTable scan ordering: k-way merge in Cassandra token order (issue #1580).
mod scan_merge;
pub mod statistics_reader;
pub mod stream_merge_probe; // Multi-generation streaming-merge resident-rows probe (issue #1579, D3).
#[cfg(feature = "tombstones")]
pub mod tombstone_merger;
pub mod validation;
pub mod verify; // Verifier contract for compressed + corrupted SSTables (epic #970, issue #1000).
pub use verify::{
    verify_sstable, verify_sstable_generation, VerifyErrorClass, VerifyFinding, VerifyMode,
    VerifyReport,
};

// M5: SSTable writer components (Issue #359)
#[cfg(feature = "write-support")]
pub mod writer;

// Test modules
/// F1: scan must not hold the table_readers read guard across its I/O (Issue #1591).
#[cfg(test)]
mod issue_1591_scan_lock_test;
/// F4: the fan-out scan merge must not deadlock under admission when it fans out
/// to more generations than the admission cap (Issue #1594).
#[cfg(all(test, feature = "scan-offload-probe"))]
mod issue_1594_fanout_deadlock_test;
/// VG1: Thread VersionGates through the read path (Issue #653).
#[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;
/// S3 verification tests for Index.db/Summary.db/BTI (epic #622, issue #625).
#[cfg(test)]
mod s3_verification_test;
/// S4 verification tests for Statistics.db/CompressionInfo.db/Filter.db (epic #622, issue #626).
#[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"))] // #1917 concat fallbacks; tombstones uses k-way merge
use crate::util::cassandra_murmur3::cmp_partition_keys_by_token;
use crate::{types::TableId, Config, Result, RowKey, ScanRow};

/// Maximum directory depth when scanning for SSTable files.
///
/// Writer creates `data_dir/keyspace/table/nb-{gen}-big-*.db` (2 levels deep),
/// so 3 levels provides a safety margin.
pub(crate) const MAX_SSTABLE_SCAN_DEPTH: usize = 3;

/// The AUTHORITATIVE partition-key shape of a table, derived schema-less from the
/// SSTable Statistics.db SerializationHeader (issue #1750).
///
/// Carries ONLY facts Cassandra actually serialises: how many partition-key
/// components and clustering keys the table has, and the REAL names of the
/// non-key (regular + static) columns. The partition-/clustering-key column NAMES
/// are deliberately absent — Cassandra never serialises them, so any pk-name a
/// schema-less parser reports is a synthesised placeholder that must not drive
/// routing decisions (no-heuristics mandate #28).
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PartitionKeyShape {
    /// Number of partition-key components (a single-component pk is `1`).
    pub partition_key_count: usize,
    /// Number of clustering-key columns.
    pub clustering_key_count: usize,
    /// The real names of the non-key (regular + static) columns.
    pub non_key_column_names: std::collections::HashSet<String>,
    /// For a single-component pk ONLY: the sole partition-key column's authoritative
    /// CQL type (from the SerializationHeader `keyType`) and its header-carried name.
    ///
    /// The CQL type is authoritative — Cassandra serialises partition-key TYPES even
    /// though it never serialises their NAMES (issue #1750, roborev 3784). It drives
    /// the TYPED single-component key encoding so a schema-less `WHERE int_pk = 42`
    /// builds the 4-byte `int` key Cassandra wrote, not an 8-byte `BigInt` key. The
    /// `name` is the header's synthesised placeholder (`id` for uuid/timeuuid, else
    /// `partition_key`) — NOT a trusted identity, only the label under which
    /// `build_row_from_scan` reconstructs the pk column so the seeked row can be
    /// re-validated against the predicate. `None` for a composite pk (the typed
    /// single-component fast path does not apply) or when no reader exposed a
    /// single-component pk type.
    pub single_pk_component: Option<PartitionKeyComponent>,
}

/// The authoritative type + header-synthesised name of a single-component
/// partition key (issue #1750, roborev 3784). See [`PartitionKeyShape`].
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PartitionKeyComponent {
    /// The header's synthesised pk column name (`id` / `partition_key`). NOT a
    /// trusted identity — only a label for pk-column reconstruction.
    pub name: String,
    /// The authoritative CQL type of the pk component (from `keyType`).
    pub cql_type: String,
}

/// Resolve the AUTHORITATIVE [`PartitionKeyShape`] by UNIONing the non-key column
/// names across EVERY reader's SerializationHeader while requiring CONSISTENT key
/// metadata (issue #1750, roborev 3786).
///
/// This is the pure core of [`SSTableManager::partition_key_shape`], factored out so
/// the multi-reader union + consistency logic is unit-testable without SSTable
/// binaries. `headers` is one `&[ColumnInfo]` per resolved reader for the table (an
/// empty or non-header reader contributes nothing).
///
/// A multi-generation / schema-evolved table can add a REGULAR column in a LATER
/// generation absent from an earlier generation's header. Using only the first
/// header would omit that column from the non-key name set, so a
/// `WHERE <later-gen-regular-col> = <val>` could be MISCLASSIFIED as a partition-key
/// seek. To prevent that, the non-key names are UNIONed across all headers, and the
/// partition-key count, clustering-key count, and single-component pk type+name must
/// agree across every header that has partition-key columns.
///
/// The classifier proves a column IS the partition key by its ABSENCE from the unioned
/// non-key set — which is sound ONLY if the union is COMPLETE, i.e. every reader's
/// columns were visible. So this is fail-safe on incomplete metadata (roborev 3788):
///
/// Returns `None` when
/// - there are no readers (fail-safe: caller full-scans), OR
/// - ANY resolved reader lacks a usable partition-key shape — an empty/missing
///   SerializationHeader, or a populated header exposing no partition-key column
///   (we cannot see that reader's regular columns, so the union is incomplete and a
///   column could be misclassified as the pk by absence), OR
/// - the headers' key metadata is INCONSISTENT (pk-count / clustering-count / single-pk
///   type+name disagreement — should never happen for one table, but if it did we
///   cannot trust the shape).
///
/// `Some(shape)` is returned only when EVERY reader contributed a usable, consistent
/// pk shape. All facts come from SerializationHeaders — no heuristics (#28).
fn partition_key_shape_from_headers<'a>(
    headers: impl IntoIterator<Item = &'a [crate::parser::header::ColumnInfo]>,
) -> Option<PartitionKeyShape> {
    // Key metadata the first header-bearing reader established; every subsequent
    // header MUST match it or the whole shape is declined (inconsistent → None).
    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 {
        // COMPLETE-UNION fail-safe (roborev 3788): the classifier proves a column is
        // the partition key by its ABSENCE from `non_key_column_names`. That proof is
        // only sound if we can see EVERY reader's columns. A reader with no usable
        // SerializationHeader (empty columns) would silently drop its regular columns
        // from the union, so a regular column present ONLY in that headerless reader
        // could be misclassified as the pk. Decline the whole shape → the caller
        // full-scans (always correct). No heuristics (#28).
        if columns.is_empty() {
            return None;
        }
        let mut partition_key_count = 0usize;
        let mut clustering_key_count = 0usize;
        // The sole partition-key component's authoritative CQL type + header-
        // synthesised name (issue #1750, roborev 3784). Recorded for the FIRST pk
        // column seen in THIS header; only surfaced when the pk is single-component.
        // Cassandra serialises pk TYPES (authoritative) but not NAMES (`name` is a
        // placeholder used only for pk reconstruction).
        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, _) => {
                    // Union non-key names across ALL readers (roborev 3786): a
                    // regular column present only in a later generation must be
                    // recognised, else it could be misread as the partition key.
                    non_key_column_names.insert(col.name.clone());
                }
            }
        }
        // A SerializationHeader always names at least one partition-key component; a
        // header that populated columns yet exposes NO partition-key column is not a
        // usable pk shape. Under the complete-union fail-safe (roborev 3788) we cannot
        // trust the union when any reader's pk shape is unknown, so decline → the
        // caller full-scans rather than record a bogus all-zero shape.
        if partition_key_count == 0 {
            return None;
        }
        // The typed single-component pk type is authoritative ONLY for a single-
        // component key; drop it for a composite pk (that fast path does not apply
        // and a composite key isn't one raw value).
        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) => {
                // Require CONSISTENT key metadata across readers; any disagreement
                // means we cannot trust the shape → decline (#28).
                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,
    })
}

/// SSTable file identifier
#[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 {
    /// Create a new SSTable ID with timestamp using Cassandra naming convention
    pub fn new() -> Self {
        let timestamp = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_micros();
        // Use Cassandra naming convention: <keyspace>-<table>-<generation>-<format>-Data.db
        // For generated files, we'll use a simplified pattern: sstable-<timestamp>-big-Data.db
        Self(format!("sstable-{}-big-Data.db", timestamp))
    }

    /// Create SSTable ID from filename
    pub fn from_filename(filename: &str) -> Self {
        Self(filename.to_string())
    }

    /// Get the filename
    pub fn filename(&self) -> &str {
        &self.0
    }
}

/// Extract table name from SSTable directory path.
///
/// SSTable files are stored in directories named `<table_name>-<uuid>`.
/// For example: `simple_table-6aa08200a25111f0a3fef1a551383fb9/na-1-big-Data.db`
///
/// This function extracts the table name portion by:
/// 1. Getting the parent directory name
/// 2. Splitting on '-' and removing the UUID suffix
///
/// Removes the UUID suffix from directory names like:
/// - `simple_table-6aa08200a25111f0a3fef1a551383fb9` → `simple_table`
/// - `my-test-table-UUID` → `my-test-table`
///
/// Returns `None` if the path doesn't contain a valid directory component.
///
/// Note: Table names can contain hyphens, so we need to be careful to only remove the UUID.
/// UUIDs in Cassandra directory names are 32 hex chars without hyphens (e.g., 6aa08200a25111f0a3fef1a551383fb9).
pub(crate) fn extract_table_name(sstable_path: &Path) -> Option<String> {
    // Get the parent directory name
    let dir_name = sstable_path.parent()?.file_name()?.to_str()?;

    // Find the last occurrence of '-' followed by 32 hex characters (UUID without hyphens)
    // Cassandra UUIDs in directory names are formatted as: tablename-<32-hex-chars>
    if let Some(uuid_start) = dir_name.rfind('-') {
        let potential_uuid = &dir_name[uuid_start + 1..];

        // Check if this looks like a UUID (32 hex characters)
        if potential_uuid.len() == 32 && potential_uuid.chars().all(|c| c.is_ascii_hexdigit()) {
            // Extract everything before the UUID
            return Some(dir_name[..uuid_start].to_string());
        }
    }

    // If we couldn't find a UUID pattern, return the whole directory name
    Some(dir_name.to_string())
}

/// Extract the fully-qualified table key (`"keyspace.table"`) from an SSTable path.
///
/// Cassandra on-disk layout is: `<data_dir>/<keyspace>/<table>-<uuid>/<sstable_files>`
///
/// This function walks up two directory levels from the SSTable file to identify both the
/// table directory (`parent`) and keyspace directory (`grandparent`), producing a
/// `"keyspace.table"` key that uniquely identifies a table across keyspaces.
///
/// When datasets-v3 added `test_oa.simple_table` alongside the existing
/// `test_basic.simple_table`, using the unqualified name `"simple_table"` as the
/// `table_readers` key caused both tables' SSTables to be registered under the same
/// entry, returning combined rows for any query.  This function is the authoritative
/// source of table identity for `SSTableManager` (Issue #680).
///
/// # Returns
///
/// `Some((keyspace, table_name))` when both directory levels can be extracted;
/// `None` if the path does not contain enough components (e.g., a flat test directory).
///
/// # Examples
///
/// ```text
/// ".../sstables/test_basic/simple_table-6aa08200a25111f0a3fef1a551383fb9/nb-1-big-Data.db"
///   → Some(("test_basic", "simple_table"))
///
/// ".../sstables/test_oa/simple_table-4b7cd05064e711f1bd3ac7dbf655c673/oa-2-big-Data.db"
///   → Some(("test_oa", "simple_table"))
///
/// "nb-1-big-Data.db"   (flat, no parent dirs)
///   → None
/// ```
pub fn extract_keyspace_and_table_name(sstable_path: &Path) -> Option<(String, String)> {
    let table_name = extract_table_name(sstable_path)?;

    // The keyspace directory is the grandparent of the SSTable file:
    //   <keyspace>/<table-uuid>/<sstable_file>
    let keyspace = sstable_path
        .parent() // table-uuid dir
        .and_then(|p| p.parent()) // keyspace dir
        .and_then(|p| p.file_name())
        .and_then(|n| n.to_str())
        .map(|s| s.to_string())?;

    Some((keyspace, table_name))
}

/// Return `true` if the filename is a macOS AppleDouble resource-fork sidecar.
///
/// macOS creates `._<name>` companion files when copying to non-Apple filesystems
/// (HFS+→FAT32, SMB shares, CI artifact tarballs, etc.).  These are NOT valid
/// Cassandra SSTable files even though they share the `-Data.db` suffix.
///
/// This predicate is the single point of truth for the `._*` filter; both
/// `load_from_table_directories` and `find_data_files` call it so the guard can
/// never silently diverge.  See Issue #481.
#[inline]
fn is_apple_double_sidecar(filename: &str) -> bool {
    filename.starts_with("._")
}

/// Deterministic test gate that pauses a scan mid-flight so a test can observe
/// whether the `table_readers` read guard is held across the scan's I/O (issue
/// #1591). Armed per-manager (never process-globally) so parallel tests never
/// interfere: only scans on the manager whose gate is armed ever wait. No
/// wall-clock sleeps — the scan signals `reached` when it arrives and blocks on
/// `release` until the test lets it proceed.
#[cfg(test)]
pub(crate) mod scan_gate {
    use std::sync::Arc;
    use tokio::sync::Notify;

    /// Two one-shot rendezvous points shared between the paused scan and the test.
    #[derive(Debug, Default)]
    pub(crate) struct Gate {
        /// Scan → test: "I have arrived at the gate."
        pub reached: Notify,
        /// Test → scan: "you may proceed."
        pub release: Notify,
    }

    /// Called from inside `scan`'s per-reader loop. When a gate is armed on this
    /// manager, signal arrival and block until the test releases it; otherwise a
    /// no-op. Placed at the per-reader I/O so it sits INSIDE the read-guarded
    /// region before the fix and OUTSIDE it after (the guard is dropped at the
    /// snapshot boundary) — the exact seam the test asserts on.
    pub(crate) async fn wait_if_armed(gate: &Option<Arc<Gate>>) {
        if let Some(gate) = gate {
            gate.reached.notify_one();
            gate.release.notified().await;
        }
    }
}

/// SSTable manager that handles multiple SSTable files
#[derive(Debug)]
pub struct SSTableManager {
    /// Base directory for SSTable files
    base_path: PathBuf,

    /// Active SSTable readers indexed by ID
    readers: Arc<RwLock<HashMap<SSTableId, Arc<reader::SSTableReader>>>>,

    /// Table name to SSTable readers mapping
    /// Maps table names (e.g., "simple_table") to their corresponding SSTable readers
    pub(crate) table_readers: Arc<RwLock<HashMap<String, Vec<Arc<reader::SSTableReader>>>>>,

    /// Platform abstraction
    platform: Arc<Platform>,

    /// Configuration
    config: Config,

    /// How this manager discovers SSTable generations, recorded so
    /// [`SSTableManager::refresh_tables`] re-runs the same discovery (issue #1749).
    discovery_source: refresh::DiscoverySource,

    /// Serializes concurrent [`SSTableManager::refresh_tables`] calls end-to-end
    /// (discovery through swap) so two refreshes cannot interleave — a stale
    /// discovery set from one refresh can never remove a generation a concurrent
    /// refresh just added (TOCTOU, issue #1749, Decision 4). Held ONLY across a
    /// refresh; queries never take this lock (they use the `RwLock` read guards).
    refresh_lock: Arc<Mutex<()>>,

    /// Schema registry for schema-aware operations (feature-gated)
    #[cfg(feature = "state_machine")]
    schema_registry: Arc<RwLock<Option<Arc<RwLock<crate::schema::SchemaRegistry>>>>>,

    /// Shared, bytes-bounded decompressed-chunk cache (issue #1567, Epic B/B1),
    /// sized once from `config.memory.block_cache.max_size` when
    /// `config.memory.block_cache.enabled` is `true` (the default). Cloned into
    /// every reader this manager opens (via `open_reader_with_schema` →
    /// `SSTableReader::open_with_cache`) so all readers of one dataset — including
    /// those added by a later `refresh_tables` — share one cache and one budget.
    ///
    /// When `block_cache.enabled == false` (issue #1568) this is a genuine no-op
    /// [`disabled`](crate::storage::cache::DecompressedChunkCache::disabled) cache
    /// so reads bypass caching entirely, and [`stats_chunk_cache`](Self::stats_chunk_cache)
    /// reports `None` so the memory-stats surface reads a structural zero.
    pub(crate) chunk_cache: Arc<crate::storage::cache::DecompressedChunkCache>,

    /// Per-manager deterministic scan gate (issue #1591 test infrastructure).
    /// `None` in production; a test arms it via [`arm_scan_gate`](Self::arm_scan_gate)
    /// to pause this manager's scans mid-flight. Being per-manager (not a global)
    /// keeps parallel tests isolated.
    #[cfg(test)]
    scan_gate: std::sync::Mutex<Option<Arc<scan_gate::Gate>>>,
}

/// Build the reader-facing decompressed-chunk cache honoring the advertised
/// `config.memory.block_cache.enabled` toggle (issue #1568). When enabled (the
/// default) the cache is sized from `block_cache.max_size`; when disabled it is a
/// genuine no-op cache so reads bypass caching entirely rather than the toggle
/// being decorative.
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())
    }
}

/// Build a reader's per-reader key→partition-offset cache (issue #1570, B4),
/// honoring the same `config.memory.block_cache.enabled` read-cache toggle B2
/// established (issue #1568). When enabled (the default) the cache is bounded by an
/// approximate BYTE budget of
/// [`DEFAULT_KEY_CACHE_BYTES`](crate::storage::cache::DEFAULT_KEY_CACHE_BYTES);
/// when disabled it is a genuine no-op cache so the point-read path bypasses key
/// caching entirely rather than the toggle being decorative. The budget is a small
/// constant, not a new config knob (the audit forbids a decorative one).
///
/// **Aggregate memory** (`<128MB` budget): this cache is per-reader (design D2), so
/// with no global ceiling the resident footprint is `≈ N_open_readers ×
/// DEFAULT_KEY_CACHE_BYTES` for typical (small) keys. Bounding each reader by BYTES
/// (not an entry count) makes this **independent of key size** for the typical case
/// — the #1570 roborev fix: partition keys are variable-length (composite/text up to
/// ~64 KB), so a count cap of 4096 entries did NOT bound resident bytes (worst case
/// ~40 readers × 4096 × ~1 KB ≈ 160 MB, over budget). The `≈` is deliberate: the
/// eviction guard never evicts the just-resolved entry, so each of the 16 shards can
/// retain one oversized (~64 KB) entry beyond its 32 KiB per-shard budget, making a
/// single reader's true worst case ~1 MB (~2× the nominal 512 KiB) and the exact
/// upper bound `N × (B + shard_count × max_key_bytes)`. With a 512 KiB per-reader
/// budget, ~40 open generations is `≈ 40 × 512 KiB ≈ 20 MB` and even ~128 readers is
/// `≈ 128 × 512 KiB = 64 MB`, both comfortably within `<128MB` for typical keys.
/// Allocation stays occupancy-proportional, so idle readers cost ~nothing. See
/// [`DEFAULT_KEY_CACHE_BYTES`](crate::storage::cache::DEFAULT_KEY_CACHE_BYTES) for
/// the full derivation. A single global bounded cache keyed on `(sstable_id, key)`
/// would bound the aggregate regardless of reader count and is a deferred future
/// optimization (see `design.md` "Deferred").
pub(crate) fn build_key_offset_cache(
    config: &Config,
) -> Arc<crate::storage::cache::KeyOffsetCache> {
    use crate::storage::cache::{KeyOffsetCache, DEFAULT_KEY_CACHE_BYTES};
    if config.memory.block_cache.enabled {
        Arc::new(KeyOffsetCache::with_budget_bytes(DEFAULT_KEY_CACHE_BYTES))
    } else {
        Arc::new(KeyOffsetCache::disabled())
    }
}

impl SSTableManager {
    /// The shared chunk cache to expose to the memory-stats surface, or `None`
    /// when block caching is disabled (issue #1568).
    ///
    /// Returns `Some` only when `config.memory.block_cache.enabled` is `true`; a
    /// disabled cache is a no-op the stats shell must report as a structural zero
    /// (via `MemoryManager::new`), never as an active cache.
    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
        }
    }

    /// Process-level aggregate of the per-reader key→partition-offset caches
    /// (issue #1571, B5). Sums each reader's real observability counters (hits,
    /// misses, evictions, resident bytes, capacity bytes) into one snapshot.
    ///
    /// Counts each distinct reader **exactly once** by unioning the by-id
    /// `self.readers` map and the by-name `table_readers` map and deduping on
    /// `Arc::as_ptr` (roborev #1571 Low). Deduping is required for correctness in
    /// **both** directions: (a) the two maps re-reference the same reader `Arc`s,
    /// so a naive sum of both would double-count; (b) crucially, `self.readers` is
    /// **not** a strict superset of `table_readers` — `SSTableId::from_filename`
    /// keys the by-id map on the bare generation filename (e.g. `nb-1-big-Data.db`),
    /// so two tables sharing that filename collide and the by-id map keeps only the
    /// last-inserted reader while `table_readers` retains both under distinct keys.
    /// Iterating `self.readers` alone would therefore **silently omit** the
    /// overwritten reader's cache and under-count the aggregate. The pointer-dedup
    /// union counts every physically-open reader once regardless of collisions.
    /// Every field is read from a live counter/aggregate — no fabricated values.
    ///
    /// Lock ordering matches every writer (`readers` before `table_readers`), so
    /// acquiring both read guards here cannot deadlock.
    pub(crate) async fn aggregate_key_cache_stats(
        &self,
    ) -> crate::storage::cache::KeyCacheSnapshot {
        let readers = self.readers.read().await;
        let table_readers = self.table_readers.read().await;
        let mut seen: std::collections::HashSet<*const reader::SSTableReader> =
            std::collections::HashSet::new();
        let mut agg = crate::storage::cache::KeyCacheSnapshot::default();
        for reader in readers.values().chain(table_readers.values().flatten()) {
            if !seen.insert(Arc::as_ptr(reader)) {
                continue; // already counted this physical reader
            }
            let s = reader.key_offset_cache.snapshot();
            agg.hits = agg.hits.saturating_add(s.hits);
            agg.misses = agg.misses.saturating_add(s.misses);
            agg.evictions = agg.evictions.saturating_add(s.evictions);
            agg.resident_bytes = agg.resident_bytes.saturating_add(s.resident_bytes);
            agg.capacity_bytes = agg.capacity_bytes.saturating_add(s.capacity_bytes);
        }
        agg
    }

    /// Create a new SSTable manager
    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),
        };

        // Load existing SSTable files
        manager.load_existing_sstables().await?;

        Ok(manager)
    }

    /// Create a new SSTable manager from pre-discovered table directories
    ///
    /// This method accepts a list of table directory paths (from DiscoveryService)
    /// and loads SSTables from those specific directories. It does not perform
    /// filesystem scanning beyond the provided directories - this avoids duplicate
    /// scanning when integrating with the discovery/engine lifecycle.
    ///
    /// Use this method when you have pre-discovered table directories and want
    /// to avoid redundant filesystem scanning. Use `new()` when you want automatic
    /// discovery from a single base directory.
    ///
    /// # Arguments
    ///
    /// * `storage_path` - Base storage path (used for context, not for scanning)
    /// * `table_dirs` - List of table directory paths from DiscoveryService
    ///   (e.g., `/data/keyspace1/table1-abc123`)
    /// * `config` - Configuration
    /// * `platform` - Platform abstraction
    ///
    /// # Returns
    ///
    /// A new `SSTableManager` with SSTables loaded from the specified directories
    ///
    /// # Errors
    ///
    /// Returns an error if any of the specified directories cannot be read.
    /// Individual SSTable loading errors are logged but do not fail the entire operation.
    ///
    /// # Example
    ///
    /// ```no_run
    /// use cqlite_core::storage::sstable::SSTableManager;
    /// use cqlite_core::{Config, Platform};
    /// use std::sync::Arc;
    /// use std::path::PathBuf;
    ///
    /// # async fn example() -> cqlite_core::Result<()> {
    /// let config = Config::default();
    /// let platform = Arc::new(Platform::new(&config).await?);
    ///
    /// // Get table directories from DiscoveryService
    /// let table_dirs = vec![
    ///     PathBuf::from("/data/keyspace1/table1-abc123"),
    ///     PathBuf::from("/data/keyspace1/table2-def456"),
    /// ];
    ///
    /// let manager = SSTableManager::new_from_discovered_paths(
    ///     &PathBuf::from("/data"),
    ///     table_dirs,
    ///     &config,
    ///     platform,
    ///     #[cfg(feature = "state_machine")]
    ///     None,
    /// ).await?;
    /// # Ok(())
    /// # }
    /// ```
    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),
        };

        // Load SSTables from the provided table directories
        manager.load_from_table_directories(table_dirs).await?;

        Ok(manager)
    }

    /// Load SSTable readers from specific table directories
    ///
    /// This method scans each provided table directory for Data.db files and loads them.
    /// It handles empty directories gracefully and logs warnings for individual file errors.
    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 {
            // Check if directory exists
            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);

            // Read directory contents
            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;
                }
            };

            // Scan for Data.db files
            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()) {
                    // Check for Cassandra SSTable data files using the *-Data.db pattern.
                    // Skip macOS AppleDouble sidecars via is_apple_double_sidecar().
                    // See Issue #481.
                    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);
                        // Open + wire registries via the shared helper so refresh
                        // opens readers identically (issue #1749). A per-file open
                        // error is logged and skipped here (best-effort load).
                        match self.open_reader_with_schema(&path).await {
                            Ok(reader_arc) => {
                                tracing::debug!(
                                    "SSTableManager successfully loaded SSTable: {}",
                                    sstable_id.0
                                );

                                // Store by SSTableId (existing)
                                readers.insert(sstable_id, reader_arc.clone());

                                // Fully-qualified "keyspace.table" key (or unqualified
                                // fallback) via the shared keying helper (Issue #680).
                                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) => {
                                // Log warning but continue loading other SSTables
                                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(())
    }

    /// Load existing SSTable files from disk
    ///
    /// Scans the base path recursively (up to 3 levels deep) to find Data.db files.
    /// This supports both flat layouts (Data.db directly in base_path) and Cassandra-style
    /// directory structures (keyspace/table_name/Data.db).
    async fn load_existing_sstables(&self) -> Result<()> {
        // Check if directory exists first
        if !self.platform.fs().exists(&self.base_path).await? {
            return Ok(()); // No directory, no SSTables to load
        }

        // Collect all Data.db paths by walking up to 3 levels deep
        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;

        // Pre-compute for the table name fallback heuristic
        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);
            // Open + wire registries via the shared helper so refresh opens
            // readers identically (issue #1749). Don't fail the whole load if one
            // file is problematic — skip it (best-effort initial load).
            match self.open_reader_with_schema(&path).await {
                Ok(reader_arc) => {
                    // Store by SSTableId
                    readers.insert(sstable_id, reader_arc.clone());

                    // Fully-qualified "keyspace.table" key (base-dir-excluded, with
                    // header fallback) via the shared keying helper (Issue #680).
                    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(_) => {
                    // Skip problematic SSTable files during initialization
                    tracing::warn!("Could not load SSTable file: {:?}", path);
                }
            }
        }

        Ok(())
    }

    /// Recursively find all *-Data.db files up to `max_depth` levels deep
    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()) {
                    // Skip macOS AppleDouble sidecars via is_apple_double_sidecar().
                    // See Issue #481.
                    if filename.ends_with("-Data.db") && !is_apple_double_sidecar(filename) {
                        results.push(path);
                    } else if max_depth > 0 {
                        // Check if it's a directory and recurse
                        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)
        })
    }

    /// Create a new SSTable from MemTable data
    ///
    /// NOTE: SSTable writing removed in Issue #176 (writer.rs deleted).
    /// This method is feature-gated behind 'experimental' but currently unimplemented.
    #[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",
        ))
    }

    /// Get a value by key from all SSTables with proper tombstone merging
    #[cfg(feature = "tombstones")]
    pub async fn get(&self, table_id: &TableId, key: &RowKey) -> Result<Option<ScanRow>> {
        // Resolve the applicable reader list FIRST, exactly like the non-tombstones
        // `get()` path (issue #1321). The previous code iterated EVERY reader in
        // `self.readers` and passed one global relaxed `fully_qualified_match` flag
        // to all of them, so same-named tables in OTHER keyspaces passed the relaxed
        // BTI guard and wrongly contributed values/tombstones to the merge — a
        // cross-keyspace data-bleed bug. `resolve_reader_list` returns precisely the
        // readers for the resolved target table across generations, so the relaxed
        // guard can only ever apply to the readers that ARE the target table; a
        // wrong-keyspace same-named reader is never in the merge set.
        //
        // Issue #1591: snapshot the resolved readers + the authoritative
        // `fully_qualified_match` signal and DROP the read guard before any I/O.
        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();

        // Collect each applicable generation's value (tombstone-merge semantics are
        // unchanged: still build a `GenerationValue` per reader and resolve via
        // `TombstoneMerger::merge_generations`). Only the SET of readers being merged
        // changed — the resolved list instead of every reader globally.
        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, // Would be extracted from SSTable metadata
                    },
                };
                all_values.push(gen_value);
            }
        }

        // Use tombstone merger to resolve conflicts across generations
        let merger = TombstoneMerger::new();
        merger.merge_generations(all_values)
    }

    /// Get a value by key from all SSTables (simple version without tombstone merging)
    ///
    /// Uses `table_readers` (keyed by fully-qualified `"keyspace.table"`) so that only the
    /// SSTables for the requested table are searched (Issue #680).  Same-named tables in
    /// different keyspaces (e.g. `test_basic.simple_table` and `test_oa.simple_table`) are
    /// now correctly distinguished.
    ///
    /// Lookup order:
    ///   1. Exact match on the full `table_id` string (e.g. `"test_basic.simple_table"`)
    ///   2. Unqualified table name (e.g. `"simple_table"`) — for backward compatibility
    ///      with flat/non-Cassandra directory layouts that have no keyspace parent.
    #[cfg(not(feature = "tombstones"))]
    pub async fn get(&self, table_id: &TableId, key: &RowKey) -> Result<Option<ScanRow>> {
        // Issue #1591: snapshot the resolved readers + the authoritative
        // `fully_qualified_match` signal and DROP the read guard before any I/O,
        // so a queued writer never FIFO-parks this point read behind a slow scan.
        //
        // `fully_qualified_match`: did resolution match the FULLY-QUALIFIED
        // `keyspace.table` key exactly, or fall back to the bare table name? An
        // unqualified query is treated as an exact match (no keyspace to mismatch).
        // This authoritative signal gates the get() point-lookup table-consistency
        // guard exactly like the seek path (#1284): only an exact FQ match may relax
        // to a name-only check across a header-keyspace divergence; a fully-qualified
        // query resolved via the bare-name fallback keeps strict keyspace matching so
        // get() never returns another keyspace's same-named rows (issue #1321).
        let (reader_list, fully_qualified_match) = self.resolve_reader_snapshot(table_id).await;

        // Return the first value found across all SSTables for this table
        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)
    }

    /// Scan a range of keys from all SSTables for a table.
    ///
    /// # Arguments
    /// * `table_id` - The table to scan
    /// * `start_key` - Optional start key for range scan
    /// * `end_key` - Optional end key for range scan
    /// * `limit` - Optional limit on number of results
    /// * `schema` - Optional table schema for schema-aware parsing. When provided,
    ///   enables accurate type detection and avoids heuristic-based parsing.
    ///   Strongly recommended for Cassandra 5.0+ formats.
    ///
    /// Cross-generation reconciliation (last-write-wins + tombstone shadowing) is
    /// applied via the authoritative k-way merger when more than one SSTable
    /// generation backs the table and `write-support` + a schema are available;
    /// otherwise rows from each reader are concatenated. That concat fallback is
    /// the documented multi-generation limitation (Issue #883) and is now
    /// IDENTICAL across every feature build: the `tombstones` build takes exactly
    /// this path too (it no longer runs its own partition-keyed merge). So no
    /// build regresses relative to the default — a `tombstones`-without-
    /// `write-support` multi-generation read behaves the same as the default
    /// `not(tombstones)`-without-`write-support` build, and the prior `tombstones`
    /// "merge" it replaces was the row-collapsing bug, not real reconciliation.
    ///
    /// Issue #1085: this is the SINGLE `scan` implementation for every feature
    /// build. The former `#[cfg(feature = "tombstones")]` variant grouped per-row
    /// results into a `HashMap` keyed on `RowKey` (which carries only the
    /// partition-key bytes, no clustering) and ran `TombstoneMerger`, so it
    /// collapsed all clustering rows of a partition into one — a full `SELECT *`
    /// over a clustered table returned ~one row per partition. Concatenating
    /// per-reader rows here (and reconciling only ACROSS generations) is correct
    /// for clustered tables in every build.
    ///
    /// Lookup order (Issue #680):
    ///   1. Exact match on the full `table_id` string (e.g. `"test_basic.simple_table"`)
    ///   2. Unqualified table name (e.g. `"simple_table"`) — for backward compatibility
    ///      with flat/non-Cassandra directory layouts that have no keyspace parent.
    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);

        // Issue #1591: snapshot the reader list and DROP the read guard before any
        // I/O. Holding it across the whole scan let one queued writer FIFO-park
        // every later point read behind the slowest in-flight scan.
        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
        );

        // Issue #883: when a table directory holds more than one SSTable
        // generation, plain concatenation of each reader's live rows is wrong —
        // it duplicates rows that exist in several generations and resurrects
        // rows deleted in a later generation (each reader suppresses only its
        // OWN tombstones). Reconcile across generations with the same
        // last-write-wins + tombstone-shadowing rule compaction uses, reusing
        // the authoritative k-way merger (write-support only; requires schema).
        #[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) => {
                        // Never fail a read because the merge path hit an
                        // unsupported format; fall back to concatenation.
                        tracing::warn!(
                            "SSTableManager::scan - cross-generation merge failed for '{}' ({}); \
                             falling back to per-reader concatenation",
                            table_id,
                            e
                        );
                    }
                }
            }
        }

        // Each reader returns rows already in Cassandra token order; k-way
        // merge the per-reader streams in token order (issue #1580) instead of
        // concatenating + re-sorting by RAW key bytes (wrong global order +
        // O(n log n) on the async worker). The merge is O(n log k) and
        // early-exits once `limit` is satisfied.
        let mut per_reader = Vec::with_capacity(reader_list.len());
        for reader in &reader_list {
            #[cfg(test)]
            {
                // Issue #1591: deterministic pause during the scan's I/O. On the
                // fixed path the `table_readers` read guard is already dropped
                // here (we operate on a cloned snapshot), so a test can prove no
                // guard is held across the scan.
                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)
    }

    /// Partition-targeted scan: return only the rows for a single partition key,
    /// touching only the SSTables whose bloom filter / BTI trie admit the key.
    ///
    /// This is the storage-layer fast path for a fully-constrained `WHERE pk = ?`
    /// (Issue #949). Rather than scanning every SSTable for the table and filtering
    /// in memory, it prunes the reader set with
    /// [`might_contain_partition`](reader::SSTableReader::might_contain_partition)
    /// — an O(1) bloom check for BIG format, an O(log n) trie walk for BTI — and
    /// only parses the surviving candidates. On a table backed by thousands of
    /// SSTables, a single-partition read drops from "open and scan all of them" to
    /// "scan only the handful that can hold the key".
    ///
    /// Output matches filtering the full [`scan`](Self::scan) result down to
    /// `partition_key`: the same per-reader parse and the same cross-generation
    /// reconciliation run, just over the pruned candidate set. Concretely, with
    /// more than one candidate generation this drives the authoritative k-way
    /// merge (write-support, schema present); the single-candidate and concat
    /// fallbacks behave exactly as the corresponding `scan` paths do — including
    /// sharing `scan`'s known multi-generation concat limitation (Issue #883) when
    /// the merge is unavailable. The caller still applies its own predicate
    /// evaluation, so any over-inclusion (e.g. a BTI prefix-collision candidate) is
    /// filtered out downstream.
    ///
    /// Gated on `not(tombstones)` because the bloom/BTI prune fast path it relies
    /// on ([`scan_partition_clustering`](reader::SSTableReader::scan_partition_clustering))
    /// is itself `not(tombstones)`-only. Under `tombstones` the executor falls back
    /// to a full [`scan`](Self::scan) + predicate filter (since #1085, `scan` is the
    /// same correct implementation in both builds, so the fallback is correct — just
    /// without the single-partition prune).
    ///
    /// `partition_key` is the raw on-disk partition-key bytes produced by
    /// [`encode_partition_key_columns`](crate::storage::partition_key_codec::encode_partition_key_columns),
    /// which match the bytes the bloom filter, Index.db/BTI trie, and scan RowKeys
    /// are keyed on.
    ///
    /// Within-SSTable seek (Issue #953): for the SINGLE-candidate case (the common
    /// point-lookup path) this seeks directly to the partition's `Data.db` offset
    /// — resolved via the BTI Partitions.db trie or the BIG `Index.db` — and
    /// decodes ONLY that partition via
    /// [`scan_single_partition_clustering`](reader::SSTableReader::scan_single_partition_clustering),
    /// instead of full-parsing the candidate and retaining one partition. The
    /// decode reuses the scan path's `parse_block_emit`, so its output is
    /// byte-for-byte identical to `scan(...).retain(matches_key)`; when the offset
    /// cannot be resolved authoritatively (no `Index.db` hit, or an unsupported
    /// format) it falls back to the full scan + retain for that candidate. The
    /// MULTI-candidate path is unchanged: it still reconciles via the k-way merge
    /// (or the per-candidate concat fallback), so cross-generation LWW / tombstone
    /// shadowing (#883) is preserved.
    ///
    /// Returns `(rows, engaged)`. On this build `engaged` is always `true`: the
    /// underlying [`scan_partition_clustering`](Self::scan_partition_clustering)
    /// prunes the SSTable set via `might_contain_partition` before decoding, so a
    /// caller may honestly report a partition-targeted access path. The
    /// `tombstones`-build counterpart returns `false` because it has no prune
    /// (Epic #951, honest access paths).
    #[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)> {
        // The clustering-aware path always prunes via the bloom/BTI candidate
        // filter, so the partition-targeted access path is genuinely engaged
        // regardless of whether the within-partition clustering seek narrowed.
        let (rows, _clustering_engaged) = self
            .scan_partition_clustering(table_id, partition_key, None, schema)
            .await?;
        Ok((rows, true))
    }

    /// Metadata-carrying partition-targeted scan (Issue #962, Epic #951).
    ///
    /// The WRITETIME/TTL-projection sibling of [`scan_partition`](Self::scan_partition):
    /// it returns only the rows for the single partition identified by the raw
    /// `partition_key` bytes, WITH per-cell write metadata
    /// ([`CellWriteMetadata`] — write timestamp / TTL), while still PRUNING the
    /// SSTable set down to the candidates whose bloom filter / BTI trie admit the
    /// key. A `SELECT WRITETIME(col), TTL(col) ... WHERE pk = ?` therefore opens
    /// only the handful of SSTables that can hold the partition, never all N — the
    /// SSTable-level prune is the must-have that distinguishes this from the
    /// full-table [`scan_with_cell_metadata`](Self::scan_with_cell_metadata).
    ///
    /// Output is identical to filtering `scan_with_cell_metadata(table, ..)` down to
    /// `partition_key`: the same per-reader metadata decode and the same
    /// cross-generation reconciliation run, just over the pruned candidate set, so
    /// the caller's post-scan predicate evaluation is a pure correctness backstop
    /// (it removes any bloom/BTI false-positive over-inclusion).
    ///
    /// Reconciliation mirrors `scan_partition`:
    /// - More than one candidate generation (write-support + schema): drive the
    ///   authoritative k-way merge via
    ///   `generation_merge::merge_generations_for_read_with_metadata` TARGETED to
    ///   just this partition. This preserves
    ///   per-cell cross-generation LWW / tombstone shadowing for WRITETIME/TTL
    ///   (Issue #885) on the targeted path.
    /// - Otherwise: decode each candidate via the reader's metadata path and retain
    ///   this partition's rows, concatenating across candidates.
    ///
    /// Within-SSTable decode currently full-decodes each surviving candidate's
    /// metadata and retains the partition; the SSTable-level prune (avoiding the
    /// full TABLE/SSTable scan) is the property #962 requires. A within-partition
    /// metadata seek (bounding the decode to the partition's `Data.db` offset, as
    /// `scan_single_partition_clustering` does for the plain path) is a documented
    /// follow-up.
    ///
    /// Gated on `not(tombstones)` to match the `scan_partition` variant it parallels.
    ///
    /// Returns `(rows, engaged)`. On this build `engaged` is always `true`: the
    /// candidate set is pruned via `might_contain_partition` before any decode, so
    /// the partition-targeted metadata access path is genuinely engaged. The
    /// `tombstones`-build counterpart returns `false` (no prune; full metadata
    /// scan + retain) so the caller reports an honest fallback (Epic #951).
    /// Prune a reader snapshot to the candidate generations that admit
    /// `partition_key`, hoisting the BTI key hash+encoding to ONCE per read instead
    /// of once per candidate (issue #1575 / C4).
    ///
    /// When ANY candidate is a BTI ("da") reader, the Murmur3-based byte-comparable
    /// trie key is computed a SINGLE time and reused for every candidate's trie
    /// prune (`might_contain_partition_encoded`); on `main` each candidate re-encoded
    /// the same key, so an N-generation fan-out paid N identical Murmur3 hashes
    /// (`KEY_HASH_CALLS == N`) where C4 pays 1. A BIG (`nb`) candidate has no BTI
    /// encoding to hoist — its raw-key bloom check runs unchanged — so a non-BTI or
    /// mixed candidate set stays correct. The pruning decision (and therefore the
    /// resulting rows) is byte-identical to the per-candidate path.
    /// Returns `(admitted, pruned)`: `admitted` are the SAME byte-for-byte
    /// filtered candidates the original single-`Vec` `prune_candidates` returned;
    /// `pruned` are every reader EXCLUDED by the presence-oracle check — each is,
    /// by construction, a definitive-negative exclusion (the filter's ONLY
    /// criterion), so `pruned` is exactly the set
    /// [`verify_pruned_candidates`](Self::verify_pruned_candidates) needs to
    /// authoritatively re-check (issue #2163's core silent-miss case: a false
    /// negative HERE, at the multi-generation candidate prune, would otherwise
    /// drop that SSTable from the read with no verification ever reached — the
    /// per-reader `get_with_resolution` verify hook only covers a single reader's
    /// OWN point read, never a candidate eliminated at this layer).
    #[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;
        // Encode once iff a BTI reader is present (BIG readers ignore `encoded`).
        let encoded = readers
            .iter()
            .any(|r| r.is_bti())
            .then(|| encode_partition_key_for_bti_trie(partition_key));
        readers.iter().cloned().partition(|r| {
            // BTI candidates prune via `might_contain_partition_encoded` (fed the
            // hoisted C4 encoding), the SAME single emit site the raw-key BIG
            // bloom prune below uses (issue #2163): the `Partitions.db` trie is
            // the authoritative presence oracle, so a `false` (trie miss) is a
            // definitive absent (drop) — recorded once as
            // `cqlite.read.sstables_pruned{format=bti}` — and an `Err` (corrupt
            // trie) is conservatively kept (not pruned, not recorded). This is
            // congruent with the façade's resolution — a BTI trie result is
            // authoritative. BIG candidates DELIBERATELY stay on the bloom-based
            // `might_contain_partition`: a BIG `Index.db` miss is NOT a definitive
            // absent (#1572 truncated-index invariant), so pruning a BIG reader on
            // the façade's index probe would drop a candidate that actually holds
            // the partition. Only the bloom filter never yields a false negative.
            // Sharing ONE emit site (`might_contain_partition[_encoded]`) for both
            // formats means a caller that also `locate`s the same key afterward
            // (the actual read) never double-emits: `locate`/`locate_encoded`
            // themselves never call `emit_sstable_pruned`.
            match (r.is_bti(), &encoded) {
                (true, Some(enc)) => r.might_contain_partition_encoded(partition_key, enc),
                // Non-BTI reader (or, defensively, no encoding): bloom prune.
                _ => r.might_contain_partition(partition_key),
            }
        })
    }

    /// Verify presence-oracle negatives for SSTables EXCLUDED by
    /// [`prune_candidates`](Self::prune_candidates) — issue #2163's core
    /// silent-miss case, roborev r6. A bloom/BTI-trie false negative during a
    /// MULTI-generation candidate prune would otherwise drop that SSTable from
    /// the read entirely, unverified: the per-reader `get_with_resolution` opt-in
    /// verify hook (`data_access/mod.rs`) never runs for a candidate eliminated
    /// one layer up, at THIS prune site — a candidate excluded here never reaches
    /// `get_with_resolution` at all for this read.
    ///
    /// STRICTLY inside the opt-in switch (default-off = zero extra work,
    /// unchanged behaviour): when
    /// [`presence_verification::enabled`](reader::presence_verification::enabled)
    /// is `false` (the default), this returns immediately without touching
    /// `pruned` at all. When enabled, each pruned reader is verified via
    /// [`SSTableReader::verify_presence_oracle_negative`] (an authoritative
    /// confirmation scan): a genuine contradiction increments
    /// `cqlite.read.bloom.false_negatives` EXACTLY ONCE per contradicted
    /// negative (the verify method's own single per-call emit — this loop calls
    /// it once per pruned reader, never more). Any `Err` fails OPEN (the
    /// caller's admitted `candidates` list is never touched) but is surfaced
    /// LOUDLY via the SAME loud-failure contract `get_with_resolution` uses
    /// (roborev r4): an error-level log with context, and a record through the
    /// EXISTING error-rate signal (`cqlite.errors.total{subsystem=reader}`,
    /// issue #1038) — never a new metric.
    #[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,
    )> {
        // Issue #1591: snapshot the reader list and DROP the read guard before any
        // I/O (bloom/BTI prune, per-candidate decode, cross-generation merge).
        let (reader_list, _fully_qualified_match) = self.resolve_reader_snapshot(table_id).await;
        if reader_list.is_empty() {
            return Ok((Vec::new(), true));
        }

        // Prune: keep only SSTables whose bloom filter / BTI trie admit the key.
        // This is the property #962 requires — only candidates are opened, never N.
        // C4 (#1575): the BTI key hash+encoding is hoisted to once per read here.
        let (candidates, pruned) = Self::prune_candidates(&reader_list, partition_key);

        // Issue #2163 (roborev r6): opt-in verification of every EXCLUDED
        // candidate — a strict no-op unless the switch is on. Runs BEFORE the
        // `candidates.is_empty()` early-return below, since the core silent-miss
        // case is exactly "every admitted candidate came up empty because a
        // false negative wrongly pruned the ONE generation holding the 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
        };

        // Multiple candidate generations may hold the same partition; reconcile
        // with the same authoritative metadata-aware k-way merge the full metadata
        // scan uses (write-support only, schema present), TARGETED to just this
        // partition (issue #1579): the merge keeps only `partition_key`'s rows and
        // stops as soon as it finds them, byte-identical to the former
        // full-merge-then-`retain(matches_key)` but without materializing every
        // other partition. This preserves per-cell cross-generation WRITETIME/TTL.
        #[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-counter gate (Issue #958): the merge parsed every
                        // surviving candidate; `merged` (already the target partition
                        // only) is exactly the partitions this lookup returns.
                        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
                        );
                    }
                }
            }
        }

        // Single candidate (common case) or the multi-candidate concat fallback:
        // decode each candidate's metadata and retain this partition's rows.
        let mut all_results = Vec::new();
        for reader in &candidates {
            // Work-counter gate (Issue #958): one real Data.db touch per surviving
            // candidate. Counted here (not at prune time) so the counter reflects
            // SSTables actually opened/scanned.
            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);
        }
        // #1917: rows here already share `partition_key` (prior retain) so this is a no-op today; kept for future multi-partition safety + parity with the IN fan-out / streaming scan's token order (raw-byte = #1580 wrong-order).
        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))
    }

    /// `tombstones`-build counterpart of
    /// [`scan_partition_with_cell_metadata`](Self::scan_partition_with_cell_metadata).
    ///
    /// That build has no bloom-prune metadata path, so a fully-constrained
    /// `WHERE pk = ?` WRITETIME/TTL read is served by scanning with metadata and
    /// filtering to the partition key, matching the `not(tombstones)` output while
    /// keeping the query executor free of `tombstones` cfg branching.
    ///
    /// Returns `(rows, engaged)` with `engaged == false`: this is a full metadata
    /// scan + retain with NO SSTable prune, so the caller MUST report an honest
    /// fallback access path (`FallbackReason::TombstonesBuildNoPrune`) rather than a
    /// targeted label, even though the rows are byte-identical to the pruned build
    /// (Epic #951, honest access paths).
    #[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))
    }

    /// Clustering-slice-aware partition-targeted scan (Issue #954, Epic #951).
    ///
    /// Identical to [`scan_partition`](Self::scan_partition) but, when `clustering`
    /// is `Some(slice)` AND exactly one candidate SSTable admits the key AND that
    /// candidate's single-partition seek can use its authoritative row index, the
    /// within-partition decode is bounded to the row-index block(s) covering the
    /// requested clustering range — so a `WHERE pk = ? AND ck </>/= ?` slice over a
    /// wide partition decodes O(matched rows + index block), not the whole
    /// partition.
    ///
    /// Returns `(rows, clustering_seek_engaged)`. `clustering_seek_engaged` is
    /// `true` only when the within-partition clustering narrowing actually bounded
    /// the decode (so the caller may report
    /// [`AccessPath::ClusteringSlice`](crate::query::access_path::AccessPath::ClusteringSlice));
    /// it is `false` for the multi-candidate / merge / full-decode fallbacks,
    /// which still return correct rows for the honest `PartitionLookup` path. The
    /// rows are ALWAYS the full partition (or its clustering-narrowed superset):
    /// the caller's post-scan `evaluate_leaf` applies the exact clustering bound,
    /// so output is byte-identical regardless of whether the seek engaged.
    #[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)> {
        // Issue #1591: snapshot the reader list + the authoritative
        // `fully_qualified_match` signal and DROP the read guard before any I/O.
        //
        // `fully_qualified_match`: did resolution match the FULLY-QUALIFIED
        // `keyspace.table` key exactly, or fall back to the bare table name? An
        // unqualified query is treated as an exact match (no keyspace to mismatch).
        // This authoritative signal gates the seek's table-consistency guard: only
        // an exact FQ match may relax to a name-only check across a header-keyspace
        // divergence; a fully-qualified query resolved via the bare-name fallback
        // keeps strict keyspace matching so it never returns another keyspace's
        // same-named rows (#1284 review).
        let (reader_list, fully_qualified_match) = self.resolve_reader_snapshot(table_id).await;
        if reader_list.is_empty() {
            return Ok((Vec::new(), false));
        }

        // Prune: keep only SSTables whose bloom filter / BTI trie admit the key.
        // C4 (#1575): the BTI key hash+encoding is hoisted to once per read here.
        let (candidates, pruned) = Self::prune_candidates(&reader_list, partition_key);

        // Issue #2163 (roborev r6): opt-in verification of every EXCLUDED
        // candidate — a strict no-op unless the switch is on. See
        // `verify_pruned_candidates` for the fail-open / loud-failure contract.
        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;

        // Multiple candidate generations may hold the same partition; reconcile
        // with the same authoritative k-way merge the full scan uses (write-support
        // only), TARGETED to just this partition (issue #1579): the merge keeps only
        // `partition_key`'s rows and stops as soon as it finds them, byte-identical
        // to the former full-merge-then-`retain(matches_key)` but without
        // materializing every other partition.
        #[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(
                    &candidates,
                    schema,
                    None,
                    None,
                    None,
                    Some(&target),
                )
                .await
                {
                    Ok(merged) => {
                        // Work-counter gate (Issue #958): the k-way merge parsed
                        // every surviving candidate, and `merged` (already the target
                        // partition only) is exactly the partitions this lookup returns.
                        work_counters::add_sstables_scanned(candidates.len() as u64);
                        work_counters::add_partitions_parsed(merged.len() as u64);
                        // The cross-generation merge decodes full partitions; the
                        // clustering seek does not engage here (#954). Correct rows
                        // via the post-scan backstop; honest non-engaged signal.
                        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
                        );
                    }
                }
            }
        }

        // Single candidate (the common case): SEEK directly to the partition's
        // Data.db offset and decode ONLY that partition (Issue #953), instead of a
        // full parse-then-retain. The seek resolves the offset via the BTI trie /
        // Index.db and runs the same per-partition decode the scan path uses, so
        // its output is byte-for-byte identical to `scan(...).retain(matches_key)`.
        // If the seek is not applicable for this reader (no authoritative offset,
        // or an unsupported format), it returns `Ok(None)` and we FALL BACK to the
        // full scan + retain for that candidate (Constraint #4: correctness over
        // optimization). The multi-candidate concat fallback below is unchanged —
        // only the single-candidate path gets the seek.
        let mut all_results = Vec::new();
        let mut clustering_engaged = false;
        for reader in &candidates {
            // Work-counter gate (Issue #958): one real Data.db touch per surviving
            // candidate. Counted here (not at prune time) so the counter reflects
            // SSTables actually opened/scanned, the cost a regression would balloon.
            work_counters::add_sstables_scanned(1);

            let mut results = if candidates.len() == 1 {
                // Issue #954: thread the clustering slice into the seek so it can
                // narrow the within-partition decode via the authoritative row
                // index. `engaged` records whether the clustering narrowing
                // actually bounded the decode (vs a full-partition decode).
                match reader
                    .scan_single_partition_clustering(
                        table_id,
                        partition_key,
                        clustering,
                        fully_qualified_match,
                        schema,
                    )
                    .await
                {
                    // Seek resolved authoritatively: use its rows directly. They
                    // already match exactly this partition's key, so no retain.
                    Ok(Some((rows, engaged))) => {
                        clustering_engaged = engaged;
                        rows
                    }
                    // Seek not applicable (Constraint #4): full scan + retain.
                    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 {
                // Multi-candidate concat fallback (merge unavailable): preserve the
                // existing full-scan + retain behaviour per candidate (Constraint #2).
                let mut r = reader.scan(table_id, None, None, None, schema).await?;
                r.retain(matches_key);
                r
            };
            all_results.append(&mut results);
        }
        // #1917: rows here already share `partition_key` (prior retain) so this is a no-op today; kept for future multi-partition safety + parity with the IN fan-out / streaming scan's token order (raw-byte = #1580 wrong-order).
        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))
    }

    /// `tombstones`-build counterpart of [`scan_partition`](Self::scan_partition).
    ///
    /// That build uses a structurally different reader map and has no bloom-prune
    /// `scan_partition` path, so a fully-constrained `WHERE pk = ?` is served by
    /// scanning and filtering to the partition key. The output is a subset of
    /// [`scan`](Self::scan) — identical to what the `not(tombstones)`
    /// `scan_partition` returns — which keeps the query executor free of any
    /// `tombstones` cfg branching.
    ///
    /// Returns `(rows, engaged)` with `engaged == false`: this is a full scan +
    /// retain with NO SSTable prune, so the caller MUST report an honest fallback
    /// access path (`FallbackReason::TombstonesBuildNoPrune`) rather than a targeted
    /// label, even though the rows match the pruned build byte-for-byte (Epic #951).
    #[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))
    }

    /// Resolve the reader list for a table id, trying the fully-qualified
    /// `keyspace.table` name first and falling back to the bare table name, so
    /// same-named tables in different keyspaces stay distinct (Issue #680).
    ///
    /// Shared by [`get`](Self::get), [`scan`](Self::scan), and
    /// [`scan_partition`](Self::scan_partition) so the resolution rule lives in
    /// one place and the targeted-lookup path can never drift from `scan`.
    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)
    }

    /// Authoritative resolution-mode signal that gates the BTI point-lookup
    /// table-consistency guard (issue #1321, mirroring the seek path #1284).
    ///
    /// Returns `true` iff the queried `table_name` matched the fully-qualified
    /// `table_readers` map EXACTLY (or is unqualified, so has no keyspace to
    /// mismatch), and `false` iff a fully-qualified `keyspace.table` query can
    /// only have reached a reader via the bare-name fallback. Only an exact FQ
    /// match may relax across a benign header-keyspace divergence; a fallback
    /// keeps strict keyspace matching so `get()` never surfaces another
    /// keyspace's same-named rows.
    ///
    /// Shared verbatim by BOTH `get()` builds (the `tombstones` and the default
    /// `not(tombstones)` managers) so the relaxation is identical in every
    /// feature build — the single source of truth for the wiring.
    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)
    }

    /// Scan a table and return per-cell write metadata alongside row values.
    ///
    /// Used when `ProjectionFlags::include_cell_metadata` is set (issue #693 — the
    /// WRITETIME/TTL threading bridge).  Delegates to each reader's
    /// `scan_with_cell_metadata`.  When multiple readers serve the same table the
    /// results are concatenated; token-order sort and LIMIT are applied afterward.
    ///
    /// Falls back to the regular `scan` with empty metadata when the reader does not
    /// surface metadata (non-V5CompressedLegacy paths).
    ///
    /// Issue #1535: this is the single implementation for both the default /
    /// `write-support` build and the `tombstones` build. The former `tombstones`
    /// variant delegated to `scan` and returned empty metadata, so `WRITETIME(col)`
    /// / `TTL(col)` wrongly resolved to null under `--features tombstones`. The
    /// reader-level `scan_with_cell_metadata` surfaces the authoritative per-cell
    /// timestamp/TTL regardless of feature flags, so the fix is simply to use it in
    /// both builds. The cross-generation metadata merge stays gated on `write-support`
    /// (it needs the `KWayMerger`); without it the per-reader concatenation still
    /// surfaces each cell's real metadata rather than fabricating or dropping it.
    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>)>> {
        // Issue #1591: snapshot the reader list and DROP the read guard before any
        // I/O (per-reader metadata decode and cross-generation merge).
        let (reader_list, _fully_qualified_match) = self.resolve_reader_snapshot(table_id).await;
        if reader_list.is_empty() {
            return Ok(Vec::new());
        }

        // Issue #885: the metadata path (WRITETIME/TTL projection) must
        // reconcile across SSTable generations exactly like the plain `scan`
        // path (#883) — otherwise a multi-generation directory returns
        // duplicate rows and resurrects rows/cells deleted in a later
        // generation. Drive the same authoritative k-way merger, then surface
        // the WINNING cell's per-cell write timestamp / TTL (write-support
        // only; requires schema). Single-generation reads skip this entirely.
        #[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) => {
                        // Never fail a read because the merge path hit an
                        // unsupported format; fall back to concatenation.
                        tracing::warn!(
                            "SSTableManager::scan_with_cell_metadata - cross-generation merge \
                             failed for '{}' ({}); falling back to per-reader concatenation",
                            table_id,
                            e
                        );
                    }
                }
            }
        }

        // K-way merge the per-reader token-ordered streams (issue #1580),
        // mirroring `scan`: the metadata payload rides alongside the row.
        // Merging in token order (not raw key bytes) matches Cassandra's
        // cross-SSTable order and avoids the full O(n log n) re-sort.
        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)
    }

    /// Snapshot the resolved reader set for `table_id` and DROP the
    /// `table_readers` read guard before returning (issue #1591).
    ///
    /// Returns the cloned `Arc<SSTableReader>` handles serving the table plus the
    /// authoritative `fully_qualified_match` signal (see
    /// [`fully_qualified_match`](Self::fully_qualified_match)), acquiring the read
    /// guard only long enough to clone the small `Vec` of `Arc`s.
    ///
    /// # Why (tail-latency fix)
    ///
    /// `tokio::sync::RwLock` is FIFO-fair: a single queued writer (reader reload,
    /// schema set, generation removal) parks EVERY later-arriving reader behind
    /// the longest in-flight read guard. Holding the guard across a whole
    /// multi-reader scan therefore made one slow scan plus one admin write stall
    /// every subsequent point read — bimodal tail latency. Cloning the `Arc` list
    /// and releasing the guard immediately means scans and point reads run without
    /// holding the lock across their I/O, so a queued writer can never park them.
    ///
    /// # Semantics (unchanged)
    ///
    /// A scan operating on this snapshot may miss a reader added to the map AFTER
    /// the snapshot was taken. That is the SAME semantics as holding the guard:
    /// the guard would have blocked the writer until the scan finished, so the
    /// scan never observed the new reader either. Readers removed from the map
    /// mid-scan stay alive for the snapshot holder — that is precisely what the
    /// `Arc` clone guarantees.
    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();
        // Guard dropped here, before any reader I/O.
        (readers, fully_qualified_match)
    }

    /// Resolve the AUTHORITATIVE partition-key shape for `table_id` from the
    /// SSTable readers' Statistics.db SerializationHeader (issue #1750).
    ///
    /// This is the metadata a schema-less reader DOES have without a CQL schema:
    /// the SerializationHeader (surfaced on each reader's `header().columns`)
    /// records, per column, the authoritative `is_primary_key` / `is_clustering`
    /// flags and the REAL names of the regular + static (non-key) columns. It does
    /// NOT record the partition-/clustering-key column NAMES — Cassandra never
    /// serialises those (they live in `system_schema`, absent schema-less), so the
    /// parser synthesises a placeholder pk name (`build_partition_key_columns`) that
    /// MUST NOT be trusted as an identity.
    ///
    /// Returns [`PartitionKeyShape`] carrying ONLY authoritative facts: the count of
    /// partition-key components, the count of clustering keys, and the set of real
    /// non-key column names. The caller confirms a predicate column is the sole
    /// partition key BY ELIMINATION (single pk component, zero clustering keys, and
    /// the column absent from the non-key name set) — never by the synthesised name.
    ///
    /// The shape is resolved across ALL readers serving the table, not just the
    /// first (issue #1750, roborev 3786). A multi-generation / schema-evolved table
    /// can add a REGULAR column in a LATER generation that is absent from an earlier
    /// generation's header. Consulting only the first reader would MISS that column
    /// from the non-key name set, so a `WHERE <later-gen-regular-col> = <val>` could
    /// be MISCLASSIFIED as a partition-key seek. To prevent that:
    ///   * the non-key column names are UNIONed across every reader's header, and
    ///   * the partition-key count, clustering-key count, and the single-component
    ///     pk type+name must be CONSISTENT across all readers with a header.
    ///
    /// Returns `None` when no reader for the table exposes a SerializationHeader OR
    /// when the readers' key metadata is INCONSISTENT (a pk-count / clustering-count
    /// / single-pk type+name disagreement — which should never happen for one table
    /// but, if it did, means we cannot trust the shape). Both are fail-safe: the
    /// caller then keeps the honest full-scan path (all from SerializationHeaders,
    /// no heuristics — #28).
    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()))
    }

    /// Arm this manager's deterministic scan gate (issue #1591 test only).
    /// The next `scan` on this manager pauses at its per-reader I/O, signalling
    /// [`Gate::reached`](scan_gate::Gate) and blocking on
    /// [`Gate::release`](scan_gate::Gate) until the test lets it continue.
    #[cfg(all(test, feature = "write-support", feature = "state_machine"))] // gated to sole caller issue_1591_scan_lock_test; not dead-code under minimal (#1981)
    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
    }

    /// Resolve the readers serving `table_id`, returning cloned `Arc` handles.
    ///
    /// Mirrors the qualified-then-unqualified lookup of [`scan`](Self::scan)
    /// (Issue #680) but yields owned handles so the caller can hold them past the
    /// `table_readers` read lock — needed by the streaming scan, which spawns a
    /// background merge task.
    #[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()
    }

    /// Reports whether [`scan_stream`](Self::scan_stream) PRE-MATERIALIZES the
    /// full reconciled result for this table before returning the channel, rather
    /// than yielding rows lazily (issue #1577, roborev round-4).
    ///
    /// A bounded LIMIT consumer of `scan_stream` (see
    /// [`SelectExecutor::capped_fallback_scan`](crate::query)) may drop the stream
    /// once `cap` rows arrive to stop the producer decoding the tail — but that
    /// decode-stop win, and the per-received-row `QUERY_ROWS_SCANNED` accounting it
    /// implies, are ONLY valid when the stream is genuinely lazy. When this returns
    /// `true`, the storage layer has already decoded EVERY row of the table (the
    /// channel is just replaying a materialized `Vec`), so the caller must charge
    /// the FULL decoded row count to the scan-work metric and take a materializing
    /// accounting path instead.
    ///
    /// The condition mirrors EXACTLY the materializing branches `scan_stream`
    /// takes, so the two can never disagree:
    /// * the `tombstones` build's `scan_stream` delegates wholesale to the
    ///   materializing [`scan`](Self::scan) — always `true`;
    /// * any BTI (`da`) reader: `run_scan_stream`'s BTI branch (issue #1577) drives
    ///   the trie-walk `bti_scan_with_metadata`, which fully materializes the
    ///   (index-less) reconciled table before streaming — so a bounded consumer
    ///   never decode-stops for a BTI table, regardless of generation count;
    /// * the default (non-`tombstones`) build's `write-support` cross-generation
    ///   branch is LAZY since #1579 (streams via
    ///   `generation_merge::stream_generations_for_read`), so a bounded consumer
    ///   DOES decode-stop there — that branch is NOT reported as materializing.
    #[cfg(feature = "tombstones")]
    pub async fn scan_stream_materializes(
        &self,
        _table_id: &TableId,
        _schema: Option<&crate::schema::TableSchema>,
    ) -> bool {
        // The `tombstones` build's `scan_stream` forwards a fully-materialized
        // `scan` result, so a bounded consumer never decode-stops.
        true
    }

    /// See the `tombstones` variant above.
    #[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;

        // Issue #1577: any BTI (`da`) reader makes `scan_stream` pre-materialize
        // (the trie-walk BTI branch decodes the whole reconciled table before
        // streaming), so a bounded LIMIT consumer must charge the full decoded
        // count — the exact condition `run_scan_stream` gates its BTI branch on
        // (`bti_partitions_db.is_some()`, surfaced as `reader.is_bti()`).
        if readers.iter().any(|r| r.is_bti()) {
            return true;
        }

        // Issue #1579 / D3 (roborev job 3669, Medium): the `not(tombstones)`
        // write-support multi-generation branch NO LONGER pre-materializes. Since
        // #1579 `scan_stream` routes the cross-generation case through the LAZY,
        // backpressure-bounded `generation_merge::stream_generations_for_read`
        // (`KWayMerger` driven on a blocking task, live rows fed STRAIGHT into the
        // bounded channel) — NOT the materializing `merge_generations_for_read`. The
        // merger reconciles every generation for a partition at the moment its key is
        // stepped (LWW + tombstone shadowing applied before emission) and emits
        // partitions in strictly increasing `(token, key)` order, so a bounded LIMIT
        // consumer can decode-stop after the authoritative first `cap` rows. Returning
        // `false` restores D1's decode-stop for multi-gen LIMIT and makes the
        // per-received-row `QUERY_ROWS_SCANNED` accounting MORE accurate (only ~`cap`
        // rows decoded). BTI is still handled above (`true`) — its own branch
        // materializes regardless of generation count.
        false
    }

    /// Streaming scan (issue #790): merge per-SSTable streams lazily into a
    /// bounded output channel, in key (token) order, without materializing the
    /// whole result.
    ///
    /// Each reader yields entries already in token order; a k-way merge over the
    /// per-reader heads produces globally ordered output while holding only one
    /// pending entry per SSTable. Live heap is bounded by `buffer_size` plus the
    /// number of SSTables, independent of total row count — the streaming analog
    /// of the materializing [`scan`](Self::scan) (concat + stable sort by key).
    ///
    /// # Multi-generation correctness (Issue #957)
    ///
    /// The lazy per-reader k-way merge above is only the streaming analog of
    /// `scan`'s **concat + sort** path, which is correct for a single generation.
    /// When a table directory holds more than one SSTable generation, the same
    /// `(partition, clustering)` row can live in several generations and a
    /// row/cell tombstone in a newer generation suppresses only its own
    /// generation's copy — so a pure key-ordered merge would emit overwritten
    /// rows twice and resurrect rows deleted in a later generation. `scan` avoids
    /// this by routing the multi-generation case through the authoritative
    /// `KWayMerger` (the same LWW + tombstone-shadowing k-way merge compaction
    /// uses); this streaming path must reconcile identically or `execute()` and
    /// `execute_streaming()` diverge.
    ///
    /// # Streaming the multi-generation merge (Issue #1579 / D3)
    ///
    /// The multi-generation case runs
    /// [`generation_merge::stream_generations_for_read`], which drives that same
    /// `KWayMerger` on a blocking task and feeds each stepped partition's live
    /// rows STRAIGHT into the bounded channel via `blocking_send` (backpressure
    /// preserved) — instead of collecting the ENTIRE reconciled table, sorting it,
    /// and only then dribbling it (the pre-D3 behaviour). Live heap is therefore
    /// O(one partition + channel), and time-to-first-row is O(first partition),
    /// not O(full merge). The merger already emits partitions in `(token, key)`
    /// order, byte-identical to `scan`'s `sort_by_token_order`, so the emitted
    /// order is unchanged (issue #1579 ordering guardrail). On a merger
    /// CONSTRUCTION error the driver reports back so this path FALLS BACK to the
    /// lazy per-reader streaming merge, mirroring `scan`'s fall-back-to-concat. The
    /// single-generation / no-schema / no-`write-support` cases keep the lazy
    /// streaming merge, which already matches `scan`'s concat path exactly and
    /// preserves LIMIT/backpressure.
    #[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;

        // Issue #957: keep the materializing `scan` and this streaming path
        // definitionally in lockstep. Reuse the EXACT guard `scan` uses for
        // cross-generation reconciliation (`reader_list.len() > 1 && schema present`,
        // write-support only) and the same merger, then forward the reconciled rows
        // through the streaming channel. Without this, a partition spread across
        // generations duplicates overwritten rows and resurrects deleted ones in the
        // stream while `scan` returns the merged, deduplicated, tombstone-honouring
        // result.
        #[cfg(feature = "write-support")]
        if readers.len() > 1 {
            // No schema: cross-generation LWW/tombstone reconciliation needs the
            // schema to drive the KWayMerger, so this deliberately falls through to
            // the lazy per-reader token-merge below — matching `scan`'s identical
            // no-schema fallback (both accept the documented Issue #883 concat
            // limitation rather than reconciling without a schema).
            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) => {
                        // Never fail a read because the merge could not be
                        // constructed (unsupported format); fall back to the lazy
                        // streaming merge, matching `scan`'s fall-back-to-concatenation.
                        //
                        // Error-path asymmetry, intentional (issue #1579): this `Err`
                        // is ONLY the merger-CONSTRUCTION failure (`KWayMerger::new`,
                        // signalled back before any row is streamed), so falling back
                        // here can never emit a partially-merged, mis-reconciled
                        // result — the streaming task has not sent anything yet. A
                        // `step()` error occurring MID-merge (after some partitions
                        // were already streamed to the caller) is NOT retried here;
                        // it is delivered downstream as an `Err` item on the output
                        // channel (see `stream_generations_for_read`), ending the
                        // stream at that point. That is deliberately SAFER than the
                        // materializing `scan`'s fallback, which — if it could fail
                        // partway through populating its `Vec` — would have no way to
                        // signal "these rows are reconciled, the rest are not"; the
                        // streaming channel's `Err` item gives the caller an honest,
                        // unambiguous cutoff instead of silently returning a
                        // half-reconciled table.
                        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));

        // Own everything the background merge task needs.
        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 {
            // Admission control (issue #1594, F4): this fan-out merge is ONE
            // top-level scan OPERATION that legitimately needs all N per-generation
            // sub-scans live AT ONCE (it primes a head from every sub-scan before
            // draining any). Acquire exactly ONE admission permit here, for the
            // whole operation, and open each sub-scan `Exempt` (below) so the
            // sub-scans do NOT each independently admit. If they did, a fan-out to
            // `N > cap` generations would deadlock: `cap` sub-scans would win
            // permits and park in backpressure while the rest blocked forever at
            // `admit`, and this priming loop — waiting on the blocked sub-scans —
            // would never drain the permit-holders. Held via this RAII guard for
            // the whole merge; released on every exit. See `scan_admission` docs.
            let _admission =
                crate::storage::sstable::reader::scan_stream_windowed::scan_admission::admit()
                    .await;

            // Open one streaming scan per reader. Each is `Exempt` — the single
            // permit above covers the whole fan-out operation (issue #1594).
            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();

            // Prime one head per stream. Each head carries its precomputed
            // Cassandra Murmur3 token so the merge orders by (token, key) — the
            // authoritative cross-SSTable order (issue #1580) — and never hashes a
            // key more than once. Comparing by raw `RowKey` bytes here (as this
            // path previously did) diverged from `scan`'s token order.
            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),
                }
            }

            // K-way merge: repeatedly emit the head with the smallest
            // (token, key), ties broken by reader index to match the stable
            // token-order merge of `scan`.
            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, // all streams exhausted
                };

                // Take the winning entry and advance only that stream.
                let entry = match heads[idx].take() {
                    Some((_, key, row)) => (key, row),
                    None => break, // unreachable: min_idx points to a Some head
                };
                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 => {} // stream exhausted; head stays None
                }

                if out_tx.send(Ok(entry)).await.is_err() {
                    return; // consumer dropped
                }
            }
        });

        Ok(out_rx)
    }

    /// Batched streaming scan (issue #1592, Epic F/F2): the additive companion to
    /// [`scan_stream`](Self::scan_stream) whose channel item is a `Vec` BATCH of
    /// entries. It forwards one batch per async wake instead of one row, undoing
    /// the per-row re-flattening on the public channel that the internal windowed
    /// pipeline (issue #1143) was built to avoid.
    ///
    /// Content + order are identical to [`scan_stream`](Self::scan_stream):
    /// flattening the batches reproduces the per-row stream exactly.
    ///
    /// - **Single generation** (one reader): the reader's windowed batches are
    ///   forwarded STRAIGHT THROUGH — no per-row channel is interposed, so the
    ///   wake amortization survives end to end (the F2 win).
    /// - **Zero / multiple generations**: reuses the fully-correct per-row
    ///   [`scan_stream`](Self::scan_stream) (cross-generation reconciliation +
    ///   token-ordered k-way merge + empty case) and re-chunks its output into
    ///   batches for the public channel. A generation-aware fully-batched merge is
    ///   a deliberate follow-up (audit §Epic F); cross-generation correctness wins
    ///   here.
    #[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))
    }

    /// Re-chunk a per-row streaming receiver into `BATCH_EMIT_ROWS`-sized `Vec`
    /// batches over a bounded channel (issue #1592). Preserves order and content
    /// exactly (FIFO push/flush) and preserves backpressure: the batch channel is
    /// bounded, so a stalled consumer stops the drain of the per-row source, which
    /// stops the upstream scan. The trailing partial batch is flushed at end. A
    /// mid-stream error is forwarded as a terminal item.
    ///
    /// Used for the zero/multi-generation and `tombstones` cases, where a
    /// straight-through single-reader hand-off is not applicable (the per-row
    /// source is a k-way merge / materialized reconciliation, not one reader).
    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;
        // Bound the batch channel so its resident-row budget stays comparable to
        // the per-row surface's `buffer_size`, not `buffer_size * 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; // consumer dropped
                            }
                            batch.reserve(BATCH_EMIT_ROWS);
                        }
                    }
                    Err(e) => {
                        // Flush already-received Ok rows BEFORE surfacing the
                        // error, to match the per-row `scan_stream` guarantee
                        // that confirmed rows are delivered ahead of a terminal
                        // error (issue #1143 / #1592). Dropping them here would
                        // silently lose up to BATCH_EMIT_ROWS-1 rows.
                        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
    }

    /// Streaming scan under the `tombstones` feature.
    ///
    /// Streaming the cross-generation tombstone merge is not yet implemented, so
    /// this falls back to the materializing [`scan`](Self::scan) and forwards the
    /// result through a bounded channel. The public API stays uniform across
    /// feature configs; the O(rows) memory win of issue #790 applies only to the
    /// default (non-`tombstones`) build.
    #[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; // consumer dropped
                }
            }
        });
        Ok(rx)
    }

    /// Batched streaming scan under the `tombstones` feature (issue #1592).
    ///
    /// Tombstone builds reconcile via the materializing [`scan`](Self::scan)
    /// inside [`scan_stream`](Self::scan_stream); this re-chunks that per-row
    /// stream into `Vec` batches. Unlike the default build there is no
    /// single-reader straight-through path — a straight-through hand-off would
    /// bypass the cross-generation tombstone reconciliation. Content and order
    /// match [`scan_stream`](Self::scan_stream) exactly.
    #[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))
    }

    /// Get list of all SSTable IDs
    pub async fn list_sstables(&self) -> Vec<SSTableId> {
        let readers = self.readers.read().await;
        readers.keys().cloned().collect()
    }

    /// Remove an SSTable
    pub async fn remove_sstable(&self, sstable_id: &SSTableId) -> Result<()> {
        // Remove from memory
        {
            let mut readers = self.readers.write().await;
            readers.remove(sstable_id);
        }

        // Delete file
        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(())
    }

    /// Get SSTable statistics
    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
            },
        })
    }

    /// Set the schema registry for schema-aware operations
    ///
    /// This method stores the schema registry and applies it to all existing SSTable readers.
    /// Future readers loaded via `load_existing_sstables` or `load_from_table_directories`
    /// will also receive the schema registry during creation.
    #[cfg(feature = "state_machine")]
    pub async fn set_schema_registry(
        &self,
        registry: Arc<RwLock<crate::schema::SchemaRegistry>>,
    ) -> Result<()> {
        // Store the schema registry
        {
            let mut schema_reg = self.schema_registry.write().await;
            *schema_reg = Some(registry.clone());
        }

        // Apply to all existing readers
        // Note: SSTableReader::set_schema_registry requires &mut self, but readers are Arc<SSTableReader>
        // This is by design - schema should be set during reader creation, not after.
        // The stored registry will be applied to future readers loaded by this manager.

        // For existing readers, we cannot mutate them directly since they're behind Arc.
        // The schema registry will be applied to new readers as they're loaded.

        Ok(())
    }

    /// Merge multiple SSTables into a new one
    ///
    /// NOTE: SSTable writing removed in Issue #176 (writer.rs deleted).
    /// This method is feature-gated behind 'experimental' but currently unimplemented.
    #[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",
        ))
    }
}

/// SSTable statistics
#[derive(Debug, Clone)]
pub struct SSTableStats {
    /// Number of SSTable files
    pub sstable_count: usize,

    /// Total size of all SSTables in bytes
    pub total_size: u64,

    /// Total number of entries across all SSTables
    pub total_entries: u64,

    /// Total number of tables across all SSTables
    pub total_tables: u64,

    /// Average SSTable size in bytes
    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());

        // Create an empty list of discovered paths
        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();

        // Should have 0 SSTables since we provided an empty list
        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());

        // Create mock table directories with Data.db files
        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();
        // Note: These are mock files that won't parse as real SSTables,
        // but they test the directory scanning logic
        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();

        // Provide table directories to manager
        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();

        // VG3 update: `na-*-big-*` files are now correctly identified as BIG-format
        // headerless SSTables (VersionGates::Big(_)), so the SSTableManager can open
        // them with a minimal header even if the data content is invalid mock bytes.
        // The exact sstable_count depends on whether opening succeeds (it creates a
        // minimal header) or fails (if the mock bytes cause a deeper parse error).
        // We only assert the manager itself was created successfully (no panic/error).
        // The directory scanning logic is validated by the successful manager creation.
        let _ = stats.sstable_count; // count may be 0 or 3 depending on parse depth
    }

    #[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"));
    }

    /// Regression test for Issue #481: `._*` AppleDouble sidecars must not be
    /// returned by `find_data_files`.
    ///
    /// Before the fix, `find_data_files` only checked `ends_with("-Data.db")`,
    /// so `._nb-1-big-Data.db` passed the filter and would later fail to open
    /// as a valid SSTable.  The test would fail on the pre-fix code because
    /// `results` would contain two paths instead of one.
    #[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());

        // Write a minimal (invalid but correctly named) SSTable file and its
        // macOS AppleDouble sidecar companion alongside it.
        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();

        // find_data_files scans `temp_dir` with max_depth=0 (single level).
        let results = SSTableManager::find_data_files(&platform, temp_dir.path(), 0)
            .await
            .unwrap();

        // Only the real Data.db file should be returned; the ._ sidecar must be excluded.
        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"
        );
    }

    /// Unit test for the is_apple_double_sidecar helper.
    #[test]
    fn test_is_apple_double_sidecar() {
        // Must match
        assert!(is_apple_double_sidecar("._nb-1-big-Data.db"));
        assert!(is_apple_double_sidecar("._anything"));
        assert!(is_apple_double_sidecar("._"));
        // Must not match
        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;

        // Test standard Cassandra table directory format
        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()));

        // Test table name with hyphens
        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()));

        // Test multi_partition_table
        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())
        );

        // Test compression_test_table
        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())
        );

        // Test edge case: directory without UUID
        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()));

        // Test edge case: no parent directory
        let path = PathBuf::from("nb-1-big-Data.db");
        assert_eq!(extract_table_name(&path), None);
    }

    /// Issue #1321: the resolution-mode signal that BOTH `get()` builds thread
    /// into the BTI point-lookup guard is the single shared helper
    /// `SSTableManager::fully_qualified_match`. This compiles and runs under EVERY
    /// feature build (incl. `tombstones`/`--all-features`), so it pins that the
    /// tombstones-build manager `get()` is wired to the SAME relaxation as the
    /// default build — the gap roborev flagged was the wiring, not the guard.
    ///
    ///   - exact FQ match present in the map → relax (`true`);
    ///   - FQ query absent (would reach a reader only via the bare-name fallback)
    ///     → strict (`false`), so no wrong-keyspace rows;
    ///   - unqualified query → exact match (`true`), no keyspace to mismatch.
    #[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());

        // Exact fully-qualified key present → relax (the #1321 acceptance signal).
        assert!(
            SSTableManager::fully_qualified_match(&table_readers, "ks_a.users"),
            "exact FQ map hit must signal an exact match (relax)"
        );

        // Fully-qualified query whose exact key is ABSENT (resolution could only
        // succeed via the bare-name fallback) → strict, so the per-row guard keeps
        // strict keyspace matching and never surfaces ks_a's rows for a ks_b query.
        assert!(
            !SSTableManager::fully_qualified_match(&table_readers, "ks_b.users"),
            "FQ query missing its exact key must signal a fallback (strict)"
        );

        // Unqualified query has no keyspace to mismatch → treated as exact match.
        assert!(
            SSTableManager::fully_qualified_match(&table_readers, "users"),
            "unqualified query must signal an exact match (relax)"
        );
    }

    /// Open a real `SSTableReader` from the dataset for `keyspace.table`, or
    /// `None` if datasets are not present (so the test can skip in CI lanes
    /// without binaries). Used to obtain distinct `Arc<SSTableReader>` objects
    /// for the cross-keyspace bleed test below.
    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
    }

    /// Issue #1321 (roborev HIGH, cross-keyspace bleed): the `tombstones`-build
    /// manager `get()` builds its tombstone-merge set from `resolve_reader_list`
    /// (the resolved target table across generations) rather than iterating EVERY
    /// reader in `self.readers`. This pins the bleed-prevention invariant at the
    /// reader-set-resolution level: a fully-qualified query for `ks_a.users`
    /// resolves to a merge set containing ONLY the `ks_a.users` reader and NEVER
    /// the same-named `ks_b.users` reader.
    ///
    /// This would FAIL against the pre-fix b469818e behavior, where `get()`
    /// iterated `self.readers` (which holds BOTH readers) and — because the
    /// global relaxed `fully_qualified_match` flag was `true` (the FQ key existed)
    /// — admitted the wrong-keyspace reader's rows/tombstones into the merge.
    ///
    /// Uses two distinct real readers as the two keyspaces' SSTables; skips when
    /// datasets are absent (CI lanes without binaries).
    #[tokio::test]
    async fn test_tombstones_get_resolves_only_target_keyspace_readers() {
        // Two distinct on-disk readers stand in for same-named tables in two
        // different keyspaces (only their distinct identity matters here).
        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"
        );

        // Register them as same-named tables under two distinct keyspaces — the
        // exact `table_readers` layout that produced the bleed (Issue #680 keying).
        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)]);

        // The merge set the new tombstones get() iterates: ONLY ks_a's readers.
        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"
        );
        // The bleed assertion: the ks_b (wrong-keyspace) reader must NEVER be in
        // the ks_a merge set — even though `self.readers` (which the old code
        // iterated) contained it and the FQ flag was relaxed.
        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"
        );

        // And the relaxation signal stays correct for the FQ query (exact hit).
        assert!(
            SSTableManager::fully_qualified_match(&table_readers, "ks_a.users"),
            "exact FQ key present → relaxed guard, applied only to the resolved (ks_a) set"
        );
    }

    /// Resolve the on-disk table directory (the one holding `*-Data.db`) for
    /// `keyspace.table`, or `None` when datasets are absent (CI-lane skip).
    ///
    /// Skip keys off the `-Data.db` binary being present, NOT merely the table
    /// directory existing: a clean worktree ships the JSONL references (so the
    /// directory exists) but not the gitignored `-Data.db` binaries. Returning
    /// `Some` for a binary-less directory would open zero readers and make the
    /// caller's `readers.is_empty()` assertion PANIC instead of skip. Keeping the
    /// gate on the binary preserves "present fixture that yields 0 readers is a
    /// hard failure" (issue #2065, same doctrine as #1860).
    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
    }

    /// True if `dir` holds a `*-Data.db` binary (the gitignored fixture data),
    /// as opposed to only JSONL references. See `dataset_table_dir`.
    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"))
    }

    /// Issue #1571 (roborev Low, aggregate omission guard): `aggregate_key_cache_stats`
    /// must count every physically-open reader's key cache **exactly once**, even
    /// though `self.readers` (by-id) is NOT a strict superset of `table_readers`
    /// (by-name). `SSTableId::from_filename` keys the by-id map on the bare
    /// generation filename (`nb-1-big-Data.db`), so two tables sharing that
    /// filename collide: the by-id map keeps only the last-inserted reader while
    /// `table_readers` retains both. This test reproduces that collision with two
    /// real tables (both ship `nb-1-big-Data.db`) and proves the dedup-union
    /// aggregate (a) is not fooled into omitting the reader reachable only by name
    /// and (b) does not double-count the readers present in both maps. A pre-fix
    /// `self.readers`-only aggregate would FAIL this (under-counting the omitted
    /// reader's capacity).
    #[tokio::test]
    async fn test_aggregate_key_cache_counts_every_reader_exactly_once() {
        // Two distinct real table directories whose Data.db share a generation
        // filename → colliding SSTableId; skip when datasets are absent.
        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();

        // Deduped set of physically-open readers across BOTH maps (by pointer),
        // and the expected capacity sum computed independently of the aggregate.
        let (by_id_len, distinct, expected_capacity) = {
            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");
            assert!(
                !table_readers.is_empty(),
                "fixture present but no readers by-name"
            );
            let mut seen: std::collections::HashSet<*const reader::SSTableReader> =
                std::collections::HashSet::new();
            let mut expected_capacity = 0usize;
            for r in readers.values().chain(table_readers.values().flatten()) {
                if seen.insert(Arc::as_ptr(r)) {
                    expected_capacity = expected_capacity
                        .saturating_add(r.key_offset_cache.snapshot().capacity_bytes);
                }
            }
            (readers.len(), seen.len(), expected_capacity)
        };

        // The collision reality this guards against: the by-id map holds strictly
        // fewer readers than physically exist, so a `self.readers`-only aggregate
        // would silently omit at least one reader's cache.
        assert!(
            by_id_len < distinct,
            "expected an SSTableId collision (by-id map {by_id_len} < {distinct} distinct readers)"
        );

        // The dedup-union aggregate counts every distinct reader exactly once:
        // capacity equals the independently-summed deduped capacity — neither
        // under-counted (omission) nor over-counted (double-count).
        let agg = manager.aggregate_key_cache_stats().await;
        assert_eq!(
            agg.capacity_bytes, expected_capacity,
            "aggregate must sum each distinct reader's key-cache capacity exactly once"
        );
    }

    /// Issue #1592 (roborev Finding 1): a MID-STREAM error must NOT drop the
    /// already-received `Ok` rows accumulated in the pending batch. `rechunk_into_batches`
    /// (the zero/multi-generation + `tombstones` forwarder) must flush the pending
    /// rows BEFORE forwarding the terminal `Err`, matching the per-row `scan_stream`
    /// guarantee (issue #1143) that confirmed rows are delivered ahead of the error.
    #[tokio::test]
    async fn rechunk_flushes_pending_rows_before_midstream_error() {
        use reader::scan_stream_windowed::BATCH_EMIT_ROWS;

        // Feed FEWER than BATCH_EMIT_ROWS Ok rows (so nothing auto-flushes and they
        // all sit in the pending batch), then a mid-stream Err.
        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); // close the source

        let mut out = SSTableManager::rechunk_into_batches(in_rx, 64);

        // First item: the flushed pending batch with ALL confirmed rows, in order.
        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");
        }

        // Second item: the terminal error, AFTER the confirmed rows.
        let second = out.recv().await.expect("expected the terminal error item");
        assert!(second.is_err(), "second item must be the forwarded error");

        // Stream then ends.
        assert!(
            out.recv().await.is_none(),
            "no items after the terminal error"
        );
    }

    // ---- partition_key_shape_from_headers: multi-generation union (roborev 3786) ----

    /// Build a minimal `ColumnInfo` for the shape-resolution unit tests.
    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,
        }
    }

    /// roborev 3786: a REGULAR column present ONLY in a LATER generation must land in
    /// the UNIONed non-key name set, so a `WHERE <later-gen-regular-col> = <val>` is
    /// NOT misclassified as a partition-key seek. Two headers, single-component `int`
    /// pk, zero clustering: gen-1 has `v1`, gen-2 adds `v2`.
    #[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);
        // The later-generation regular column IS in the unioned non-key set — so the
        // classifier's by-elimination check will NOT treat it as the partition key.
        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"),
        );

        // The misclassification the fix prevents: had we used only gen-1, `v2` would
        // be absent from the non-key set and thus admitted by elimination as the pk.
        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",
        );
    }

    /// Inconsistent key metadata across readers (a pk-count disagreement) declines
    /// the shape → `None` → the caller full-scans (fail-safe, no heuristics).
    #[test]
    fn partition_key_shape_declines_inconsistent_readers() {
        // gen-1: single-component pk; gen-2: composite (2-component) pk.
        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)",
        );

        // A clustering-count disagreement likewise declines.
        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",
        );

        // A single-pk TYPE disagreement declines too.
        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",
        );
    }

    /// No header with a partition-key column → `None` (fail-safe). No readers, or an
    /// empty header, or a header exposing no pk column all decline (never a zero-key
    /// shape).
    #[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);

        // A populated header that names no partition-key column is not a usable pk
        // shape either → decline (complete-union fail-safe, roborev 3788).
        let no_pk = vec![col("v", "text", false, false)];
        assert_eq!(partition_key_shape_from_headers([no_pk.as_slice()]), None);
    }

    /// COMPLETE-UNION fail-safe (roborev 3788): if ANY resolved reader lacks a usable
    /// header, the union of non-key names is INCOMPLETE, so we cannot prove a column is
    /// the pk by its absence. One headerless reader among several usable ones ⇒ decline
    /// (`None`) so the classifier full-scans instead of misclassifying a regular column
    /// (present only in the headerless reader) as a partition-key seek.
    #[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![];

        // Headerless reader FIRST among several.
        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)",
        );
        // Headerless reader LAST — the union is still incomplete, still declines.
        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)",
        );
        // Headerless reader BETWEEN two usable ones.
        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",
        );

        // Control: the SAME usable readers WITHOUT the headerless one DO resolve — so
        // it is specifically the headerless reader that forces the fail-safe decline.
        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)",
        );
    }
}