cqlite-core 0.11.0

//! SSTable version-letter gates for Cassandra BIG and BTI formats.
//!
//! This module implements the per-letter feature-gate logic that mirrors
//! `BigFormat.java` and `BtiFormat.java` from Cassandra 5.0.8.  Each gate
//! is a `bool` field derived **only** from the two-letter version string found
//! in the SSTable filename prefix (e.g. `nb`, `oa`, `da`).
//!
//! ## Authority chain
//!
//! Cassandra 5.0.8 source (primary) > audit report B10 Part 2 > guide ch.22
//!
//! ### BIG format version letters (BigFormat.java:341-526)
//!
//! | Letter | Cassandra release | Notable additions |
//! |--------|-------------------|--------------------|
//! | `ma`   | 3.0.0             | Native row storage, BF hash swap |
//! | `mb`   | 3.0.7 / 3.7       | Commit-log lower bound |
//! | `mc`   | 3.0.8 / 3.9       | Commit-log intervals |
//! | `md`   | 3.0.18 / 3.11.4   | Accurate min/max clustering |
//! | `me`   | 3.0.25 / 3.11.11  | Originating host ID (first appearance) |
//! | `na`   | 4.0-rc1           | Uncompressed chunks, pending repair, metadata checksum |
//! | `nb`   | 4.0-rc2           | Default BIG letter for stock Cassandra 5.0 compat mode |
//! | `oa`   | 5.0               | Improved min/max, uint deletion time, key range, token coverage |
//!
//! ### BTI format version letters (BtiFormat.java:287-420)
//!
//! | Letter | Cassandra release | Notes |
//! |--------|------------------|-------|
//! | `da`   | 5.0              | Only BTI letter; all gates TRUE |
//!
//! ## Storage-compatibility-mode note
//!
//! Stock Cassandra 5.0 writes **`nb`-versioned BIG** SSTables when
//! `storage_compatibility_mode` is `CASSANDRA_4` (the default).  `oa` is
//! only written after explicitly raising the mode to `NONE`.
//!
//! ## SSTable ID forms (Descriptor.java:85, 95)
//!
//! Cassandra 5.0 supports **two** SSTable ID forms:
//! - Sequential: `nb-1-big-Data.db`  (integer id)
//! - UUID-based: `nb-6aa08200a25111f0a3fef1a551383fb9-big-Data.db` (hex string)
//!
//! Both forms are generated by real Cassandra 5.0 clusters; the UUID form is
//! the default since 5.0.0 (`uuid_sstable_identifiers_enabled: true`).

use std::path::Path;

use crate::{Error, Result};

/// SSTable format family: BIG (`big`) or BTI (`bti`).
///
/// Matches the `<format>` segment of the Cassandra filename pattern
/// `<version>-<id>-<format>-<component>.db`.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum SsTableFormat {
    /// "big" – the classic BIG format (Cassandra 3.0 – 5.0).
    Big,
    /// "bti" – the trie-based BTI format (Cassandra 5.0+).
    Bti,
}

impl SsTableFormat {
    /// Parse format name from string (`"big"` or `"bti"`).
    pub fn parse(s: &str) -> Option<Self> {
        match s {
            "big" => Some(Self::Big),
            "bti" => Some(Self::Bti),
            _ => None,
        }
    }

    /// Return the canonical lowercase name used in filenames.
    pub fn as_str(&self) -> &'static str {
        match self {
            Self::Big => "big",
            Self::Bti => "bti",
        }
    }
}

/// Parsed Cassandra SSTable descriptor extracted from a filename.
///
/// Filename pattern (Descriptor.java:251):
/// ```text
/// <version>-<id>-<format>-<component>.db
/// ```
///
/// Both sequential integer IDs (`1`, `2`, …) and UUID-ish hex string IDs
/// (`6aa08200a25111f0a3fef1a551383fb9`) are accepted.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SsTableDescriptor {
    /// Two-letter version string, e.g. `"nb"`, `"oa"`, `"da"`.
    pub version: String,
    /// Raw SSTable id as found in the filename (integer string or hex UUID).
    pub sstable_id: String,
    /// Format family (`big` or `bti`).
    pub format: SsTableFormat,
    /// Component suffix after the last `-`, e.g. `"Data.db"`.
    pub component: String,
}

impl SsTableDescriptor {
    /// Parse a Cassandra SSTable descriptor from a filename or file path.
    ///
    /// Accepts both:
    /// - `nb-1-big-Data.db`               (sequential integer id)
    /// - `nb-6aa08200a25111f0a3fef1a551383fb9-big-Data.db`  (UUID hex id)
    /// - `oa-00000000-0000-0000-0000-000000000001-big-Data.db` (hyphenated UUID)
    ///
    /// Returns an error if the filename does not contain at least four
    /// dash-separated segments or if the format segment is not `big` or `bti`.
    pub fn parse(path: &Path) -> Result<Self> {
        let filename = path
            .file_name()
            .and_then(|f| f.to_str())
            .ok_or_else(|| Error::InvalidPath(format!("Invalid SSTable path: {:?}", path)))?;

        Self::parse_filename(filename)
    }

    /// Parse from a bare filename string (no directory component required).
    pub fn parse_filename(filename: &str) -> Result<Self> {
        // Strip the `.db` extension if present so we can reason about the parts.
        let base = if let Some(b) = filename.strip_suffix(".db") {
            b
        } else if let Some(b) = filename.strip_suffix(".txt") {
            // TOC.txt – strip .txt instead
            b
        } else {
            filename
        };

        // Split on `-`.  The component itself may contain `-` (e.g. `TOC`
        // doesn't, but `CompressionInfo` doesn't either – however, future
        // components could).  We therefore split from the left and treat
        // everything from part[3] onwards as the component.
        //
        // Pattern: <version>-<id>-<format>-<component>
        //   parts[0] = version  (always 2 lowercase letters: [a-z]{2})
        //   parts[1..n-2] = id  (one or more dash-joined segments)
        //   parts[n-1] = format ("big" or "bti")
        //   parts[n] = component (rest of original, including original `.db` suffix)
        //
        // We search for the format segment by scanning right-to-left after
        // the first part for "big" or "bti", which avoids being tripped up
        // by dash-separated UUID ids.

        let parts: Vec<&str> = base.split('-').collect();
        if parts.len() < 4 {
            return Err(Error::InvalidFormat(format!(
                "SSTable filename has fewer than 4 dash-separated segments: {:?}",
                filename
            )));
        }

        let version = parts[0];
        // Validate version is exactly 2 lowercase letters.
        if version.len() != 2 || !version.chars().all(|c| c.is_ascii_lowercase()) {
            return Err(Error::InvalidFormat(format!(
                "SSTable version segment must be 2 lowercase letters, got {:?} in {:?}",
                version, filename
            )));
        }

        // Find the format segment by scanning right-to-left (skip the last
        // component part), starting from parts[2].
        // Strategy: look for "big" or "bti" starting from the second-to-last
        // non-component position.  The component name never equals "big" or "bti".
        let format_idx = parts[2..]
            .iter()
            .enumerate()
            .rev()
            .find(|(_, p)| **p == "big" || **p == "bti")
            .map(|(i, _)| i + 2); // offset back to original parts index

        let format_idx = format_idx.ok_or_else(|| {
            Error::InvalidFormat(format!(
                "No 'big' or 'bti' format segment found in {:?}",
                filename
            ))
        })?;

        let format = SsTableFormat::parse(parts[format_idx]).ok_or_else(|| {
            Error::InvalidFormat(format!(
                "Unknown format {:?} in {:?}",
                parts[format_idx], filename
            ))
        })?;

        // id is everything between version and format
        let sstable_id = parts[1..format_idx].join("-");

        // component is everything after format, re-joined and with extension restored
        let component_base = parts[format_idx + 1..].join("-");
        // Re-attach original extension
        let extension = if filename.ends_with(".db") {
            ".db"
        } else {
            ".txt"
        };
        let component = format!("{}{}", component_base, extension);

        Ok(Self {
            version: version.to_string(),
            sstable_id,
            format,
            component,
        })
    }
}

/// Per-letter feature gates for a BIG-format SSTable.
///
/// Each boolean field corresponds exactly to the gate computed in
/// `BigFormat.BigVersion` (Cassandra 5.0.8, lines 395-410).
///
/// Gates are derived solely from the **two-letter version string**; they do
/// not depend on file content.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct BigVersionGates {
    /// Raw version string this gate set was computed from.
    pub version: String,

    // ---- Gates in order as they appear in BigFormat.java ----
    /// `hasCommitLogLowerBound` — version >= `mb`
    /// (BigFormat.java:395)
    pub has_commit_log_lower_bound: bool,

    /// `hasCommitLogIntervals` — version >= `mc`
    /// (BigFormat.java:396)
    pub has_commit_log_intervals: bool,

    /// `hasAccurateMinMax` — matches `m[d-z]` or `n[a-z]`; **deprecated in `oa`**
    /// (BigFormat.java:397)
    pub has_accurate_min_max: bool,

    /// `hasLegacyMinMax` — matches `m[a-z]` or `n[a-z]`; **deprecated in `oa`**
    /// (BigFormat.java:398)
    pub has_legacy_min_max: bool,

    /// `hasOriginatingHostId` — version >= `nb` **OR** matches `m[e-z]`
    ///
    /// This is the straddle gate: it fires for the `me`–`mz` block of the `m`
    /// series AND for all versions >= `nb` in the `n`/`o` series.
    /// (BigFormat.java:400)
    pub has_originating_host_id: bool,

    /// `hasMaxCompressedLength` — version >= `na`
    /// (BigFormat.java:401)
    pub has_max_compressed_length: bool,

    /// `hasPendingRepair` — version >= `na`
    /// (BigFormat.java:402)
    pub has_pending_repair: bool,

    /// `hasIsTransient` — version >= `na`
    /// (BigFormat.java:403)
    pub has_is_transient: bool,

    /// `hasMetadataChecksum` — version >= `na`
    /// (BigFormat.java:404)
    pub has_metadata_checksum: bool,

    /// `hasOldBfFormat` — version < `na`  (old bloom-filter format)
    /// (BigFormat.java:405)
    pub has_old_bf_format: bool,

    /// `hasImprovedMinMax` — version >= `oa`  (**oa-only**)
    /// (BigFormat.java:406)
    pub has_improved_min_max: bool,

    /// `hasPartitionLevelDeletionPresenceMarker` — version >= `oa`  (**oa-only**)
    /// (BigFormat.java:407)
    pub has_partition_level_deletion_presence_marker: bool,

    /// `hasKeyRange` — version >= `oa`  (**oa-only**)
    /// (BigFormat.java:408)
    pub has_key_range: bool,

    /// `hasUIntDeletionTime` — version >= `oa`  (**oa-only**, 2106-safe TTL)
    /// (BigFormat.java:409)
    pub has_uint_deletion_time: bool,

    /// `hasTokenSpaceCoverage` — version >= `oa`  (**oa-only**)
    /// (BigFormat.java:410)
    pub has_token_space_coverage: bool,
}

impl BigVersionGates {
    /// Compute all gates for the given two-letter BIG-format version string.
    ///
    /// The version comparison uses lexicographic ordering of the raw string,
    /// which is correct because Cassandra uses single-character prefix letters
    /// (`m`, `n`, `o`) followed by a single lowercase suffix.  The Cassandra
    /// source code does the same (`version.compareTo("oa") >= 0`).
    ///
    /// # Errors
    ///
    /// Returns `Err` if `version` is not exactly two ASCII lowercase letters.
    pub fn from_version(version: &str) -> Result<Self> {
        if version.len() != 2 || !version.chars().all(|c| c.is_ascii_lowercase()) {
            return Err(Error::InvalidFormat(format!(
                "BIG version must be 2 lowercase letters, got {:?}",
                version
            )));
        }

        let v = version;

        // `version.matches("(m[d-z])|(n[a-z])")` from BigFormat.java line 397.
        let has_accurate_min_max = {
            let first = v.chars().next().unwrap();
            let second = v.chars().nth(1).unwrap();
            (first == 'm' && ('d'..='z').contains(&second))
                || (first == 'n' && second.is_ascii_lowercase())
        };

        // `version.matches("(m[a-z])|(n[a-z])")` from BigFormat.java line 398.
        let has_legacy_min_max = {
            let first = v.chars().next().unwrap();
            let second = v.chars().nth(1).unwrap();
            (first == 'm' && second.is_ascii_lowercase())
                || (first == 'n' && second.is_ascii_lowercase())
        };

        // `version.compareTo("nb") >= 0 || version.matches("(m[e-z])")` (line 400).
        let has_originating_host_id = {
            let first = v.chars().next().unwrap();
            let second = v.chars().nth(1).unwrap();
            v >= "nb" || (first == 'm' && ('e'..='z').contains(&second))
        };

        Ok(Self {
            version: version.to_string(),
            has_commit_log_lower_bound: v >= "mb",
            has_commit_log_intervals: v >= "mc",
            has_accurate_min_max,
            has_legacy_min_max,
            has_originating_host_id,
            has_max_compressed_length: v >= "na",
            has_pending_repair: v >= "na",
            has_is_transient: v >= "na",
            has_metadata_checksum: v >= "na",
            has_old_bf_format: v < "na",
            // oa-only gates: all false for nb, all true for oa
            has_improved_min_max: v >= "oa",
            has_partition_level_deletion_presence_marker: v >= "oa",
            has_key_range: v >= "oa",
            has_uint_deletion_time: v >= "oa",
            has_token_space_coverage: v >= "oa",
        })
    }

    /// Returns `true` if this version is compatible for reading according to
    /// `BigVersion.isCompatible()` (BigFormat.java:516-519).
    ///
    /// A version is compatible when:
    /// - It is >= `ma` (the earliest supported version), **and**
    /// - Its first letter is <= `o` (the first letter of the current `oa`)
    pub fn is_compatible(&self) -> bool {
        let v = self.version.as_str();
        v >= "ma" && v.chars().next().is_some_and(|c| c <= 'o')
    }

    /// Returns `true` when this is a stock Cassandra 5.0 default-mode SSTable
    /// (`nb` version — `storage_compatibility_mode = CASSANDRA_4`).
    pub fn is_cassandra5_compat_mode(&self) -> bool {
        self.version == "nb"
    }

    /// Returns `true` when this is a full Cassandra 5.0 SSTable (`oa` version —
    /// `storage_compatibility_mode = NONE`).
    pub fn is_cassandra5_native(&self) -> bool {
        self.version == "oa"
    }

    /// Infallible constructor returning gates for the `nb` version (stock Cassandra 5.0
    /// `storage_compatibility_mode = CASSANDRA_4`).
    ///
    /// Use this instead of `from_version("nb").expect(…)` in library code, which
    /// violates the project's no-`expect` mandate.  The field values are the literal
    /// results of evaluating `from_version("nb")`; a unit test in this module keeps
    /// them in sync with `from_version`.
    ///
    /// VG3 fall-back: when the SSTable filename cannot be parsed the reader defaults
    /// to these gates so existing behaviour is preserved.
    pub fn nb_fallback() -> Self {
        Self {
            version: "nb".to_string(),
            // Gates matching BigFormat.java for version "nb" ----------------
            has_commit_log_lower_bound: true, // "nb" >= "mb"
            has_commit_log_intervals: true,   // "nb" >= "mc"
            has_accurate_min_max: true,       // "nb" in n[a-z]
            has_legacy_min_max: true,         // "nb" in n[a-z]
            has_originating_host_id: true,    // "nb" >= "nb"
            has_max_compressed_length: true,  // "nb" >= "na"
            has_pending_repair: true,         // "nb" >= "na"
            has_is_transient: true,           // "nb" >= "na"
            has_metadata_checksum: true,      // "nb" >= "na"
            has_old_bf_format: false,         // "nb" NOT < "na"
            // oa-only gates — all FALSE for nb
            has_improved_min_max: false,
            has_partition_level_deletion_presence_marker: false,
            has_key_range: false,
            has_uint_deletion_time: false,
            has_token_space_coverage: false,
        }
    }
}

/// Feature gates for a BTI-format SSTable.
///
/// BtiFormat only has one version (`da`).  All modern feature gates are TRUE
/// for `da` (BtiFormat.java:321-418).
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct BtiVersionGates {
    /// Raw version string (always `"da"` for BTI).
    pub version: String,

    /// All gates are TRUE for `da`.  Fields mirror the BIG gates for API parity.
    pub has_commit_log_lower_bound: bool,
    pub has_commit_log_intervals: bool,
    pub has_max_compressed_length: bool,
    pub has_pending_repair: bool,
    pub has_is_transient: bool,
    pub has_metadata_checksum: bool,
    /// `hasOldBfFormat` is **FALSE** for BTI (BtiFormat.java:357-360).
    pub has_old_bf_format: bool,
    pub has_originating_host_id: bool,
    /// `hasAccurateMinMax` — **TRUE** for BTI `da`.
    ///
    /// Source: BtiFormat.java:363-366
    /// ```java
    /// public boolean hasAccurateMinMax() { return true; }
    /// ```
    pub has_accurate_min_max: bool,
    /// `hasLegacyMinMax` — **FALSE** for BTI `da`.
    ///
    /// Source: BtiFormat.java:368-371
    /// ```java
    /// public boolean hasLegacyMinMax() { return false; }
    /// ```
    pub has_legacy_min_max: bool,
    pub has_improved_min_max: bool,
    pub has_token_space_coverage: bool,
    pub has_partition_level_deletion_presence_marker: bool,
    pub has_key_range: bool,
    pub has_uint_deletion_time: bool,
}

impl BtiVersionGates {
    /// Compute BTI gates for the given version string.
    ///
    /// # Errors
    ///
    /// Returns `Err` if the version is not `"da"` (the only BTI version).
    pub fn from_version(version: &str) -> Result<Self> {
        if version != "da" {
            return Err(Error::InvalidFormat(format!(
                "BTI format only supports version 'da', got {:?}",
                version
            )));
        }
        Ok(Self {
            version: version.to_string(),
            has_commit_log_lower_bound: true,
            has_commit_log_intervals: true,
            has_max_compressed_length: true,
            has_pending_repair: true,
            has_is_transient: true,
            has_metadata_checksum: true,
            has_old_bf_format: false, // Always false for BTI (BtiFormat.java:357-360)
            has_originating_host_id: true,
            // BtiFormat.java:363-366: `public boolean hasAccurateMinMax() { return true; }`
            has_accurate_min_max: true,
            // BtiFormat.java:368-371: `public boolean hasLegacyMinMax() { return false; }`
            has_legacy_min_max: false,
            has_improved_min_max: true,
            has_token_space_coverage: true,
            has_partition_level_deletion_presence_marker: true,
            has_key_range: true,
            has_uint_deletion_time: true,
        })
    }
}

/// Combined version-gate result for any SSTable (BIG or BTI).
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum VersionGates {
    /// BIG-format gates.
    Big(BigVersionGates),
    /// BTI-format gates.
    Bti(BtiVersionGates),
}

impl VersionGates {
    /// Compute gates from a parsed `SsTableDescriptor`.
    pub fn from_descriptor(desc: &SsTableDescriptor) -> Result<Self> {
        match desc.format {
            SsTableFormat::Big => BigVersionGates::from_version(&desc.version).map(Self::Big),
            SsTableFormat::Bti => BtiVersionGates::from_version(&desc.version).map(Self::Bti),
        }
    }

    /// Compute gates directly from a file path.
    pub fn from_path(path: &Path) -> Result<Self> {
        let desc = SsTableDescriptor::parse(path)?;
        Self::from_descriptor(&desc)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::path::PathBuf;

    // -----------------------------------------------------------------------
    // SsTableDescriptor filename parsing
    // -----------------------------------------------------------------------

    #[test]
    fn test_descriptor_sequential_id() {
        let desc = SsTableDescriptor::parse_filename("nb-1-big-Data.db").unwrap();
        assert_eq!(desc.version, "nb");
        assert_eq!(desc.sstable_id, "1");
        assert_eq!(desc.format, SsTableFormat::Big);
        assert_eq!(desc.component, "Data.db");
    }

    #[test]
    fn test_descriptor_uuid_id_no_hyphens() {
        // UUID form used in the CQLite test corpus: 32-hex-char id with no hyphens
        let filename = "nb-6aa08200a25111f0a3fef1a551383fb9-big-Data.db";
        let desc = SsTableDescriptor::parse_filename(filename).unwrap();
        assert_eq!(desc.version, "nb");
        assert_eq!(desc.sstable_id, "6aa08200a25111f0a3fef1a551383fb9");
        assert_eq!(desc.format, SsTableFormat::Big);
        assert_eq!(desc.component, "Data.db");
    }

    #[test]
    fn test_descriptor_oa_version() {
        let desc = SsTableDescriptor::parse_filename("oa-1-big-Data.db").unwrap();
        assert_eq!(desc.version, "oa");
        assert_eq!(desc.format, SsTableFormat::Big);
    }

    #[test]
    fn test_descriptor_da_bti_version() {
        let desc = SsTableDescriptor::parse_filename("da-1-bti-Partitions.db").unwrap();
        assert_eq!(desc.version, "da");
        assert_eq!(desc.format, SsTableFormat::Bti);
        assert_eq!(desc.component, "Partitions.db");
    }

    #[test]
    fn test_descriptor_legacy_versions() {
        for version in &["ma", "mb", "mc", "md", "me", "na"] {
            let filename = format!("{}-3-big-Data.db", version);
            let desc = SsTableDescriptor::parse_filename(&filename).unwrap();
            assert_eq!(desc.version, *version, "version mismatch for {}", filename);
            assert_eq!(desc.format, SsTableFormat::Big);
        }
    }

    #[test]
    fn test_descriptor_toc_txt() {
        let desc = SsTableDescriptor::parse_filename("nb-1-big-TOC.txt").unwrap();
        assert_eq!(desc.version, "nb");
        assert_eq!(desc.component, "TOC.txt");
    }

    #[test]
    fn test_descriptor_compression_info() {
        let desc = SsTableDescriptor::parse_filename("nb-1-big-CompressionInfo.db").unwrap();
        assert_eq!(desc.component, "CompressionInfo.db");
    }

    #[test]
    fn test_descriptor_invalid_too_few_parts() {
        assert!(SsTableDescriptor::parse_filename("nb-Data.db").is_err());
        assert!(SsTableDescriptor::parse_filename("Data.db").is_err());
    }

    #[test]
    fn test_descriptor_invalid_version_not_two_letters() {
        assert!(SsTableDescriptor::parse_filename("nba-1-big-Data.db").is_err());
        assert!(SsTableDescriptor::parse_filename("n-1-big-Data.db").is_err());
    }

    #[test]
    fn test_descriptor_invalid_no_format_segment() {
        assert!(SsTableDescriptor::parse_filename("nb-1-xxx-Data.db").is_err());
    }

    #[test]
    fn test_descriptor_from_path() {
        let path = PathBuf::from(
            "test-data/datasets/sstables/test_basic/simple_table-6aa08200/nb-1-big-Data.db",
        );
        let desc = SsTableDescriptor::parse(&path).unwrap();
        assert_eq!(desc.version, "nb");
        assert_eq!(desc.format, SsTableFormat::Big);
    }

    // -----------------------------------------------------------------------
    // BigVersionGates: nb (stock Cassandra 5.0 default)
    // -----------------------------------------------------------------------

    #[test]
    fn test_big_nb_gates() {
        let g = BigVersionGates::from_version("nb").unwrap();

        // Gates that ARE set for nb
        assert!(g.has_commit_log_lower_bound, "nb: hasCommitLogLowerBound");
        assert!(g.has_commit_log_intervals, "nb: hasCommitLogIntervals");
        assert!(g.has_max_compressed_length, "nb: hasMaxCompressedLength");
        assert!(g.has_pending_repair, "nb: hasPendingRepair");
        assert!(g.has_is_transient, "nb: hasIsTransient");
        assert!(g.has_metadata_checksum, "nb: hasMetadataChecksum");
        assert!(!g.has_old_bf_format, "nb: !hasOldBfFormat");
        assert!(
            g.has_originating_host_id,
            "nb: hasOriginatingHostId (nb >= nb)"
        );

        // oa-only gates must be FALSE for nb
        assert!(!g.has_improved_min_max, "nb: !hasImprovedMinMax (oa-only)");
        assert!(
            !g.has_partition_level_deletion_presence_marker,
            "nb: !hasPartitionLevelDeletionPresenceMarker (oa-only)"
        );
        assert!(!g.has_key_range, "nb: !hasKeyRange (oa-only)");
        assert!(
            !g.has_uint_deletion_time,
            "nb: !hasUIntDeletionTime (oa-only)"
        );
        assert!(
            !g.has_token_space_coverage,
            "nb: !hasTokenSpaceCoverage (oa-only)"
        );
    }

    // -----------------------------------------------------------------------
    // BigVersionGates: oa (Cassandra 5.0 native mode)
    // -----------------------------------------------------------------------

    #[test]
    fn test_big_oa_gates() {
        let g = BigVersionGates::from_version("oa").unwrap();

        // All na+ gates still set
        assert!(g.has_commit_log_lower_bound);
        assert!(g.has_commit_log_intervals);
        assert!(g.has_max_compressed_length);
        assert!(g.has_pending_repair);
        assert!(g.has_is_transient);
        assert!(g.has_metadata_checksum);
        assert!(!g.has_old_bf_format);
        assert!(g.has_originating_host_id, "oa >= nb");

        // oa-only gates must be TRUE for oa
        assert!(g.has_improved_min_max, "oa: hasImprovedMinMax");
        assert!(
            g.has_partition_level_deletion_presence_marker,
            "oa: hasPartitionLevelDeletionPresenceMarker"
        );
        assert!(g.has_key_range, "oa: hasKeyRange");
        assert!(g.has_uint_deletion_time, "oa: hasUIntDeletionTime");
        assert!(g.has_token_space_coverage, "oa: hasTokenSpaceCoverage");

        // AccurateMinMax is deprecated in oa — should be FALSE
        assert!(
            !g.has_accurate_min_max,
            "oa: hasAccurateMinMax MUST be false (deprecated)"
        );
        // LegacyMinMax also deprecated in oa
        assert!(
            !g.has_legacy_min_max,
            "oa: hasLegacyMinMax MUST be false (deprecated)"
        );
    }

    // -----------------------------------------------------------------------
    // BigVersionGates: oa-only gates are NOT set on nb  (core correctness)
    // -----------------------------------------------------------------------

    #[test]
    fn test_oa_only_gates_absent_from_nb() {
        let nb = BigVersionGates::from_version("nb").unwrap();
        let oa = BigVersionGates::from_version("oa").unwrap();

        let oa_only_gate_names = [
            (
                "hasImprovedMinMax",
                nb.has_improved_min_max,
                oa.has_improved_min_max,
            ),
            (
                "hasPartitionLevelDeletionPresenceMarker",
                nb.has_partition_level_deletion_presence_marker,
                oa.has_partition_level_deletion_presence_marker,
            ),
            ("hasKeyRange", nb.has_key_range, oa.has_key_range),
            (
                "hasUIntDeletionTime",
                nb.has_uint_deletion_time,
                oa.has_uint_deletion_time,
            ),
            (
                "hasTokenSpaceCoverage",
                nb.has_token_space_coverage,
                oa.has_token_space_coverage,
            ),
        ];

        for (name, nb_val, oa_val) in &oa_only_gate_names {
            assert!(!nb_val, "nb.{} must be FALSE (oa-only gate)", name);
            assert!(oa_val, "oa.{} must be TRUE", name);
        }
    }

    // -----------------------------------------------------------------------
    // BigVersionGates: hasOriginatingHostId straddle gate
    // -----------------------------------------------------------------------

    /// `hasOriginatingHostId` introduced in `me` (straddles letter boundary).
    /// Source: BigFormat.java:400
    ///   `version.compareTo("nb") >= 0 || version.matches("(m[e-z])")`
    #[test]
    fn test_originating_host_id_straddle_gate() {
        // Must be FALSE for versions before me in the m-series
        for v in &["ma", "mb", "mc", "md"] {
            let g = BigVersionGates::from_version(v).unwrap();
            assert!(
                !g.has_originating_host_id,
                "{}: hasOriginatingHostId must be FALSE",
                v
            );
        }

        // Must be TRUE for me..mz
        for v in &["me", "mf", "mz"] {
            let g = BigVersionGates::from_version(v).unwrap();
            assert!(
                g.has_originating_host_id,
                "{}: hasOriginatingHostId must be TRUE (m[e-z] match)",
                v
            );
        }

        // Must be TRUE for na..nz (>= nb is lexicographically satisfied by the
        // whole n-series above nb: na < nb so na must be FALSE)
        let na = BigVersionGates::from_version("na").unwrap();
        assert!(
            !na.has_originating_host_id,
            "na: hasOriginatingHostId must be FALSE (na < nb, not m[e-z])"
        );

        // nb and above: TRUE
        for v in &["nb", "nc", "oa"] {
            let g = BigVersionGates::from_version(v).unwrap();
            assert!(
                g.has_originating_host_id,
                "{}: hasOriginatingHostId must be TRUE (>= nb)",
                v
            );
        }
    }

    // -----------------------------------------------------------------------
    // BigVersionGates: older versions
    // -----------------------------------------------------------------------

    #[test]
    fn test_big_ma_gates() {
        let g = BigVersionGates::from_version("ma").unwrap();
        // ma has none of the later features
        assert!(!g.has_commit_log_lower_bound);
        assert!(!g.has_commit_log_intervals);
        assert!(!g.has_accurate_min_max);
        assert!(g.has_legacy_min_max, "ma is in m[a-z]");
        assert!(!g.has_originating_host_id);
        assert!(!g.has_max_compressed_length);
        assert!(g.has_old_bf_format, "ma: hasOldBfFormat");
        assert!(!g.has_improved_min_max);
        assert!(!g.has_key_range);
        assert!(!g.has_uint_deletion_time);
        assert!(!g.has_token_space_coverage);
    }

    #[test]
    fn test_big_na_gates() {
        let g = BigVersionGates::from_version("na").unwrap();
        assert!(g.has_commit_log_lower_bound);
        assert!(g.has_commit_log_intervals);
        assert!(g.has_accurate_min_max, "na is in n[a-z]");
        assert!(g.has_legacy_min_max, "na is in n[a-z]");
        assert!(!g.has_originating_host_id, "na < nb");
        assert!(g.has_max_compressed_length);
        assert!(g.has_pending_repair);
        assert!(!g.has_old_bf_format);
        assert!(!g.has_improved_min_max, "oa-only");
    }

    #[test]
    fn test_big_md_gates() {
        let g = BigVersionGates::from_version("md").unwrap();
        assert!(g.has_accurate_min_max, "md is m[d-z]");
        assert!(g.has_legacy_min_max, "md is m[a-z]");
        assert!(!g.has_originating_host_id, "md < me");
    }

    #[test]
    fn test_big_me_gates() {
        let g = BigVersionGates::from_version("me").unwrap();
        assert!(g.has_accurate_min_max, "me is m[d-z]");
        assert!(g.has_originating_host_id, "me matches m[e-z]");
    }

    // -----------------------------------------------------------------------
    // BigVersionGates: isCompatible
    // -----------------------------------------------------------------------

    #[test]
    fn test_big_is_compatible() {
        // All known valid versions should be compatible
        for v in &["ma", "mb", "mc", "md", "me", "na", "nb", "oa"] {
            let g = BigVersionGates::from_version(v).unwrap();
            assert!(g.is_compatible(), "{} should be compatible", v);
        }
        // 'pa' would be next major after oa — not compatible if current is oa
        // (first letter 'p' > 'o')
        let pa = BigVersionGates::from_version("pa").unwrap();
        assert!(
            !pa.is_compatible(),
            "pa is beyond current 'oa' major letter"
        );
    }

    #[test]
    fn test_big_is_cassandra5_mode() {
        let nb = BigVersionGates::from_version("nb").unwrap();
        assert!(nb.is_cassandra5_compat_mode());
        assert!(!nb.is_cassandra5_native());

        let oa = BigVersionGates::from_version("oa").unwrap();
        assert!(!oa.is_cassandra5_compat_mode());
        assert!(oa.is_cassandra5_native());
    }

    // -----------------------------------------------------------------------
    // BigVersionGates::nb_fallback — must match from_version("nb") exactly
    // -----------------------------------------------------------------------

    /// Verify that `BigVersionGates::nb_fallback()` produces the same gate
    /// values as `BigVersionGates::from_version("nb")`.  This test is the
    /// automated guard that keeps the two in sync.
    #[test]
    fn test_nb_fallback_matches_from_version() {
        let from_fn = BigVersionGates::from_version("nb").unwrap();
        let fallback = BigVersionGates::nb_fallback();

        assert_eq!(fallback.version, from_fn.version);
        assert_eq!(
            fallback.has_commit_log_lower_bound,
            from_fn.has_commit_log_lower_bound
        );
        assert_eq!(
            fallback.has_commit_log_intervals,
            from_fn.has_commit_log_intervals
        );
        assert_eq!(fallback.has_accurate_min_max, from_fn.has_accurate_min_max);
        assert_eq!(fallback.has_legacy_min_max, from_fn.has_legacy_min_max);
        assert_eq!(
            fallback.has_originating_host_id,
            from_fn.has_originating_host_id
        );
        assert_eq!(
            fallback.has_max_compressed_length,
            from_fn.has_max_compressed_length
        );
        assert_eq!(fallback.has_pending_repair, from_fn.has_pending_repair);
        assert_eq!(fallback.has_is_transient, from_fn.has_is_transient);
        assert_eq!(
            fallback.has_metadata_checksum,
            from_fn.has_metadata_checksum
        );
        assert_eq!(fallback.has_old_bf_format, from_fn.has_old_bf_format);
        assert_eq!(fallback.has_improved_min_max, from_fn.has_improved_min_max);
        assert_eq!(
            fallback.has_partition_level_deletion_presence_marker,
            from_fn.has_partition_level_deletion_presence_marker
        );
        assert_eq!(fallback.has_key_range, from_fn.has_key_range);
        assert_eq!(
            fallback.has_uint_deletion_time,
            from_fn.has_uint_deletion_time
        );
        assert_eq!(
            fallback.has_token_space_coverage,
            from_fn.has_token_space_coverage
        );
    }

    // -----------------------------------------------------------------------
    // BigVersionGates: invalid input
    // -----------------------------------------------------------------------

    #[test]
    fn test_big_invalid_version() {
        assert!(BigVersionGates::from_version("n").is_err());
        assert!(BigVersionGates::from_version("nba").is_err());
        assert!(BigVersionGates::from_version("NB").is_err());
        assert!(BigVersionGates::from_version("").is_err());
    }

    // -----------------------------------------------------------------------
    // BtiVersionGates: da
    // -----------------------------------------------------------------------

    #[test]
    fn test_bti_da_gates() {
        let g = BtiVersionGates::from_version("da").unwrap();
        assert!(g.has_commit_log_lower_bound);
        assert!(g.has_commit_log_intervals);
        assert!(g.has_max_compressed_length);
        assert!(g.has_pending_repair);
        assert!(g.has_is_transient);
        assert!(g.has_metadata_checksum);
        assert!(!g.has_old_bf_format, "da: !hasOldBfFormat");
        assert!(g.has_originating_host_id);
        // BtiFormat.java:363-366: hasAccurateMinMax() → true
        assert!(
            g.has_accurate_min_max,
            "da: hasAccurateMinMax (BtiFormat.java:363)"
        );
        // BtiFormat.java:368-371: hasLegacyMinMax() → false
        assert!(
            !g.has_legacy_min_max,
            "da: !hasLegacyMinMax (BtiFormat.java:368)"
        );
        assert!(g.has_improved_min_max);
        assert!(g.has_token_space_coverage);
        assert!(g.has_partition_level_deletion_presence_marker);
        assert!(g.has_key_range);
        assert!(g.has_uint_deletion_time);
    }

    #[test]
    fn test_bti_rejects_non_da() {
        assert!(BtiVersionGates::from_version("nb").is_err());
        assert!(BtiVersionGates::from_version("oa").is_err());
        assert!(BtiVersionGates::from_version("na").is_err());
    }

    // -----------------------------------------------------------------------
    // VersionGates combined
    // -----------------------------------------------------------------------

    #[test]
    fn test_version_gates_from_path_nb() {
        let path = PathBuf::from("nb-1-big-Data.db");
        let gates = VersionGates::from_path(&path).unwrap();
        match gates {
            VersionGates::Big(g) => assert_eq!(g.version, "nb"),
            VersionGates::Bti(_) => panic!("Expected Big"),
        }
    }

    #[test]
    fn test_version_gates_from_path_oa() {
        let path = PathBuf::from("oa-1-big-Data.db");
        let gates = VersionGates::from_path(&path).unwrap();
        match gates {
            VersionGates::Big(g) => {
                assert_eq!(g.version, "oa");
                assert!(g.has_uint_deletion_time);
            }
            VersionGates::Bti(_) => panic!("Expected Big"),
        }
    }

    #[test]
    fn test_version_gates_from_path_da() {
        let path = PathBuf::from("da-1-bti-Partitions.db");
        let gates = VersionGates::from_path(&path).unwrap();
        match gates {
            VersionGates::Bti(g) => assert_eq!(g.version, "da"),
            VersionGates::Big(_) => panic!("Expected Bti"),
        }
    }

    /// Verify that UUID-based ids (corpus filenames) parse correctly into gates.
    #[test]
    fn test_version_gates_from_corpus_filename() {
        let path = PathBuf::from("nb-6aa08200a25111f0a3fef1a551383fb9-big-Data.db");
        let gates = VersionGates::from_path(&path).unwrap();
        match gates {
            VersionGates::Big(g) => {
                assert_eq!(g.version, "nb");
                // oa-only gates must be absent
                assert!(!g.has_improved_min_max);
                assert!(!g.has_uint_deletion_time);
            }
            VersionGates::Bti(_) => panic!("Expected Big"),
        }
    }

    // -----------------------------------------------------------------------
    // Docker-generated fixture filenames
    // These filenames come from Cassandra 5.0.8 containers run with:
    //   storage_compatibility_mode: NONE  (for oa)
    //   sstable.selected_format: bti       (for da)
    // -----------------------------------------------------------------------

    /// `oa-2-big-Data.db` generated by Cassandra 5.0.8 with
    /// `storage_compatibility_mode: NONE`.
    #[test]
    fn test_descriptor_docker_oa_sequential() {
        let desc = SsTableDescriptor::parse_filename("oa-2-big-Data.db").unwrap();
        assert_eq!(desc.version, "oa");
        assert_eq!(desc.sstable_id, "2");
        assert_eq!(desc.format, SsTableFormat::Big);
        assert_eq!(desc.component, "Data.db");
    }

    /// Gates for the Docker-generated `oa` fixture must have all 5 oa-only
    /// gates TRUE.
    #[test]
    fn test_gates_docker_oa_fixture() {
        let gates = VersionGates::from_path(&PathBuf::from("oa-2-big-Data.db")).unwrap();
        match gates {
            VersionGates::Big(g) => {
                assert_eq!(g.version, "oa");
                assert!(g.has_improved_min_max, "oa fixture: hasImprovedMinMax");
                assert!(
                    g.has_partition_level_deletion_presence_marker,
                    "oa fixture: hasPartitionLevelDeletionPresenceMarker"
                );
                assert!(g.has_key_range, "oa fixture: hasKeyRange");
                assert!(g.has_uint_deletion_time, "oa fixture: hasUIntDeletionTime");
                assert!(
                    g.has_token_space_coverage,
                    "oa fixture: hasTokenSpaceCoverage"
                );
                // deprecated in oa
                assert!(
                    !g.has_accurate_min_max,
                    "oa fixture: hasAccurateMinMax deprecated"
                );
                assert!(
                    !g.has_legacy_min_max,
                    "oa fixture: hasLegacyMinMax deprecated"
                );
            }
            VersionGates::Bti(_) => panic!("Expected Big gates for oa-2-big-Data.db"),
        }
    }

    /// `da-2-bti-Data.db` generated by Cassandra 5.0.8 with BTI format enabled.
    #[test]
    fn test_descriptor_docker_da_bti() {
        let desc = SsTableDescriptor::parse_filename("da-2-bti-Data.db").unwrap();
        assert_eq!(desc.version, "da");
        assert_eq!(desc.sstable_id, "2");
        assert_eq!(desc.format, SsTableFormat::Bti);
        assert_eq!(desc.component, "Data.db");
    }

    /// `da-2-bti-Partitions.db` — BTI-specific index component.
    #[test]
    fn test_descriptor_docker_da_bti_partitions() {
        let desc = SsTableDescriptor::parse_filename("da-2-bti-Partitions.db").unwrap();
        assert_eq!(desc.version, "da");
        assert_eq!(desc.format, SsTableFormat::Bti);
        assert_eq!(desc.component, "Partitions.db");
    }

    /// Gates for the Docker-generated `da` fixture: all BTI gates TRUE.
    #[test]
    fn test_gates_docker_da_fixture() {
        let gates = VersionGates::from_path(&PathBuf::from("da-2-bti-Data.db")).unwrap();
        match gates {
            VersionGates::Bti(g) => {
                assert_eq!(g.version, "da");
                assert!(g.has_improved_min_max, "da: hasImprovedMinMax");
                assert!(g.has_key_range, "da: hasKeyRange");
                assert!(g.has_uint_deletion_time, "da: hasUIntDeletionTime");
                assert!(g.has_token_space_coverage, "da: hasTokenSpaceCoverage");
                assert!(
                    g.has_partition_level_deletion_presence_marker,
                    "da: hasPartitionLevelDeletionPresenceMarker"
                );
                assert!(!g.has_old_bf_format, "da: !hasOldBfFormat");
                assert!(g.has_originating_host_id, "da: hasOriginatingHostId");
                // BtiFormat.java:363-371
                assert!(g.has_accurate_min_max, "da: hasAccurateMinMax");
                assert!(!g.has_legacy_min_max, "da: !hasLegacyMinMax");
            }
            VersionGates::Big(_) => panic!("Expected Bti gates for da-2-bti-Data.db"),
        }
    }
}