cqlite-core 0.15.0

//! Process-global read-work counters for partition-targeted lookups
//! (Issue #958, Epic #951).
//!
//! # Why this exists
//!
//! #949 added a partition-targeted lookup ([`SSTableManager::scan_partition`])
//! that prunes the SSTable set via the bloom filter / BTI trie before parsing,
//! so a `WHERE pk = ?` over a table backed by *N* SSTables touches only the
//! handful of candidates that can hold the key — not all *N*. Correct result
//! rows do not prove that pruning happened: a regression could quietly revert to
//! "open and scan every SSTable, then filter in memory" and still return the
//! right answer.
//!
//! These counters make the *work* observable so a CI test can fail the moment a
//! single-partition read starts scaling with the total SSTable count. They
//! mirror the [`scan_for_key_call_count`](crate::storage::sstable::SSTableReader::scan_for_key_call_count)
//! probe (issue #831) and the access-path probe (issue #960): a process-global
//! atomic with [`reset`]/getter accessors, observable from an integration test
//! without parsing logs and from the streaming path's spawned task.
//!
//! # What each counter counts
//!
//! - [`sstables_scanned`] — incremented **once per candidate SSTable reader that
//!   `scan_partition` actually parses** for a partition-targeted lookup. After
//!   bloom/BTI pruning drops the SSTables that cannot hold the key, every
//!   surviving candidate whose `Data.db` is parsed (whether through the
//!   cross-generation k-way merge or the per-reader concat fallback) bumps this
//!   by one. It is the O(candidates) signal: for a key living in one SSTable it
//!   must stay near 1 (plus any bloom false-positives), never grow to N. It is
//!   **not** incremented on the full-scan path — only the targeted lookup is
//!   instrumented, which is exactly the path #949/#958 protect.
//!
//! - [`partitions_parsed`] — incremented **once per partition row a
//!   partition-targeted lookup returns** (after retaining only the target key /
//!   the merge emits the partition). For a single-partition point lookup this is
//!   0 (absent key) or the number of rows that partition holds; it never scales
//!   with the table's total partition count, which a full scan would.
//!
//! - [`chunks_decompressed`] — incremented **once per compression chunk the
//!   single-candidate seek path actually decompresses** while buffering the
//!   target partition's bytes (Issue #953 / #951). Where `partitions_decoded`
//!   proves we returned only one partition, this proves the seek bounded its
//!   *decompression I/O* to that partition's chunk span rather than reading the
//!   whole `Data.db` to EOF. A head-of-file point lookup must NOT decompress the
//!   tail chunks of a large SSTable: a regression that stitches to EOF (the bug
//!   the bound replaces) bumps this by the file's whole chunk count, which the
//!   `issue_953` bound test catches even though `partitions_decoded` stays 1.
//!   Incremented at the seek's single decompression site
//!   (`SSTableReader::bti_pull_decompressed_chunk`), so every chunk the seek
//!   materializes — BIG (`nb`, bounded by the `Index.db` size) or BTI (`da`,
//!   bounded by the next-partition boundary) — is counted exactly once. The
//!   whole-section `stitch_all_chunks` fallback does **not** bump it (it is the
//!   unbounded path the seek avoids when chunk targeting is possible).
//!
//! - [`partitions_decoded`] — incremented **once per partition actually decoded
//!   out of `Data.db`** by the single-candidate *seek* path (Issue #953). Where
//!   `sstables_scanned` proves we touched few SSTables, this proves that *within*
//!   a touched SSTable we did not decode every partition: the seek resolves the
//!   target partition's `Data.db` offset (via the BTI trie or `Index.db`) and
//!   decodes only that one partition, so this stays O(1) for a point lookup. A
//!   regression that reverts the single-candidate path to a full parse-then-retain
//!   would bump this by the SSTable's whole partition count (~N), failing the
//!   `issue_953` bound. It is incremented at the per-partition decode site of the
//!   seek (`SSTableReader::scan_single_partition` → the emit closure that captures
//!   a complete partition), NOT at the result-count boundary. The full-scan +
//!   retain fallback does **not** bump it (it is the unoptimized path the seek
//!   replaces); a candidate that cannot be seeked therefore reads 0 here, which is
//!   why the test asserts a *small upper bound* (decode happened cheaply) rather
//!   than an exact equality.
//!
//! # Cost
//!
//! Each increment is a single `Relaxed` atomic add on the cold per-lookup
//! boundary (once per candidate / once per returned partition), never inside an
//! inner byte-decoding loop, so the hot path is unaffected. The counters are not
//! gated behind `cfg(test)` because integration tests in `tests/` compile
//! against the library crate without its `test` cfg (same rationale as
//! `SCAN_FOR_KEY_CALLS`).

use std::sync::atomic::{AtomicU64, Ordering};

/// The five read-work counters as one value.
///
/// Production code shares a single process-global instance ([`COUNTERS`]) reached
/// through the free functions below; the increment sites and integration probes
/// all operate on that instance. Bundling the atomics in a struct also lets a
/// unit test exercise the add/get/reset contract against a *local* instance,
/// immune to other tests concurrently mutating the global (issue #1071) — the
/// global is shared with read-path code that any parallel test can drive.
struct Counters {
    /// Candidate SSTables actually parsed by a partition-targeted lookup.
    sstables_scanned: AtomicU64,
    /// Partitions a partition-targeted lookup has returned.
    partitions_parsed: AtomicU64,
    /// Partitions DECODED from `Data.db` by the single-candidate seek (Issue #953).
    partitions_decoded: AtomicU64,
    /// Compression chunks DECOMPRESSED by the single-candidate seek (Issue #953 / #951).
    chunks_decompressed: AtomicU64,
    /// Individual rows DECODED within a partition by the seek path (Issue #954).
    rows_decoded: AtomicU64,
    /// Promoted-index blocks DECODED back-to-front by the BIG reverse partition
    /// iterator (Issue #1184). One per block the reverse walk visits.
    reverse_blocks_decoded: AtomicU64,
    /// Peak number of decoded rows held in a SINGLE block buffer by the BIG
    /// reverse iterator (Issue #1184) — the per-iteration memory high-water mark.
    /// A regression that materialises the whole partition before reversing pushes
    /// this to the partition's full row count instead of one block's worth.
    reverse_peak_block_rows: AtomicU64,
    /// Full re-reads of a finished `Data.db` performed to compute checksums
    /// (Issue #1663). Bumped once each time the re-read CRC oracle
    /// (`crc_writer::build_crc_bytes`) or the digest re-read
    /// (`SSTableWriter::compute_crc32`) OPENS `Data.db`. The streaming write path
    /// accumulates both checksums as it writes, so a non-empty
    /// `SSTableWriter::finish()` must leave this at 0; a regression that
    /// reintroduces the finish-time double re-read bumps it (2 on the old path).
    data_db_checksum_full_reads: AtomicU64,
    /// Partition BODIES read + parsed by a `do_get` compaction-merge scan — one
    /// per partition whose `Data.db` slice is decoded (Issue #2398). Incremented on
    /// BOTH per-SSTable enumeration paths: the streaming full-index walk
    /// (`stream_all_partitions_via_full_index`, the uncompressed `V5_0Uncompressed`
    /// field path, issue #2361) AND the chunk-stitching walk
    /// (`drain_compaction_window`, the `nb`/`V5CompressedLegacy` path). This is the
    /// O(partitions actually walked) signal for a scan. A token-range split must
    /// walk only the entries whose token falls in the split's `(start, end]` range
    /// (the index/data are token-ordered), NOT every partition in the SSTable: the
    /// current read paths apply the token filter only DOWNSTREAM at the consumer
    /// (`MergeProducer::drive_merge`), so this bumps by the SSTable's WHOLE
    /// partition count regardless of how narrow the split range (or `LIMIT`) is —
    /// the fixed multi-second warm-scan setup this counter exists to make visible.
    stream_walk_partitions_parsed: AtomicU64,
    /// Merge ENTRIES decoded from `Data.db` by ANY [`KWayMerger`]-adapter-driven
    /// run (Issue #2096) — full scans, compaction, AND multi-candidate point
    /// reads all share the same [`SSTableRowIteratorAdapter`]/`PathProbe::Seeked`
    /// increment sites, so this counts entries for all of them, not point reads
    /// alone. Bumped once per merge entry a run actually materialises out of
    /// `Data.db`: once per `Ok(entry)` streamed on the full-scan adapter run, and
    /// once per entry built from a seeked partition on the seek run's
    /// `PathProbe::Seeked` arm. A single partition with N clustering rows bumps
    /// this N times (an ENTRY-granularity counter, not partition-granularity).
    ///
    /// This makes a merge run's decode volume observable, which the existing
    /// `partitions_decoded` (#953, single-candidate seek only) does not cover.
    /// For a multi-candidate `WHERE pk = ?` point read specifically: the OLD
    /// full-scan `KWayMerger::new` decodes every partition with token <= the
    /// target in every generation, so the DELTA around that call balloons far
    /// past the number of candidates holding the key; the partition-SEEKING
    /// merger (#2096) decodes only the target partition's entries per candidate,
    /// so its delta stays O(target rows). It is process-global, so a bound
    /// test's delta assertion around one call is polluted by ANY concurrent
    /// scan/compaction in the same test binary — callers must `reset()` first
    /// and serialize (`#[serial_test::serial]`) against other counter-reading
    /// tests, exactly like the other counters in this file.
    merge_run_entries_decoded: AtomicU64,
}

impl Counters {
    const fn new() -> Self {
        Self {
            sstables_scanned: AtomicU64::new(0),
            partitions_parsed: AtomicU64::new(0),
            partitions_decoded: AtomicU64::new(0),
            chunks_decompressed: AtomicU64::new(0),
            rows_decoded: AtomicU64::new(0),
            reverse_blocks_decoded: AtomicU64::new(0),
            reverse_peak_block_rows: AtomicU64::new(0),
            data_db_checksum_full_reads: AtomicU64::new(0),
            stream_walk_partitions_parsed: AtomicU64::new(0),
            merge_run_entries_decoded: AtomicU64::new(0),
        }
    }

    #[cfg(not(feature = "tombstones"))]
    fn add_sstables_scanned(&self, count: u64) {
        self.sstables_scanned.fetch_add(count, Ordering::Relaxed);
    }

    #[cfg(not(feature = "tombstones"))]
    fn add_partitions_parsed(&self, count: u64) {
        self.partitions_parsed.fetch_add(count, Ordering::Relaxed);
    }

    #[cfg(not(feature = "tombstones"))]
    fn add_partition_decoded(&self) {
        self.partitions_decoded.fetch_add(1, Ordering::Relaxed);
    }

    #[cfg(not(feature = "tombstones"))]
    fn add_chunk_decompressed(&self) {
        self.chunks_decompressed.fetch_add(1, Ordering::Relaxed);
    }

    #[cfg(not(feature = "tombstones"))]
    fn add_rows_decoded(&self, count: u64) {
        self.rows_decoded.fetch_add(count, Ordering::Relaxed);
    }

    #[cfg(not(feature = "tombstones"))]
    fn add_reverse_block_decoded(&self) {
        self.reverse_blocks_decoded.fetch_add(1, Ordering::Relaxed);
    }

    #[cfg(not(feature = "tombstones"))]
    fn observe_reverse_block_rows(&self, rows: u64) {
        self.reverse_peak_block_rows
            .fetch_max(rows, Ordering::Relaxed);
    }

    #[cfg(feature = "write-support")]
    fn add_data_db_checksum_full_read(&self) {
        self.data_db_checksum_full_reads
            .fetch_add(1, Ordering::Relaxed);
    }

    fn data_db_checksum_full_reads(&self) -> u64 {
        self.data_db_checksum_full_reads.load(Ordering::Relaxed)
    }

    fn add_stream_walk_partition_parsed(&self) {
        self.stream_walk_partitions_parsed
            .fetch_add(1, Ordering::Relaxed);
    }

    fn stream_walk_partitions_parsed(&self) -> u64 {
        self.stream_walk_partitions_parsed.load(Ordering::Relaxed)
    }

    #[cfg(feature = "write-support")]
    fn add_merge_run_entry_decoded(&self) {
        self.merge_run_entries_decoded
            .fetch_add(1, Ordering::Relaxed);
    }

    fn merge_run_entries_decoded(&self) -> u64 {
        self.merge_run_entries_decoded.load(Ordering::Relaxed)
    }

    fn reverse_blocks_decoded(&self) -> u64 {
        self.reverse_blocks_decoded.load(Ordering::Relaxed)
    }

    fn reverse_peak_block_rows(&self) -> u64 {
        self.reverse_peak_block_rows.load(Ordering::Relaxed)
    }

    fn sstables_scanned(&self) -> u64 {
        self.sstables_scanned.load(Ordering::Relaxed)
    }

    fn partitions_parsed(&self) -> u64 {
        self.partitions_parsed.load(Ordering::Relaxed)
    }

    fn partitions_decoded(&self) -> u64 {
        self.partitions_decoded.load(Ordering::Relaxed)
    }

    fn chunks_decompressed(&self) -> u64 {
        self.chunks_decompressed.load(Ordering::Relaxed)
    }

    fn rows_decoded(&self) -> u64 {
        self.rows_decoded.load(Ordering::Relaxed)
    }

    fn reset(&self) {
        self.sstables_scanned.store(0, Ordering::Relaxed);
        self.partitions_parsed.store(0, Ordering::Relaxed);
        self.partitions_decoded.store(0, Ordering::Relaxed);
        self.chunks_decompressed.store(0, Ordering::Relaxed);
        self.rows_decoded.store(0, Ordering::Relaxed);
        self.reverse_blocks_decoded.store(0, Ordering::Relaxed);
        self.reverse_peak_block_rows.store(0, Ordering::Relaxed);
        self.data_db_checksum_full_reads.store(0, Ordering::Relaxed);
        self.stream_walk_partitions_parsed
            .store(0, Ordering::Relaxed);
        self.merge_run_entries_decoded.store(0, Ordering::Relaxed);
    }
}

/// The process-global counters every read-path increment site and integration
/// probe shares. Unit tests that assert absolute values use a local
/// [`Counters`] instead (issue #1071).
static COUNTERS: Counters = Counters::new();

/// Record that `count` candidate SSTables were parsed by a partition-targeted
/// lookup. Called once per `scan_partition` invocation with the number of
/// surviving (post-prune) candidates whose `Data.db` is parsed.
///
/// Only the default (`not(tombstones)`) build has the bloom/BTI-pruning
/// `scan_partition`; the `tombstones` build serves a single-partition read by a
/// full scan + filter and has no candidate set to count, so the mutators are
/// compiled only for the build whose pruning they instrument. The getters and
/// [`reset`] remain available in every build for the test API.
#[cfg(not(feature = "tombstones"))]
pub(crate) fn add_sstables_scanned(count: u64) {
    COUNTERS.add_sstables_scanned(count);
}

/// Record that `count` partitions were returned by a partition-targeted lookup.
#[cfg(not(feature = "tombstones"))]
pub(crate) fn add_partitions_parsed(count: u64) {
    COUNTERS.add_partitions_parsed(count);
}

/// Record that one partition was DECODED from `Data.db` by the single-candidate
/// seek path (Issue #953). Called once per complete partition the seek decodes
/// at the resolved offset — exactly one for a point lookup that hits, zero for a
/// verified-absent key.
///
/// Gated on `not(tombstones)` like the other mutators: the seek path
/// (`SSTableReader::scan_single_partition`) is only reachable from the default
/// build's `scan_partition`; under `tombstones` the full-scan fallback never
/// seeks, so the counter stays at 0 and the mutator would be dead code.
#[cfg(not(feature = "tombstones"))]
pub(crate) fn add_partition_decoded() {
    COUNTERS.add_partition_decoded();
}

/// Record that one compression chunk was DECOMPRESSED by the single-candidate
/// seek path (Issue #953 / #951). Called once per chunk the seek materializes
/// into its decompressed window while buffering the target partition; the bound
/// (BIG `Index.db` size, BTI next-partition boundary) stops the loop so this
/// stays O(partition chunk span), never the file's whole chunk count.
///
/// Gated on `not(tombstones)` like the other seek-path mutators: only the
/// default build reaches the seek (`bti_pull_decompressed_chunk`); under
/// `tombstones` the full-scan fallback never seeks, so the counter stays 0.
#[cfg(not(feature = "tombstones"))]
pub(crate) fn add_chunk_decompressed() {
    COUNTERS.add_chunk_decompressed();
}

/// Record that one row was DECODED from `Data.db` within a partition by the
/// single-candidate seek path (Issue #954). Called once per row the partition
/// decoder actually parses out of the (clustering-narrowed) byte window — the
/// row-granularity signal that proves a `WHERE pk = ? AND ck </>/= ?` slice
/// query decodes O(matched rows + index block slack), not the whole partition.
///
/// Where [`add_partition_decoded`] counts WHICH partition was touched (1 for a
/// hit), this counts HOW MANY of its clustering rows were parsed: a regression
/// that reverts the clustering seek to a full-partition decode bumps this by the
/// partition's whole row count, failing the `issue_954` bound even though
/// `partitions_decoded` stays 1.
///
/// Gated on `not(tombstones)` like the other seek-path mutators: only the
/// default build reaches the seek; under `tombstones` the full-scan fallback
/// never seeks, so the counter stays 0.
#[cfg(not(feature = "tombstones"))]
pub(crate) fn add_rows_decoded(count: u64) {
    COUNTERS.add_rows_decoded(count);
}

/// Record that the BIG reverse partition iterator decoded one promoted-index
/// block (Issue #1184). Called once per block the back-to-front walk visits, so
/// a test can prove the reverse scan is block-driven (count == block count) and
/// not a post-fetch in-memory `sort_by` over a single full-partition read.
#[cfg(not(feature = "tombstones"))]
pub(crate) fn add_reverse_block_decoded() {
    COUNTERS.add_reverse_block_decoded();
}

/// Observe the row count of ONE block buffer the reverse iterator just decoded
/// (Issue #1184); keeps the running peak. A test asserts the peak stays bounded
/// to a single block, proving per-iteration memory is O(block), not O(partition).
#[cfg(not(feature = "tombstones"))]
pub(crate) fn observe_reverse_block_rows(rows: u64) {
    COUNTERS.observe_reverse_block_rows(rows);
}

/// Record one full re-read of a finished `Data.db` performed to compute
/// checksums (Issue #1663). Called once at each site that OPENS `Data.db` to
/// checksum it: the re-read CRC oracle (`crc_writer::build_crc_bytes`) and the
/// digest re-read (`SSTableWriter::compute_crc32`).
///
/// Gated on `write-support` because both increment sites live in the writer,
/// which is only compiled with that feature; the getter and [`reset`] are
/// available in every build for the test API.
#[cfg(feature = "write-support")]
pub fn add_data_db_checksum_full_read() {
    COUNTERS.add_data_db_checksum_full_read();
}

/// Record that a `do_get` compaction-merge scan read + parsed one partition body
/// (issue #2361 / #2398 / #2430). Called once per partition whose `Data.db` slice
/// is decoded, on the streaming full-index walk, the chunk-stitching walk, AND the
/// MATERIALISING full-index walk
/// ([`iterate_all_partitions_via_full_index`](crate::storage::sstable::reader::SSTableReader::iterate_all_partitions_via_full_index)).
///
/// Not gated behind any feature: these walks run on the default read path (BIG
/// SSTable, no BTI), and integration/flight tests compile against the library
/// crate without its `test` cfg, so the increment must be present in every build
/// (same rationale as the other read-work counters).
pub(crate) fn add_stream_walk_partition_parsed() {
    COUNTERS.add_stream_walk_partition_parsed();
    #[cfg(test)]
    stream_walk_scope::record();
}

/// Number of partition bodies a `do_get` scan read + parsed since the last
/// [`reset`] (Issue #2398).
///
/// For a token-range split this SHOULD stay bounded by the number of partitions
/// whose token falls in the split's `(start, end]` range, NOT the SSTable's whole
/// partition count. The current read paths filter tokens only at the consumer, so
/// today this reads the full count for any overlapping SSTable — the fixed
/// warm-scan setup cost independent of the split range and of `LIMIT` that issue
/// #2398 tracks.
///
/// This global getter is for CROSS-THREAD / integration observability (a
/// detached compaction producer thread bumping the counter, read from the test
/// thread). A same-thread *delta* assertion — "this scan drove EXACTLY N
/// partition parses" — must NOT use it: the counter is process-global, so a
/// concurrent test on another thread contaminates the delta under thread-parallel
/// `cargo test` (issue #2428). Such assertions use [`stream_walk_scope`]'s
/// thread-local [`StreamWalkScope`](stream_walk_scope::StreamWalkScope) instead,
/// which is immune by construction.
pub fn stream_walk_partitions_parsed() -> u64 {
    COUNTERS.stream_walk_partitions_parsed()
}

/// Thread-local scoping for [`stream_walk_partitions_parsed`] delta assertions
/// (issue #2428).
///
/// # Why this exists
///
/// [`stream_walk_partitions_parsed`] is a process-global `AtomicU64` shared by
/// every read-path scan and every test that drives one. A test that asserts a
/// *delta* — "my scan drove EXACTLY N partition parses" — by
/// `reset()`-ing the global, scanning, then reading the global back is
/// contaminated the moment ANY other test in the same test binary drives a scan
/// concurrently between the reset and the read: under thread-parallel
/// `cargo test --lib --all-features` (the CI "Required PR Gate" invocation, which
/// does NOT use nextest's per-process isolation) the observed delta jumps to an
/// arbitrary inflated value. `#[serial(work_counters)]` only serialises tests
/// that BOTH carry the tag, so a single untagged scan-driving test anywhere in
/// the crate reintroduces the flake — a fragile, easy-to-miss invariant across a
/// large and growing set of reachable tests.
///
/// # The structural fix
///
/// cargo runs each `#[test]` on its own OS thread, and a default (current-thread)
/// `#[tokio::test]` drives all of its `.await`s on that one thread. So a
/// thread-local counter, activated for the duration of one test's scan, records
/// ONLY the increments that execute on that test's own thread — structurally
/// immune to any concurrent test on another thread mutating the process-global
/// counter. A delta assertion opens a [`StreamWalkScope`] before its scan and
/// reads [`StreamWalkScope::count`] after; no `reset()`, no global read, no
/// serial tag, and no way for a future scan-driving test to contaminate it.
///
/// # Boundaries
///
/// The scope only sees increments on THE THREAD that opened it, so it is for
/// same-thread delta assertions of an inline (non-`spawn`) scan. A scan that
/// fans work onto detached producer threads (the compaction merge path) would
/// bump the global on those threads without touching the scope — such tests keep
/// using the global [`stream_walk_partitions_parsed`] getter. This module is
/// `#[cfg(test)]`: it exists only in the library's own test build (the binary
/// where the contamination occurs); integration tests in `tests/` compile the
/// library without its `test` cfg and never see it.
#[cfg(test)]
pub(crate) mod stream_walk_scope {
    use std::cell::Cell;

    thread_local! {
        /// `Some(count)` while a [`StreamWalkScope`] is active on this thread,
        /// `None` otherwise. Only `add_stream_walk_partition_parsed()` calls that
        /// execute on this thread bump it.
        static SCOPED: Cell<Option<u64>> = const { Cell::new(None) };
    }

    /// Bump the active scope on the current thread, if any. A no-op on threads
    /// (production scans, detached producers, other tests) with no active scope.
    pub(crate) fn record() {
        SCOPED.with(|c| {
            if let Some(v) = c.get() {
                c.set(Some(v.saturating_add(1)));
            }
        });
    }

    /// A per-thread recording scope for `stream_walk_partitions_parsed`. Open one
    /// before an inline scan whose partition-parse count you assert, and read
    /// [`count`](StreamWalkScope::count) after. Immune to concurrent tests on
    /// other threads (issue #2428). Dropping it clears the scope.
    ///
    /// Deliberately `!Send` (holds a `PhantomData<*const ()>`): the scope is
    /// meaningful only on the thread that opened it, so the type system forbids
    /// moving it to another thread where its `count()` would be wrong.
    pub(crate) struct StreamWalkScope {
        _not_send: std::marker::PhantomData<*const ()>,
    }

    impl StreamWalkScope {
        /// Begin recording on the current thread. Panics if a scope is already
        /// active on this thread (one scope per assertion; nesting unsupported).
        pub(crate) fn new() -> Self {
            SCOPED.with(|c| {
                assert!(
                    c.get().is_none(),
                    "a StreamWalkScope is already active on this thread (nesting unsupported)"
                );
                c.set(Some(0));
            });
            Self {
                _not_send: std::marker::PhantomData,
            }
        }

        /// Partition-parse increments recorded on this thread since the scope
        /// opened.
        pub(crate) fn count(&self) -> u64 {
            SCOPED.with(|c| c.get().unwrap_or(0))
        }
    }

    impl Drop for StreamWalkScope {
        fn drop(&mut self) {
            SCOPED.with(|c| c.set(None));
        }
    }
}

/// A thread-local recording scope for `MergeEntry::clone` calls, mirroring
/// [`stream_walk_scope`] exactly (issue #2428's parallel-test-pollution-immune
/// design). Used by the #1664 double-clone regression guard: the k-way merge
/// runs single-threaded, so a [`MergeEntryCloneScope`] opened around a full
/// compaction captures exactly the `MergeEntry` clones performed during it.
///
/// `#[cfg(test)]`: the `record()` call in `MergeEntry::clone` is likewise
/// `#[cfg(test)]`-gated, so production clone pays ZERO added cost. Gated on
/// `feature = "write-support"` as well because every consumer (the
/// `MergeEntry::clone` recorder and the `clone_regression_tests` guard) lives
/// in the write-support-only merge module — under the minimal feature set those
/// callers vanish, so an unconditional `#[cfg(test)]` here would be dead code
/// and trip the `-D warnings` dead-code lint in the all-compression build.
#[cfg(all(test, feature = "write-support"))]
pub(crate) mod merge_entry_clone_scope {
    use std::cell::Cell;

    thread_local! {
        /// `Some(count)` while a [`MergeEntryCloneScope`] is active on this
        /// thread, `None` otherwise. Only `MergeEntry::clone` calls executing
        /// on this thread bump it.
        static SCOPED: Cell<Option<u64>> = const { Cell::new(None) };
    }

    /// Bump the active scope on the current thread, if any. A no-op on threads
    /// (production compaction, other tests) with no active scope.
    pub(crate) fn record() {
        SCOPED.with(|c| {
            if let Some(v) = c.get() {
                c.set(Some(v.saturating_add(1)));
            }
        });
    }

    /// A per-thread recording scope for `MergeEntry::clone`. Open one before
    /// driving a `KWayMerger` compaction to completion and read
    /// [`count`](MergeEntryCloneScope::count) after. Immune to concurrent
    /// tests on other threads (issue #2428). Dropping it clears the scope.
    ///
    /// Deliberately `!Send` (holds a `PhantomData<*const ()>`): the scope is
    /// meaningful only on the thread that opened it.
    pub(crate) struct MergeEntryCloneScope {
        _not_send: std::marker::PhantomData<*const ()>,
    }

    impl MergeEntryCloneScope {
        /// Begin recording on the current thread. Panics if a scope is already
        /// active on this thread (one scope per assertion; nesting unsupported).
        pub(crate) fn new() -> Self {
            SCOPED.with(|c| {
                assert!(
                    c.get().is_none(),
                    "a MergeEntryCloneScope is already active on this thread (nesting unsupported)"
                );
                c.set(Some(0));
            });
            Self {
                _not_send: std::marker::PhantomData,
            }
        }

        /// `MergeEntry::clone` increments recorded on this thread since the
        /// scope opened.
        pub(crate) fn count(&self) -> u64 {
            SCOPED.with(|c| c.get().unwrap_or(0))
        }
    }

    impl Drop for MergeEntryCloneScope {
        fn drop(&mut self) {
            SCOPED.with(|c| c.set(None));
        }
    }
}

/// A thread-local recording scope for `CellData::clone` calls, mirroring
/// [`merge_entry_clone_scope`] exactly (issue #2428's parallel-test-pollution-
/// immune design). Used by the #1665 reconcile micro-alloc guard: reconciliation
/// runs single-threaded, so a [`CellDataCloneScope`] opened around one
/// clustering-group reconcile captures exactly the `CellData` clones performed
/// during it — enough to prove Site 2 (`filter_dropped_columns`) no longer
/// deep-clones the survivor set.
///
/// `#[cfg(test)]`: the `record()` call in `CellData::clone` is likewise
/// `#[cfg(test)]`-gated, so production clone pays ZERO added cost. Gated on
/// `feature = "write-support"` as well because `CellData` and its only consumer
/// (the `filter_dropped_columns` guard) live in the write-support-only merge
/// module — under the minimal feature set those callers vanish, so an
/// unconditional `#[cfg(test)]` here would be dead code and trip the
/// `-D warnings` dead-code lint in the all-compression build.
#[cfg(all(test, feature = "write-support"))]
pub(crate) mod cell_data_clone_scope {
    use std::cell::Cell;

    thread_local! {
        /// `Some(count)` while a [`CellDataCloneScope`] is active on this thread,
        /// `None` otherwise. Only `CellData::clone` calls executing on this
        /// thread bump it.
        static SCOPED: Cell<Option<u64>> = const { Cell::new(None) };
    }

    /// Bump the active scope on the current thread, if any. A no-op on threads
    /// (production reconcile, other tests) with no active scope.
    pub(crate) fn record() {
        SCOPED.with(|c| {
            if let Some(v) = c.get() {
                c.set(Some(v.saturating_add(1)));
            }
        });
    }

    /// A per-thread recording scope for `CellData::clone`. Open one before
    /// driving a clustering-group reconcile and read
    /// [`count`](CellDataCloneScope::count) after. Immune to concurrent tests on
    /// other threads (issue #2428). Dropping it clears the scope.
    ///
    /// Deliberately `!Send` (holds a `PhantomData<*const ()>`): the scope is
    /// meaningful only on the thread that opened it.
    pub(crate) struct CellDataCloneScope {
        _not_send: std::marker::PhantomData<*const ()>,
    }

    impl CellDataCloneScope {
        /// Begin recording on the current thread. Panics if a scope is already
        /// active on this thread (one scope per assertion; nesting unsupported).
        pub(crate) fn new() -> Self {
            SCOPED.with(|c| {
                assert!(
                    c.get().is_none(),
                    "a CellDataCloneScope is already active on this thread (nesting unsupported)"
                );
                c.set(Some(0));
            });
            Self {
                _not_send: std::marker::PhantomData,
            }
        }

        /// `CellData::clone` increments recorded on this thread since the scope
        /// opened.
        pub(crate) fn count(&self) -> u64 {
            SCOPED.with(|c| c.get().unwrap_or(0))
        }
    }

    impl Drop for CellDataCloneScope {
        fn drop(&mut self) {
            SCOPED.with(|c| c.set(None));
        }
    }
}

/// A thread-local recording scope for `range_tombstone_covers_ck` calls,
/// mirroring [`cell_data_clone_scope`] exactly (issue #2428's parallel-test-
/// pollution-immune design). Used by the #1669 range-shadowing binary-search
/// guard: range shadowing runs single-threaded, so a [`RangeCoverageScope`]
/// opened around one partition's `apply_range_shadowing` calls captures exactly
/// the range-coverage COMPARISONS performed during it.
///
/// The former linear scan invoked `range_tombstone_covers_ck` once per
/// (clustering row × coalesced range) — O(rows × ranges). The binary-search fix
/// (#1669) invokes it at most once per clustering row (only the single
/// binary-search candidate, since the coalesced ranges are sorted+disjoint per
/// partition so at most ONE covers a given `ck`). This scope makes the
/// comparison count observable so a bound test fails the moment the search
/// reverts to a full linear scan.
///
/// `#[cfg(test)]`: the `record()` call in `range_tombstone_covers_ck` is likewise
/// `#[cfg(test)]`-gated, so production coverage pays ZERO added cost. Gated on
/// `feature = "write-support"` as well because every consumer (the
/// `range_tombstone_covers_ck` recorder and the `range_shadowing_binsearch_tests`
/// guard) lives in the write-support-only merge module.
#[cfg(all(test, feature = "write-support"))]
pub(crate) mod range_coverage_scope {
    use std::cell::Cell;

    thread_local! {
        /// `Some(count)` while a [`RangeCoverageScope`] is active on this thread,
        /// `None` otherwise. Only `range_tombstone_covers_ck` calls executing on
        /// this thread bump it.
        static SCOPED: Cell<Option<u64>> = const { Cell::new(None) };
    }

    /// Bump the active scope on the current thread, if any. A no-op on threads
    /// (production shadowing, other tests) with no active scope.
    pub(crate) fn record() {
        SCOPED.with(|c| {
            if let Some(v) = c.get() {
                c.set(Some(v.saturating_add(1)));
            }
        });
    }

    /// A per-thread recording scope for `range_tombstone_covers_ck`. Open one
    /// before driving a partition's range shadowing and read
    /// [`count`](RangeCoverageScope::count) after. Immune to concurrent tests on
    /// other threads (issue #2428). Dropping it clears the scope.
    ///
    /// Deliberately `!Send` (holds a `PhantomData<*const ()>`): the scope is
    /// meaningful only on the thread that opened it.
    pub(crate) struct RangeCoverageScope {
        _not_send: std::marker::PhantomData<*const ()>,
    }

    impl RangeCoverageScope {
        /// Begin recording on the current thread. Panics if a scope is already
        /// active on this thread (one scope per assertion; nesting unsupported).
        pub(crate) fn new() -> Self {
            SCOPED.with(|c| {
                assert!(
                    c.get().is_none(),
                    "a RangeCoverageScope is already active on this thread (nesting unsupported)"
                );
                c.set(Some(0));
            });
            Self {
                _not_send: std::marker::PhantomData,
            }
        }

        /// `range_tombstone_covers_ck` increments recorded on this thread since
        /// the scope opened.
        pub(crate) fn count(&self) -> u64 {
            SCOPED.with(|c| c.get().unwrap_or(0))
        }
    }

    impl Drop for RangeCoverageScope {
        fn drop(&mut self) {
            SCOPED.with(|c| c.set(None));
        }
    }
}

/// Record that one merge ENTRY was decoded from `Data.db` by ANY
/// [`KWayMerger`](crate::storage::write_engine::KWayMerger)-adapter-driven run
/// (Issue #2096) — a full scan, a compaction, OR a multi-candidate point read,
/// whichever run this increment site's caller is executing. Called once per
/// entry a run materialises out of `Data.db`: the full-scan adapter's
/// per-`Ok(entry)` yield and the seek run's per-built-entry `PathProbe::Seeked`
/// arm. This is an ENTRY-granularity count — a partition with N clustering rows
/// bumps it N times, not once.
///
/// Gated on `write-support` because both increment sites live in the k-way
/// merge machinery, which is only compiled with that feature; the getter and
/// [`reset`] are available in every build for the test API.
#[cfg(feature = "write-support")]
pub(crate) fn add_merge_run_entry_decoded() {
    COUNTERS.add_merge_run_entry_decoded();
}

/// Number of merge entries decoded from `Data.db` by ANY `KWayMerger`-adapter-
/// driven run since the last [`reset`] (Issue #2096) — full scans, compaction,
/// and multi-candidate point reads all share this counter, so it is NOT
/// point-read-specific. It is process-global: a delta assertion around one call
/// (e.g. "did this point read stay O(target rows)?") must `reset()` first and
/// run with no concurrent scan/compaction in the same test binary — serialize
/// with `#[serial_test::serial]` against other counter-reading tests, exactly
/// like the other counters in this module.
///
/// For a multi-candidate `WHERE pk = ?` point read specifically, the delta
/// around that call stays O(target rows) once the partition-SEEKING merger is
/// wired in: only the target partition's entries per candidate are decoded. A
/// regression that reverts to the full-scan `KWayMerger::new` merge decodes
/// every partition with token <= the target in every generation, ballooning the
/// delta far past the candidate count — which the `issue_2096` bound test catches.
pub fn merge_run_entries_decoded() -> u64 {
    COUNTERS.merge_run_entries_decoded()
}

/// Number of full `Data.db` re-reads performed for checksum computation since
/// the last [`reset`] (Issue #1663).
///
/// The streaming write path accumulates the whole-file digest and the per-chunk
/// `CRC.db` values as it writes, so a non-empty `SSTableWriter::finish()` must
/// leave this at 0. A regression that reintroduces the finish-time double
/// re-read (one for `Digest.crc32`, one for `CRC.db`) bumps it to 2, failing the
/// issue-#1663 work guard.
pub fn data_db_checksum_full_reads() -> u64 {
    COUNTERS.data_db_checksum_full_reads()
}

/// Number of candidate SSTables parsed by partition-targeted lookups since the
/// last [`reset`].
///
/// Tests assert this stays O(candidates) — near 1 for a key in a single SSTable
/// (plus a small allowance for bloom false-positives) — so a regression that
/// reopens every SSTable for a single-partition read fails CI.
pub fn sstables_scanned() -> u64 {
    COUNTERS.sstables_scanned()
}

/// Number of partitions returned by partition-targeted lookups since the last
/// [`reset`].
pub fn partitions_parsed() -> u64 {
    COUNTERS.partitions_parsed()
}

/// Number of partitions DECODED from `Data.db` by the single-candidate seek path
/// since the last [`reset`] (Issue #953).
///
/// Tests assert this stays O(1) for a point lookup — a small bound, near 1 for a
/// hit — so a regression that reverts the single-candidate path to a full parse
/// (decoding every partition in the SSTable, then retaining one) fails CI.
pub fn partitions_decoded() -> u64 {
    COUNTERS.partitions_decoded()
}

/// Number of compression chunks DECOMPRESSED by the single-candidate seek path
/// since the last [`reset`] (Issue #953 / #951).
///
/// Tests assert this stays bounded by the target partition's chunk span — a
/// small constant for a point lookup — so a regression that stitches the
/// `Data.db` section to EOF (decompressing every chunk after the target,
/// including the whole tail of a large file for a head-of-file lookup) fails the
/// `issue_953` bound, even though `partitions_decoded` would still read 1.
pub fn chunks_decompressed() -> u64 {
    COUNTERS.chunks_decompressed()
}

/// Number of individual partition rows DECODED from `Data.db` by the
/// single-candidate seek path since the last [`reset`] (Issue #954).
///
/// Tests assert this stays bounded by the requested clustering slice (plus one
/// index block of block-granularity slack) for a `WHERE pk = ? AND ck </>/= ?`
/// query — proving the within-partition seek decodes O(slice), not the whole
/// partition. A regression that decodes every clustering row of the partition
/// (then post-filters) bumps this by the partition's full row count and fails
/// the bound, even though `partitions_decoded` would still read 1.
pub fn rows_decoded() -> u64 {
    COUNTERS.rows_decoded()
}

/// Number of promoted-index blocks the BIG reverse partition iterator decoded
/// back-to-front since the last [`reset`] (Issue #1184). For a wide partition
/// this equals the partition's block count; a regression to a forward
/// full-partition read + in-memory sort leaves it at 0.
pub fn reverse_blocks_decoded() -> u64 {
    COUNTERS.reverse_blocks_decoded()
}

/// Peak rows held in a single reverse-iterator block buffer since the last
/// [`reset`] (Issue #1184). Tests assert this stays bounded to one block's worth
/// of rows — far below the partition total — proving bounded per-iteration memory.
pub fn reverse_peak_block_rows() -> u64 {
    COUNTERS.reverse_peak_block_rows()
}

/// Clear all five process-global counters. Integration tests call this before a
/// query so a stale value from an earlier query cannot satisfy a later
/// assertion. Because the global is shared, an integration test that asserts on
/// it must run without a concurrent query on another thread (the integration
/// tests serialize their own setup); the in-crate unit test sidesteps this
/// entirely by asserting against a local [`Counters`] instance (issue #1071).
pub fn reset() {
    COUNTERS.reset();
}

#[cfg(all(test, not(feature = "tombstones")))]
mod tests {
    use super::*;

    // Exercises the add/get/reset contract against a *local* [`Counters`] rather
    // than the process-global instance reached through the free functions. The
    // global is shared with read-path increment sites that any concurrent test
    // in this binary can drive, so absolute-value assertions on it race
    // nondeterministically (issue #1071). A local instance is owned by this test
    // alone, so the exact-equality checks below are deterministic.
    #[test]
    fn counters_round_trip() {
        let c = Counters::new();
        c.reset();
        assert_eq!(c.sstables_scanned(), 0);
        assert_eq!(c.partitions_parsed(), 0);
        assert_eq!(c.partitions_decoded(), 0);
        assert_eq!(c.chunks_decompressed(), 0);
        assert_eq!(c.rows_decoded(), 0);
        c.add_sstables_scanned(2);
        c.add_partitions_parsed(5);
        c.add_partition_decoded();
        c.add_partition_decoded();
        c.add_chunk_decompressed();
        c.add_chunk_decompressed();
        c.add_chunk_decompressed();
        c.add_rows_decoded(7);
        assert_eq!(c.sstables_scanned(), 2);
        assert_eq!(c.partitions_parsed(), 5);
        assert_eq!(c.partitions_decoded(), 2);
        assert_eq!(c.chunks_decompressed(), 3);
        assert_eq!(c.rows_decoded(), 7);
        c.reset();
        assert_eq!(c.sstables_scanned(), 0);
        assert_eq!(c.partitions_parsed(), 0);
        assert_eq!(c.partitions_decoded(), 0);
        assert_eq!(c.chunks_decompressed(), 0);
        assert_eq!(c.rows_decoded(), 0);
    }
}

// NB: this module is gated ONLY on `test` (NOT `not(tombstones)`): the CI
// Required-PR-Gate invocation this issue exists to fix is
// `cargo test --lib --all-features`, which enables `tombstones`. The
// `add_stream_walk_partition_parsed` mutator and the `stream_walk_scope` are both
// present under every feature set, so this regression must run under all of them.
#[cfg(test)]
mod stream_walk_scope_tests {
    use super::*;

    /// Structural regression for issue #2428: a [`StreamWalkScope`] records ONLY
    /// the `add_stream_walk_partition_parsed()` increments that execute on its own
    /// thread, so a *concurrent* thread bumping the same process-global counter
    /// cannot contaminate a same-thread delta assertion.
    ///
    /// This is the mechanism that made
    /// `sequential_scan_fallback_counts_each_partition_exactly_once` flake under
    /// thread-parallel `cargo test --lib` (the CI Required-PR-Gate invocation,
    /// which does NOT isolate tests per-process like nextest): another
    /// scan-driving test running between that test's `reset()` and its read
    /// inflated the observed global delta. Here we reproduce that exact shape —
    /// a foreign thread hammering the global while a scope is open — and prove the
    /// scoped count stays exactly the number of increments made on THIS thread.
    #[test]
    fn stream_walk_scope_is_immune_to_a_concurrent_thread() {
        use super::stream_walk_scope::StreamWalkScope;
        use std::sync::mpsc;

        const LOCAL: u64 = 10; // mirrors the flaky test's N
        const FOREIGN: u64 = 500; // the contaminating "other test" load

        let scope = StreamWalkScope::new();

        // A foreign thread bumps the SAME process-global counter (its own thread
        // has no active scope, so `record()` is a no-op there). Handshakes force
        // its increments to interleave DURING this thread's scope, exactly like a
        // concurrent scan-driving test.
        let (started_tx, started_rx) = mpsc::channel::<()>();
        let (proceed_tx, proceed_rx) = mpsc::channel::<()>();
        let foreign = std::thread::spawn(move || {
            started_tx.send(()).expect("send started");
            proceed_rx.recv().expect("recv proceed");
            for _ in 0..FOREIGN {
                add_stream_walk_partition_parsed();
            }
        });

        started_rx.recv().expect("foreign thread must start");
        // Bump on THIS thread before, ...
        for _ in 0..(LOCAL / 2) {
            add_stream_walk_partition_parsed();
        }
        // ... let the foreign thread run its whole contaminating load, ...
        proceed_tx.send(()).expect("release foreign thread");
        foreign.join().expect("foreign thread must not panic");
        // ... and after. The foreign 500 landed squarely between our increments.
        for _ in 0..(LOCAL - LOCAL / 2) {
            add_stream_walk_partition_parsed();
        }

        assert_eq!(
            scope.count(),
            LOCAL,
            "the scope must count only this thread's {LOCAL} increments, never the \
             foreign thread's {FOREIGN} on the shared global (issue #2428)"
        );
        drop(scope);
        // After drop the scope is cleared: a fresh increment records nowhere.
        assert_eq!(
            StreamWalkScope::new().count(),
            0,
            "a fresh scope starts at zero (the previous scope's count did not leak)"
        );
    }
}