cqlite-core 0.13.0

//! Compression support for SSTable storage

use crate::{error::Error, Result};
use std::io::Read;
// use async_trait::async_trait; // Commented out - unused

/// Compression algorithms supported
#[derive(Debug, Clone, Copy, PartialEq, serde::Serialize, serde::Deserialize, Default)]
pub enum CompressionAlgorithm {
    /// No compression
    None,
    /// LZ4 compression (fast)
    #[default]
    Lz4,
    /// Snappy compression (balanced)
    Snappy,
    /// Deflate compression (high ratio)
    Deflate,
    /// Zstd compression (high efficiency)
    Zstd,
}

/// Maximum allowed decompressed size to prevent memory exhaustion attacks (128MB)
const MAX_DECOMPRESSED_SIZE: usize = 128 * 1024 * 1024;

impl CompressionAlgorithm {
    /// Map a recognized compressor name to its enum variant.
    ///
    /// Accepts CQLite short names (`LZ4`, `SNAPPY`, ...), Cassandra simple names
    /// (`LZ4Compressor`, `SnappyCompressor`, ...) and the explicit no-compression
    /// markers (`NONE`, `NoopCompressor`, `NoCompressor`). Returns `None` for any
    /// other (unrecognized) name — callers MUST treat `None` here as "unknown",
    /// not as "uncompressed".
    fn from_name_opt(s: &str) -> Option<Self> {
        // Strip any fully-qualified class prefix Cassandra may emit.
        let simple = s.rsplit('.').next().unwrap_or(s);
        match simple.to_uppercase().as_str() {
            "NONE" | "NOOPCOMPRESSOR" | "NOCOMPRESSOR" | "NULLCOMPRESSOR" => {
                Some(CompressionAlgorithm::None)
            }
            "LZ4" | "LZ4COMPRESSOR" => Some(CompressionAlgorithm::Lz4),
            "SNAPPY" | "SNAPPYCOMPRESSOR" => Some(CompressionAlgorithm::Snappy),
            "DEFLATE" | "DEFLATECOMPRESSOR" => Some(CompressionAlgorithm::Deflate),
            "ZSTD" | "ZSTDCOMPRESSOR" => Some(CompressionAlgorithm::Zstd),
            _ => None,
        }
    }

    /// Fallible parse of a compressor name (issue #1001).
    ///
    /// This is the path the SSTable open / `CompressionInfo.db` flow MUST use: an
    /// unrecognized name produces an explicit `UnsupportedFormat` error (including the
    /// exact offending string) rather than silently falling back to uncompressed.
    /// No content-based guessing is performed (no-heuristics mandate, issue #28).
    pub fn parse(s: &str) -> Result<Self> {
        Self::from_name_opt(s).ok_or_else(|| {
            Error::UnsupportedFormat(format!(
                "Unsupported compression algorithm '{}'. CQLite supports: \
                 LZ4Compressor, SnappyCompressor, DeflateCompressor, ZstdCompressor \
                 (or NONE for uncompressed).",
                s
            ))
        })
    }
}

impl From<String> for CompressionAlgorithm {
    fn from(s: String) -> Self {
        Self::from(s.as_str())
    }
}

impl From<&str> for CompressionAlgorithm {
    /// Infallible best-effort mapping. Unrecognized names map to `None`.
    ///
    /// WARNING: this MUST NOT be used on the SSTable read path — an unknown name here
    /// is indistinguishable from genuinely-disabled compression. Use the fallible
    /// [`CompressionAlgorithm::parse`] for any path that opens real `CompressionInfo.db`
    /// metadata (issue #1001). This `From` is retained only for the legacy header
    /// (`SSTableHeader.compression.algorithm`) path, which guards with an explicit
    /// `!= "NONE"` check before conversion and re-validates the resulting variant.
    fn from(s: &str) -> Self {
        Self::from_name_opt(s).unwrap_or(CompressionAlgorithm::None)
    }
}

/// Configuration for chunked decompression
#[derive(Debug, Clone)]
pub struct ChunkedDecompressionConfig {
    /// Maximum memory limit for decompression buffer (default: 32MB)
    pub max_memory_mb: usize,
    /// Chunk size for streaming reads (default: 1MB)
    pub chunk_size: usize,
    /// Maximum decompressed output size to prevent memory bombs
    pub max_output_size: usize,
}

impl Default for ChunkedDecompressionConfig {
    fn default() -> Self {
        Self {
            max_memory_mb: 32,                  // 32MB limit, well below 64MB
            chunk_size: 1024 * 1024,            // 1MB chunks
            max_output_size: 128 * 1024 * 1024, // 128MB max output to be conservative
        }
    }
}

/// Streaming decompression context for handling large blocks
pub struct StreamingDecompressor {
    algorithm: CompressionAlgorithm,
    config: ChunkedDecompressionConfig,
    bytes_processed: usize,
    bytes_output: usize,
}

/// Validates that decompressed size does not exceed safety limits
///
/// # Security
/// Prevents decompression bomb attacks by rejecting sizes > 128MB
fn validate_decompression_size(uncompressed_size: usize) -> Result<()> {
    if uncompressed_size > MAX_DECOMPRESSED_SIZE {
        return Err(Error::storage(format!(
            "Decompression bomb protection: size {} exceeds limit {} (128MB)",
            uncompressed_size, MAX_DECOMPRESSED_SIZE
        )));
    }
    Ok(())
}

/// Decode a RAW Snappy block (Cassandra 5.0 `SnappyCompressor`: no length prefix)
/// in EXACTLY one attempt.
///
/// The authoritative CompressionInfo.db algorithm determines the single format —
/// no framed-then-raw guessing (no-heuristics mandate #28, issue #1588). A guess
/// could silently mis-decode an adversarial chunk to wrong bytes; strict raw
/// decoding surfaces a typed error instead.
///
/// `decode_attempts` is incremented once per decode call. Production passes a
/// throwaway `&mut 0`; tests thread a real counter to assert a single attempt.
#[cfg(feature = "snappy")]
fn snappy_decompress_raw(data: &[u8], decode_attempts: &mut usize) -> Result<Vec<u8>> {
    use snap::raw::Decoder;
    *decode_attempts += 1;

    // CENTRALIZED bomb guard (issue #1588): a raw Snappy block carries its
    // decompressed length as a leading varint. Inspect it FIRST and reject an
    // over-limit block WITHOUT calling `decompress_vec` — `decompress_vec`
    // pre-allocates `decompress_len` bytes up front, so an adversarial block
    // declaring a huge size would allocate before any post-decode guard runs.
    // This single choke point protects EVERY caller (chunk decode + streaming).
    let advertised = snap::raw::decompress_len(data)
        .map_err(|e| Error::storage(format!("Snappy (raw) length decode failed: {}", e)))?;
    if advertised > MAX_DECOMPRESSED_SIZE {
        return Err(Error::storage(format!(
            "Decompression bomb protection: advertised size {} exceeds limit {} (128MB)",
            advertised, MAX_DECOMPRESSED_SIZE
        )));
    }

    let mut decoder = Decoder::new();
    let decompressed = decoder
        .decompress_vec(data)
        .map_err(|e| Error::storage(format!("Snappy (raw) decompression failed: {}", e)))?;
    // Belt-and-suspenders: the advertised length is attacker-controlled, so
    // re-check the ACTUAL decoded size against the hard cap.
    if decompressed.len() > MAX_DECOMPRESSED_SIZE {
        return Err(Error::storage(format!(
            "Decompression bomb protection: decompressed size {} exceeds limit {} (128MB)",
            decompressed.len(),
            MAX_DECOMPRESSED_SIZE
        )));
    }
    Ok(decompressed)
}

/// Compression handler
pub struct Compression {
    algorithm: CompressionAlgorithm,
}

impl Compression {
    /// Create a new compression handler
    pub fn new(algorithm: CompressionAlgorithm) -> Result<Self> {
        Ok(Self { algorithm })
    }

    /// Compress data with Cassandra-compatible parameters
    pub fn compress(&self, data: &[u8]) -> Result<Vec<u8>> {
        match self.algorithm {
            CompressionAlgorithm::None => Ok(data.to_vec()),
            CompressionAlgorithm::Lz4 => {
                #[cfg(feature = "lz4")]
                {
                    // Use Cassandra-compatible LZ4 compression
                    use lz4_flex::compress_prepend_size;

                    // Cassandra uses LZ4 frame format with specific parameters
                    let compressed = compress_prepend_size(data);
                    Ok(compressed)
                }
                #[cfg(not(feature = "lz4"))]
                {
                    Err(Error::storage("LZ4 compression not available".to_string()))
                }
            }
            CompressionAlgorithm::Snappy => {
                #[cfg(feature = "snappy")]
                {
                    use snap::raw::Encoder;

                    // Cassandra 5.0's SnappyCompressor writes a RAW Snappy block with NO
                    // length prefix (see writer::compressed_data_writer::SnappyCompressor
                    // and org.apache.cassandra.io.compress.SnappyCompressor). Emit raw so
                    // it round-trips through the strict raw decode path below (#1588). A
                    // 4-byte size prefix was NEVER a Cassandra-compatible format.
                    let mut encoder = Encoder::new();
                    encoder
                        .compress_vec(data)
                        .map_err(|e| Error::storage(format!("Snappy compression failed: {}", e)))
                }
                #[cfg(not(feature = "snappy"))]
                {
                    Err(Error::storage(
                        "Snappy compression not available".to_string(),
                    ))
                }
            }
            CompressionAlgorithm::Deflate => {
                #[cfg(feature = "deflate")]
                {
                    use flate2::write::ZlibEncoder;
                    use flate2::Compression as DeflateCompression;
                    use std::io::Write;

                    // Cassandra's DeflateCompressor uses java.util.zip.Deflater, which
                    // emits a ZLIB-wrapped stream (2-byte header + DEFLATE body +
                    // Adler-32 trailer) with NO 4-byte size prefix. Match it exactly so
                    // the output reads back through the zlib-aware decode path. (#1082)
                    let mut encoder = ZlibEncoder::new(Vec::new(), DeflateCompression::new(6));
                    encoder.write_all(data).map_err(|e| {
                        Error::storage(format!("Deflate compression failed: {}", e))
                    })?;
                    encoder
                        .finish()
                        .map_err(|e| Error::storage(format!("Deflate finish failed: {}", e)))
                }
                #[cfg(not(feature = "deflate"))]
                {
                    Err(Error::storage(
                        "Deflate compression not available".to_string(),
                    ))
                }
            }
            CompressionAlgorithm::Zstd => {
                #[cfg(feature = "zstd")]
                {
                    use zstd::stream::encode_all;

                    // Cassandra's ZstdCompressor writes a BARE zstd frame with NO
                    // 4-byte size prefix. Match it so the output reads back through
                    // the bare-frame decode path (#1082).
                    encode_all(data, 3)
                        .map_err(|e| Error::storage(format!("Zstd compression failed: {}", e)))
                }
                #[cfg(not(feature = "zstd"))]
                {
                    Err(Error::storage("Zstd compression not available".to_string()))
                }
            }
        }
    }

    /// Create a streaming decompressor for large blocks
    pub fn create_streaming_decompressor(
        &self,
        config: ChunkedDecompressionConfig,
    ) -> StreamingDecompressor {
        StreamingDecompressor {
            algorithm: self.algorithm,
            config,
            bytes_processed: 0,
            bytes_output: 0,
        }
    }

    /// Decompress data using traditional method (for small blocks)
    pub fn decompress(&self, data: &[u8]) -> Result<Vec<u8>> {
        // A5 read-work counter (DECOMPRESS_CALLS; consumers B1/E3): one per chunk
        // decompress. This is the single choke point every compressed-chunk read
        // path funnels through. No-op in release (design.md Decision 1/2).
        crate::storage::sstable::read_work_counters::record_decompress();
        match self.algorithm {
            CompressionAlgorithm::None => Ok(data.to_vec()),
            CompressionAlgorithm::Lz4 => {
                #[cfg(feature = "lz4")]
                {
                    use lz4_flex::decompress_size_prepended;

                    // LZ4 format: 4-byte size prefix (little-endian) + compressed data
                    if data.len() < 4 {
                        return Err(Error::storage("Invalid LZ4 data: too short".to_string()));
                    }

                    // Extract uncompressed size (4 bytes, little-endian for LZ4)
                    let uncompressed_size =
                        u32::from_le_bytes([data[0], data[1], data[2], data[3]]) as usize;

                    // Validate size to prevent decompression bombs
                    // SECURITY: lz4_flex::decompress_size_prepended does NOT validate the size
                    // prefix before allocating memory, making it vulnerable to memory exhaustion
                    // attacks if a malicious file contains an excessively large size value.
                    validate_decompression_size(uncompressed_size)?;

                    // Decompress using library function (now safe after validation)
                    decompress_size_prepended(data)
                        .map_err(|e| Error::storage(format!("LZ4 decompression failed: {}", e)))
                }
                #[cfg(not(feature = "lz4"))]
                {
                    Err(Error::storage("LZ4 compression not available".to_string()))
                }
            }
            CompressionAlgorithm::Snappy => {
                #[cfg(feature = "snappy")]
                {
                    // Decode EXACTLY one format (raw Snappy), determined by the
                    // authoritative CompressionInfo.db algorithm — never by trying
                    // framed-then-raw and keeping whichever "succeeds" (no-heuristics
                    // mandate #28, issue #1588). The framed-guess could silently return
                    // wrong bytes for an adversarial chunk; strict raw decoding rejects
                    // it. `&mut 0` discards the (test-only) attempt counter.
                    snappy_decompress_raw(data, &mut 0)
                }
                #[cfg(not(feature = "snappy"))]
                {
                    Err(Error::storage(
                        "Snappy compression not available".to_string(),
                    ))
                }
            }
            CompressionAlgorithm::Deflate => {
                #[cfg(feature = "deflate")]
                {
                    use flate2::read::ZlibDecoder;
                    use std::io::Read;

                    // Cassandra's DeflateCompressor uses java.util.zip.Deflater/Inflater,
                    // which emit ZLIB-wrapped streams: a 2-byte header (0x78 0x9c) +
                    // DEFLATE body + 4-byte Adler-32 trailer. There is NO 4-byte
                    // uncompressed-size prefix (that is an LZ4/Zstd convention) and the
                    // body is NOT raw DEFLATE. Decode with ZlibDecoder. (#1082)
                    if data.is_empty() {
                        return Err(Error::storage(
                            "Invalid Deflate data: empty chunk".to_string(),
                        ));
                    }

                    // Decompression-bomb guard: the decoder reads into a growing Vec,
                    // so we cap the output length rather than trusting any in-stream
                    // size field (none exists for zlib). The caller (chunk reader)
                    // separately bounds chunks by CompressionInfo.db lengths.
                    let mut decoder = ZlibDecoder::new(data).take(MAX_DECOMPRESSED_SIZE as u64 + 1);
                    let mut decompressed = Vec::new();
                    decoder.read_to_end(&mut decompressed).map_err(|e| {
                        Error::storage(format!("Deflate decompression failed: {}", e))
                    })?;

                    if decompressed.len() > MAX_DECOMPRESSED_SIZE {
                        return Err(Error::storage(format!(
                            "Decompression bomb protection: Deflate output exceeds limit {} (128MB)",
                            MAX_DECOMPRESSED_SIZE
                        )));
                    }

                    Ok(decompressed)
                }
                #[cfg(not(feature = "deflate"))]
                {
                    Err(Error::storage(
                        "Deflate compression not available".to_string(),
                    ))
                }
            }
            CompressionAlgorithm::Zstd => {
                #[cfg(feature = "zstd")]
                {
                    use std::io::Read;
                    use zstd::stream::read::Decoder as ZstdDecoder;

                    // Cassandra's ZstdCompressor writes a BARE zstd frame (magic
                    // 0x28 0xB5 0x2F 0xFD ...) with NO 4-byte uncompressed-size
                    // prefix — the same as the chunk-targeted decode path
                    // (chunk_decompressor.rs::decompress_zstd_chunk). The previous
                    // 4-byte-prefix assumption mis-read the frame magic as a ~650MB
                    // size and tripped the bomb guard on every stitched scan (#1082,
                    // same root cause as the Deflate fix). Decode the frame directly
                    // and bound the OUTPUT length instead of trusting an in-stream
                    // size field.
                    if data.is_empty() {
                        return Err(Error::storage("Invalid Zstd data: empty chunk".to_string()));
                    }

                    // Decompression-bomb guard: stream through a capped reader so a
                    // small malicious frame cannot allocate past the limit BEFORE we
                    // check the length (mirrors the Deflate path). A zstd frame can
                    // declare a huge content size, so `decode_all` would pre-allocate
                    // it up front — `Read::take` bounds the work instead.
                    let mut decoder = ZstdDecoder::new(data)
                        .map_err(|e| Error::storage(format!("Zstd decoder init failed: {}", e)))?
                        .take(MAX_DECOMPRESSED_SIZE as u64 + 1);
                    let mut decompressed = Vec::new();
                    decoder
                        .read_to_end(&mut decompressed)
                        .map_err(|e| Error::storage(format!("Zstd decompression failed: {}", e)))?;

                    if decompressed.len() > MAX_DECOMPRESSED_SIZE {
                        return Err(Error::storage(format!(
                            "Decompression bomb protection: Zstd output exceeds limit {} (128MB)",
                            MAX_DECOMPRESSED_SIZE
                        )));
                    }

                    Ok(decompressed)
                }
                #[cfg(not(feature = "zstd"))]
                {
                    Err(Error::storage("Zstd compression not available".to_string()))
                }
            }
        }
    }

    /// Get compression algorithm
    pub fn algorithm(&self) -> &CompressionAlgorithm {
        &self.algorithm
    }

    /// Check if we should use streaming decompression based on size
    pub fn should_use_streaming(
        &self,
        compressed_size: usize,
        config: &ChunkedDecompressionConfig,
    ) -> bool {
        compressed_size > config.max_memory_mb * 1024 * 1024 / 4 // Use streaming if compressed > 1/4 of memory limit
    }
}

impl StreamingDecompressor {
    /// Decompress data in chunks with memory limit enforcement
    pub async fn decompress_streaming<R: Read + Send>(
        &mut self,
        reader: R,
        expected_size: Option<usize>,
    ) -> Result<Vec<u8>> {
        let memory_limit_bytes = self.config.max_memory_mb * 1024 * 1024;

        // Pre-allocate output buffer if we know the expected size
        let mut output = if let Some(size) = expected_size {
            if size > self.config.max_output_size {
                return Err(Error::storage(format!(
                    "Expected decompressed size {} exceeds limit {}",
                    size, self.config.max_output_size
                )));
            }
            Vec::with_capacity(size.min(memory_limit_bytes / 2))
        } else {
            Vec::with_capacity(self.config.chunk_size)
        };

        match self.algorithm {
            CompressionAlgorithm::None => {
                // For uncompressed data, just copy in chunks
                self.copy_chunks_with_limit(reader, &mut output, memory_limit_bytes)
                    .await?;
            }
            CompressionAlgorithm::Lz4 => {
                self.decompress_lz4_streaming(reader, &mut output, memory_limit_bytes)
                    .await?;
            }
            CompressionAlgorithm::Snappy => {
                self.decompress_snappy_streaming(reader, &mut output, memory_limit_bytes)
                    .await?;
            }
            CompressionAlgorithm::Deflate => {
                self.decompress_deflate_streaming(reader, &mut output, memory_limit_bytes)
                    .await?;
            }
            CompressionAlgorithm::Zstd => {
                self.decompress_zstd_streaming(reader, &mut output, memory_limit_bytes)
                    .await?;
            }
        }

        self.bytes_output = output.len();
        Ok(output)
    }

    /// Copy uncompressed data in chunks
    async fn copy_chunks_with_limit<R: Read>(
        &mut self,
        mut reader: R,
        output: &mut Vec<u8>,
        memory_limit: usize,
    ) -> Result<()> {
        let mut buffer = vec![0u8; self.config.chunk_size];

        loop {
            let bytes_read = reader
                .read(&mut buffer)
                .map_err(|e| Error::storage(format!("Failed to read chunk: {}", e)))?;

            if bytes_read == 0 {
                break; // EOF
            }

            // Check memory limits
            if output.len() + bytes_read > memory_limit {
                return Err(Error::storage(format!(
                    "Memory limit exceeded: {} bytes (limit: {} bytes)",
                    output.len() + bytes_read,
                    memory_limit
                )));
            }

            output.extend_from_slice(&buffer[..bytes_read]);
            self.bytes_processed += bytes_read;

            // Yield control periodically for large operations
            if self.bytes_processed % (8 * 1024 * 1024) == 0 {
                tokio::task::yield_now().await;
            }
        }

        Ok(())
    }

    /// Streaming LZ4 decompression with proper frame handling
    async fn decompress_lz4_streaming<R: Read>(
        &mut self,
        reader: R,
        output: &mut Vec<u8>,
        memory_limit: usize,
    ) -> Result<()> {
        #[cfg(feature = "lz4")]
        {
            // For LZ4, we need to handle the size-prepended format used by Cassandra
            let mut buf_reader = std::io::BufReader::new(reader);
            let mut size_bytes = [0u8; 4];
            use std::io::Read;

            buf_reader
                .read_exact(&mut size_bytes)
                .map_err(|e| Error::storage(format!("Failed to read LZ4 size header: {}", e)))?;

            let expected_size = u32::from_le_bytes(size_bytes) as usize;

            if expected_size > memory_limit {
                return Err(Error::storage(format!(
                    "LZ4 expected size {} exceeds memory limit {}",
                    expected_size, memory_limit
                )));
            }

            // Read compressed data in chunks and decompress
            let mut compressed_buffer = Vec::new();
            let mut chunk_buffer = vec![0u8; self.config.chunk_size];

            loop {
                let bytes_read = buf_reader.read(&mut chunk_buffer).map_err(|e| {
                    Error::storage(format!("Failed to read LZ4 compressed chunk: {}", e))
                })?;

                if bytes_read == 0 {
                    break;
                }

                compressed_buffer.extend_from_slice(&chunk_buffer[..bytes_read]);
                self.bytes_processed += bytes_read;

                // Yield control periodically
                if self.bytes_processed % (4 * 1024 * 1024) == 0 {
                    tokio::task::yield_now().await;
                }
            }

            // Decompress the complete buffer
            use lz4_flex::decompress;
            let decompressed = decompress(&compressed_buffer, expected_size)
                .map_err(|e| Error::storage(format!("LZ4 decompression failed: {}", e)))?;

            output.extend_from_slice(&decompressed);
            Ok(())
        }
        #[cfg(not(feature = "lz4"))]
        {
            Err(Error::storage("LZ4 compression not available".to_string()))
        }
    }

    /// Streaming Snappy decompression
    async fn decompress_snappy_streaming<R: Read>(
        &mut self,
        reader: R,
        output: &mut Vec<u8>,
        memory_limit: usize,
    ) -> Result<()> {
        #[cfg(feature = "snappy")]
        {
            use std::io::BufReader;

            // Cassandra 5.0's `SnappyCompressor` emits a RAW Snappy block (no
            // stream framing, no length prefix) — the SAME single authoritative
            // format that `compress` and the chunk `decompress` path use
            // (no-heuristics, issue #1588; this closes #1862). The previous
            // `snap::read::FrameDecoder` decoded the DIFFERENT *framed* Snappy
            // format, so the public streaming decompressor could not read bytes
            // produced by `CompressionAlgorithm::Snappy`. Raw Snappy is not
            // self-delimiting and carries no length prefix, so the whole
            // compressed block must be read before it can be decoded — decode it
            // through the same `snappy_decompress_raw` helper as the chunk path.
            //
            // SECURITY (issue #1588): bound BOTH allocations before they happen so
            // a huge/malicious reader cannot exceed the streaming memory budget or
            // OOM before the guards run:
            //  1. Cap the COMPRESSED read. Any Snappy block that legitimately
            //     decodes to <= `memory_limit` bytes cannot be larger than the
            //     maximum Snappy encoding of `memory_limit` bytes
            //     (`snap::raw::max_compress_len`). A larger input cannot produce
            //     in-budget output, so we refuse to buffer it (read `cap + 1` via
            //     `take` to detect overrun without reading unbounded input).
            //  2. Reject a decompression bomb using the advertised uncompressed
            //     length (`snap::raw::decompress_len`, the raw block's varint
            //     prefix) BEFORE allocating the output buffer.
            let max_compressed = snap::raw::max_compress_len(memory_limit);
            if max_compressed == 0 {
                return Err(Error::storage(format!(
                    "Snappy streaming memory limit {} too large to bound compressed input",
                    memory_limit
                )));
            }
            let read_cap = max_compressed
                .checked_add(1)
                .ok_or_else(|| Error::storage("Snappy compressed read cap overflow".to_string()))?;
            let mut buf_reader = BufReader::new(reader).take(read_cap as u64);
            let mut compressed = Vec::new();
            buf_reader.read_to_end(&mut compressed).map_err(|e| {
                Error::storage(format!("Failed to read Snappy compressed data: {}", e))
            })?;
            if compressed.len() > max_compressed {
                return Err(Error::storage(format!(
                    "Snappy compressed input exceeds bound {} bytes (memory limit: {} bytes)",
                    max_compressed, memory_limit
                )));
            }
            self.bytes_processed += compressed.len();

            // Pre-allocation bomb guard: reject before allocating the output buffer.
            // Bound by the MINIMUM of every relevant limit — the streaming memory
            // budget AND the configured output cap (issue #1588). A `max_memory_mb`
            // set above `max_output_size` must not be allowed to allocate past the
            // intended output cap. `snappy_decompress_raw` additionally enforces the
            // hard `MAX_DECOMPRESSED_SIZE` ceiling at the shared choke point.
            let advertised = snap::raw::decompress_len(&compressed)
                .map_err(|e| Error::storage(format!("Snappy (raw) length decode failed: {}", e)))?;
            let effective_limit = memory_limit.min(self.config.max_output_size);
            let projected = output.len().checked_add(advertised);
            if projected.is_none_or(|total| total > effective_limit) {
                return Err(Error::storage(format!(
                    "Decompression bomb protection: advertised Snappy size {} exceeds limit {} bytes",
                    advertised, effective_limit
                )));
            }

            // Belt-and-suspenders: enforce against the ACTUAL decoded size (the
            // advertised length is attacker-controlled); `snappy_decompress_raw`
            // additionally caps at MAX_DECOMPRESSED_SIZE.
            let decompressed = snappy_decompress_raw(&compressed, &mut 0)?;

            if output.len() + decompressed.len() > memory_limit {
                return Err(Error::storage(format!(
                    "Memory limit exceeded during Snappy decompression: {} bytes (limit: {} bytes)",
                    output.len() + decompressed.len(),
                    memory_limit
                )));
            }

            output.extend_from_slice(&decompressed);
            // Yield once after a potentially large decode so we do not starve the
            // runtime (the read+decode above is a single bounded operation).
            tokio::task::yield_now().await;

            Ok(())
        }
        #[cfg(not(feature = "snappy"))]
        {
            let _ = (reader, output, memory_limit);
            Err(Error::storage(
                "Snappy compression not available".to_string(),
            ))
        }
    }

    /// Streaming Deflate decompression
    #[allow(clippy::ptr_arg)] // output.extend_from_slice() requires &mut Vec<u8>
    async fn decompress_deflate_streaming<R: Read>(
        &mut self,
        #[cfg_attr(not(feature = "deflate"), allow(unused_variables))] reader: R,
        #[cfg_attr(not(feature = "deflate"), allow(unused_variables))] output: &mut Vec<u8>,
        #[cfg_attr(not(feature = "deflate"), allow(unused_variables))] memory_limit: usize,
    ) -> Result<()> {
        #[cfg(feature = "deflate")]
        {
            use flate2::read::ZlibDecoder;
            use std::io::BufReader;

            // Cassandra's DeflateCompressor emits ZLIB-wrapped streams (header
            // 0x78 0x9c + DEFLATE body + Adler-32 trailer), NOT raw DEFLATE and
            // with NO 4-byte size prefix. Decode with ZlibDecoder. (#1082)
            let buf_reader = BufReader::new(reader);
            let mut decoder = ZlibDecoder::new(buf_reader);
            let mut chunk_buffer = vec![0u8; self.config.chunk_size];

            loop {
                let bytes_read = decoder.read(&mut chunk_buffer).map_err(|e| {
                    Error::storage(format!("Deflate streaming decompression failed: {}", e))
                })?;

                if bytes_read == 0 {
                    break; // EOF
                }

                // Check memory limits
                if output.len() + bytes_read > memory_limit {
                    return Err(Error::storage(format!(
                        "Memory limit exceeded during Deflate decompression: {} bytes (limit: {} bytes)",
                        output.len() + bytes_read,
                        memory_limit
                    )));
                }

                output.extend_from_slice(&chunk_buffer[..bytes_read]);
                self.bytes_processed += bytes_read;

                // Yield control for large operations
                if self.bytes_processed % (4 * 1024 * 1024) == 0 {
                    tokio::task::yield_now().await;
                }
            }

            Ok(())
        }
        #[cfg(not(feature = "deflate"))]
        {
            Err(Error::storage(
                "Deflate compression not available".to_string(),
            ))
        }
    }

    /// Streaming Zstd decompression
    #[allow(clippy::ptr_arg)] // output.extend_from_slice() requires &mut Vec<u8>
    async fn decompress_zstd_streaming<R: Read>(
        &mut self,
        #[cfg_attr(not(feature = "zstd"), allow(unused_variables))] reader: R,
        #[cfg_attr(not(feature = "zstd"), allow(unused_variables))] output: &mut Vec<u8>,
        #[cfg_attr(not(feature = "zstd"), allow(unused_variables))] memory_limit: usize,
    ) -> Result<()> {
        #[cfg(feature = "zstd")]
        {
            use std::io::BufReader;

            let buf_reader = BufReader::new(reader);
            let mut decoder = zstd::stream::read::Decoder::new(buf_reader)
                .map_err(|e| Error::storage(format!("Failed to create Zstd decoder: {}", e)))?;
            let mut chunk_buffer = vec![0u8; self.config.chunk_size];

            loop {
                let bytes_read = decoder.read(&mut chunk_buffer).map_err(|e| {
                    Error::storage(format!("Zstd streaming decompression failed: {}", e))
                })?;

                if bytes_read == 0 {
                    break; // EOF
                }

                // Check memory limits
                if output.len() + bytes_read > memory_limit {
                    return Err(Error::storage(format!(
                        "Memory limit exceeded during Zstd decompression: {} bytes (limit: {} bytes)",
                        output.len() + bytes_read,
                        memory_limit
                    )));
                }

                output.extend_from_slice(&chunk_buffer[..bytes_read]);
                self.bytes_processed += bytes_read;

                // Yield control for large operations
                if self.bytes_processed % (4 * 1024 * 1024) == 0 {
                    tokio::task::yield_now().await;
                }
            }

            Ok(())
        }
        #[cfg(not(feature = "zstd"))]
        {
            Err(Error::storage("Zstd compression not available".to_string()))
        }
    }

    /// Get decompression statistics
    pub fn stats(&self) -> (usize, usize) {
        (self.bytes_processed, self.bytes_output)
    }

    /// Reset decompressor state for reuse
    pub fn reset(&mut self) {
        self.bytes_processed = 0;
        self.bytes_output = 0;
    }

    /// Get compression ratio estimate
    pub fn estimated_ratio(&self) -> f64 {
        match self.algorithm {
            CompressionAlgorithm::None => 1.0,
            CompressionAlgorithm::Lz4 => 0.6,    // ~40% compression
            CompressionAlgorithm::Snappy => 0.5, // ~50% compression
            CompressionAlgorithm::Deflate => 0.3, // ~70% compression
            CompressionAlgorithm::Zstd => 0.25,  // ~75% compression
        }
    }

    /// Select optimal compression algorithm based on data characteristics
    pub fn select_optimal_algorithm(
        data_sample: &[u8],
        performance_priority: CompressionPriority,
    ) -> CompressionAlgorithm {
        // Analyze data characteristics
        let entropy = calculate_entropy(data_sample);
        let repetition_score = calculate_repetition_score(data_sample);
        let data_size = data_sample.len();

        match performance_priority {
            CompressionPriority::Speed => {
                // Prioritize speed over compression ratio
                if entropy > 0.9 {
                    CompressionAlgorithm::None // High entropy data doesn't compress well
                } else {
                    CompressionAlgorithm::Lz4 // Fast compression
                }
            }
            CompressionPriority::Balanced => {
                // Balance speed and compression ratio
                if entropy > 0.95 {
                    CompressionAlgorithm::None
                } else if repetition_score > 0.7 || data_size > 1024 * 1024 {
                    CompressionAlgorithm::Snappy // Good balance for large or repetitive data
                } else {
                    CompressionAlgorithm::Lz4
                }
            }
            CompressionPriority::Ratio => {
                // Prioritize compression ratio
                if entropy > 0.98 {
                    CompressionAlgorithm::None
                } else if repetition_score > 0.5 {
                    CompressionAlgorithm::Deflate // Best compression for repetitive data
                } else {
                    CompressionAlgorithm::Snappy
                }
            }
        }
    }
}

/// Compression priority for algorithm selection
#[derive(Debug, Clone, Copy, PartialEq)]
pub enum CompressionPriority {
    /// Prioritize compression/decompression speed
    Speed,
    /// Balance speed and compression ratio
    Balanced,
    /// Prioritize maximum compression ratio
    Ratio,
}

/// Compression statistics
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct CompressionStats {
    /// Original size in bytes
    pub original_size: u64,

    /// Compressed size in bytes
    pub compressed_size: u64,

    /// Compression ratio (compressed / original)
    pub ratio: f64,

    /// Compression algorithm used
    pub algorithm: CompressionAlgorithm,
}

impl CompressionStats {
    /// Calculate compression statistics
    pub fn calculate(
        original_size: u64,
        compressed_size: u64,
        algorithm: CompressionAlgorithm,
    ) -> Self {
        let ratio = if original_size > 0 {
            compressed_size as f64 / original_size as f64
        } else {
            1.0
        };

        Self {
            original_size,
            compressed_size,
            ratio,
            algorithm,
        }
    }

    /// Get space saved in bytes
    pub fn space_saved(&self) -> u64 {
        self.original_size.saturating_sub(self.compressed_size)
    }

    /// Get compression percentage
    pub fn compression_percentage(&self) -> f64 {
        (1.0 - self.ratio) * 100.0
    }
}

/// Calculate entropy of data sample (0.0 = no entropy, 1.0 = maximum entropy)
fn calculate_entropy(data: &[u8]) -> f64 {
    if data.is_empty() {
        return 0.0;
    }

    let mut counts = [0u32; 256];
    for &byte in data {
        counts[byte as usize] += 1;
    }

    let total = data.len() as f64;
    let mut entropy = 0.0;

    for &count in &counts {
        if count > 0 {
            let probability = count as f64 / total;
            entropy -= probability * probability.log2();
        }
    }

    // Normalize to 0.0-1.0 range
    entropy / 8.0 // 8 bits per byte
}

/// Calculate repetition score (0.0 = no repetition, 1.0 = highly repetitive)
fn calculate_repetition_score(data: &[u8]) -> f64 {
    if data.len() < 4 {
        return 0.0;
    }

    let mut repeated_bytes = 0;
    let mut pattern_matches = 0;

    // Check for byte repetitions
    for i in 1..data.len() {
        if data[i] == data[i - 1] {
            repeated_bytes += 1;
        }
    }

    // Check for 2-byte pattern repetitions
    // Need at least 4 bytes to check 2-byte patterns (i-3 must be valid)
    // Starting at i=3 prevents arithmetic underflow when accessing data[i-3]
    for i in 3..data.len() {
        if data[i] == data[i - 2] && data[i - 1] == data[i - 3] {
            pattern_matches += 1;
        }
    }

    let byte_repetition_score = repeated_bytes as f64 / (data.len() - 1) as f64;
    let pattern_repetition_score = if data.len() > 3 {
        pattern_matches as f64 / (data.len() - 3) as f64
    } else {
        0.0
    };

    // Combine scores with weights
    (byte_repetition_score * 0.6 + pattern_repetition_score * 0.4).min(1.0)
}

/// Normalize Cassandra compression algorithm names to standard names
fn normalize_algorithm_name(raw_name: &str) -> String {
    match raw_name {
        "LZ4Compressor" => "LZ4".to_string(),
        "SnappyCompressor" => "SNAPPY".to_string(),
        "DeflateCompressor" => "DEFLATE".to_string(),
        "ZstdCompressor" => "ZSTD".to_string(),
        "NoCompressor" | "NullCompressor" => "NONE".to_string(),
        // If it's already normalized or unknown, return as-is
        other => other.to_string(),
    }
}

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

    #[test]
    fn test_no_compression() {
        let compression = Compression::new(CompressionAlgorithm::None).unwrap();
        let data = b"hello world";

        let compressed = compression.compress(data).unwrap();
        assert_eq!(compressed, data);

        let decompressed = compression.decompress(&compressed).unwrap();
        assert_eq!(decompressed, data);
    }

    #[test]
    fn test_compression_stats() {
        let stats = CompressionStats::calculate(1000, 600, CompressionAlgorithm::Lz4);

        assert_eq!(stats.original_size, 1000);
        assert_eq!(stats.compressed_size, 600);
        assert_eq!(stats.ratio, 0.6);
        assert_eq!(stats.space_saved(), 400);
        assert_eq!(stats.compression_percentage(), 40.0);
    }

    // Note: Test methods temporarily disabled due to compilation issues
    // The functionality is tested via integration tests

    /// Adversarial oracle (issue #1588, decision #14): a chunk whose leading 4
    /// bytes ALSO parse as a plausible framed big-endian length header, followed
    /// by a valid RAW-snappy body of exactly that many output bytes.
    ///
    /// Under the (deleted) framed-then-raw guessing, `decompress` read the 4-byte
    /// BE prefix `S`, decoded `data[4..]` as raw snappy to `P_wrong` (whose length
    /// equals `S`), and RETURNED those bytes — silently wrong. The authoritative
    /// Cassandra 5.0 format for a `SnappyCompressor` chunk is RAW snappy with NO
    /// length prefix, so strict raw decoding of the WHOLE chunk must NOT return
    /// `P_wrong` (it rejects the malformed leading zero-varint stream). This is the
    /// no-heuristics enforcement: decode exactly one format determined by metadata.
    #[cfg(feature = "snappy")]
    #[test]
    fn test_snappy_decode_is_strict_raw_only_no_format_guessing() {
        use snap::raw::Encoder;
        let p_wrong = b"WRONG-framed-decode-abcdefghijklmnopqrstuvwxyz".to_vec();
        let s = p_wrong.len() as u32;
        let mut enc = Encoder::new();
        let inner = enc.compress_vec(&p_wrong).unwrap(); // valid raw snappy -> p_wrong
        let mut adversarial = s.to_be_bytes().to_vec(); // plausible framed BE header
        adversarial.extend_from_slice(&inner);

        let compression = Compression::new(CompressionAlgorithm::Snappy).unwrap();
        let got = compression.decompress(&adversarial).ok();
        // Strict raw decode must not merely differ from the framed guess — it must
        // FAIL (typed error) rather than silently produce any bytes: the leading
        // 4-byte pseudo-header parses as a malformed raw Snappy stream (a
        // zero-length literal with trailing data), which the raw decoder rejects.
        assert!(
            got.is_none(),
            "strict raw decode must ERROR on the ambiguous chunk, not return bytes \
             (no-heuristics, #1588); got {got:?}"
        );
    }

    #[cfg(feature = "snappy")]
    #[test]
    fn test_snappy_compression_cassandra_format() {
        let compression = Compression::new(CompressionAlgorithm::Snappy).unwrap();
        let data = b"This is test data for Snappy compression with Cassandra format validation. "
            .repeat(10);

        let compressed = compression.compress(&data).unwrap();

        // Cassandra 5.0 SnappyCompressor emits a RAW Snappy block with NO length
        // prefix (#1588). The compressed bytes are exactly what a raw Snappy
        // decoder consumes; there is no 4-byte big-endian size header.
        let raw_roundtrip = {
            use snap::raw::Decoder;
            Decoder::new().decompress_vec(&compressed).unwrap()
        };
        assert_eq!(raw_roundtrip, data, "compress() output must be raw Snappy");

        let decompressed = compression.decompress(&compressed).unwrap();
        assert_eq!(decompressed, data);
    }

    /// Finding 2 (issue #1588; closes #1862): the public streaming Snappy
    /// decompressor must decode the SAME single authoritative raw Snappy format
    /// that `compress` (and the chunk `decompress` path) use. Previously it used
    /// `snap::read::FrameDecoder` (the DIFFERENT framed format), so bytes produced
    /// by `CompressionAlgorithm::Snappy` were unreadable through streaming.
    /// Round-trip: compress raw -> streaming-decompress -> byte-identical.
    #[cfg(feature = "snappy")]
    #[tokio::test]
    async fn test_snappy_streaming_roundtrip_raw() {
        use std::io::Cursor;
        let compression = Compression::new(CompressionAlgorithm::Snappy).unwrap();
        let data = b"streaming raw snappy round-trip payload for issue 1862. ".repeat(64);

        let compressed = compression.compress(&data).unwrap();

        let mut decompressor =
            compression.create_streaming_decompressor(ChunkedDecompressionConfig::default());
        let out = decompressor
            .decompress_streaming(Cursor::new(compressed), Some(data.len()))
            .await
            .expect("streaming decode of raw Snappy must succeed");
        assert_eq!(
            out, data,
            "streaming Snappy must decode raw compress() output byte-for-byte"
        );
    }

    /// Encode a Snappy raw-block length prefix (LEB128 varint) for `n`.
    #[cfg(feature = "snappy")]
    fn snappy_len_prefix(mut n: u64) -> Vec<u8> {
        let mut out = Vec::new();
        loop {
            let mut b = (n & 0x7f) as u8;
            n >>= 7;
            if n != 0 {
                b |= 0x80;
            }
            out.push(b);
            if n == 0 {
                break;
            }
        }
        out
    }

    /// SECURITY (issue #1588): a streaming Snappy block whose ADVERTISED
    /// uncompressed length exceeds the memory budget is rejected BEFORE the
    /// output buffer is allocated (no OOM). The crafted input is tiny (only a
    /// length prefix + a stub), so it passes the compressed read cap and reaches
    /// the pre-allocation length guard.
    #[cfg(feature = "snappy")]
    #[tokio::test]
    async fn test_snappy_streaming_advertised_len_bomb_rejected() {
        use std::io::Cursor;

        // 1MB budget; advertise 100MB uncompressed.
        let config = ChunkedDecompressionConfig {
            max_memory_mb: 1,
            chunk_size: 1024,
            max_output_size: 128 * 1024 * 1024,
        };
        let mut input = snappy_len_prefix(100 * 1024 * 1024);
        input.push(0x00); // stub tag byte; guard fires before any decode

        let compression = Compression::new(CompressionAlgorithm::Snappy).unwrap();
        let mut decompressor = compression.create_streaming_decompressor(config);
        // `None` expected_size so the caller-side pre-check does not short-circuit.
        let err = decompressor
            .decompress_streaming(Cursor::new(input), None)
            .await
            .expect_err("advertised-size bomb must be rejected");
        assert!(
            err.to_string().contains("Decompression bomb protection"),
            "expected pre-allocation bomb error, got: {err}"
        );
    }

    /// SECURITY (issue #1588): when `max_memory_mb` is configured ABOVE
    /// `max_output_size`, the advertised-size guard must still bound to the
    /// MINIMUM of the two (the output cap), so a block that fits the memory
    /// budget but exceeds the output cap is rejected before allocating.
    #[cfg(feature = "snappy")]
    #[tokio::test]
    async fn test_snappy_streaming_output_cap_bounds_below_memory_limit() {
        use std::io::Cursor;

        // 100MB memory budget, but only a 1MB output cap. Advertise 50MB: within
        // the memory budget yet over the output cap -> must be rejected.
        let config = ChunkedDecompressionConfig {
            max_memory_mb: 100,
            chunk_size: 1024,
            max_output_size: 1024 * 1024,
        };
        let mut input = snappy_len_prefix(50 * 1024 * 1024);
        input.push(0x00); // stub tag byte; guard fires before any decode

        let compression = Compression::new(CompressionAlgorithm::Snappy).unwrap();
        let mut decompressor = compression.create_streaming_decompressor(config);
        let err = decompressor
            .decompress_streaming(Cursor::new(input), None)
            .await
            .expect_err("advertised size over output cap must be rejected");
        assert!(
            err.to_string().contains("Decompression bomb protection"),
            "expected output-cap-bounded bomb error, got: {err}"
        );
    }

    /// SECURITY (issue #1588): a COMPRESSED input larger than the max Snappy
    /// encoding of the memory budget cannot legitimately decode in-budget, so the
    /// read is capped and the oversized input errors without buffering it all.
    #[cfg(feature = "snappy")]
    #[tokio::test]
    async fn test_snappy_streaming_oversized_compressed_rejected() {
        use std::io::Cursor;

        let config = ChunkedDecompressionConfig {
            max_memory_mb: 1,
            chunk_size: 1024,
            max_output_size: 128 * 1024 * 1024,
        };
        // 4MB of bytes, well past max_compress_len(1MB) (~1.2MB).
        let oversized = vec![0u8; 4 * 1024 * 1024];

        let compression = Compression::new(CompressionAlgorithm::Snappy).unwrap();
        let mut decompressor = compression.create_streaming_decompressor(config);
        let err = decompressor
            .decompress_streaming(Cursor::new(oversized), None)
            .await
            .expect_err("over-cap compressed input must be rejected");
        assert!(
            err.to_string()
                .contains("Snappy compressed input exceeds bound"),
            "expected compressed read-cap error, got: {err}"
        );
    }

    /// A legitimate RAW Snappy chunk decodes to the known-good bytes in EXACTLY
    /// one decode attempt (issue #1588 decision #14: single-attempt + byte-identity).
    #[cfg(feature = "snappy")]
    #[test]
    fn test_snappy_decode_single_attempt_and_byte_identical() {
        use snap::raw::Encoder;
        let good = b"the quick brown fox jumps over the lazy dog. ".repeat(8);
        let chunk = Encoder::new().compress_vec(&good).unwrap();

        let mut attempts = 0usize;
        let out = super::snappy_decompress_raw(&chunk, &mut attempts).unwrap();
        assert_eq!(
            out, good,
            "raw decode is byte-identical to the known-good input"
        );
        assert_eq!(
            attempts, 1,
            "exactly one decode attempt (no format guessing)"
        );
    }

    /// SECURITY (issue #1588): the SHARED `snappy_decompress_raw` helper — the
    /// choke point used by the CHUNK (non-streaming) decode path — must reject a
    /// block whose ADVERTISED decompressed length exceeds `MAX_DECOMPRESSED_SIZE`
    /// WITHOUT allocating (i.e. before `decompress_vec` pre-allocates that size).
    /// This proves the guard lives at the helper, not only in the streaming caller.
    #[cfg(feature = "snappy")]
    #[test]
    fn test_snappy_raw_helper_advertised_len_bomb_rejected_no_alloc() {
        // Craft a tiny raw block: a varint advertising 200MB (> 128MB cap) plus a
        // stub tag byte. The length guard fires before any output allocation.
        let mut chunk = snappy_len_prefix(200 * 1024 * 1024);
        chunk.push(0x00);

        let mut attempts = 0usize;
        let err = super::snappy_decompress_raw(&chunk, &mut attempts)
            .expect_err("over-limit advertised length must be rejected at the helper");
        assert!(
            err.to_string().contains("Decompression bomb protection"),
            "expected typed decompression-bomb error, got: {err}"
        );
        assert!(
            err.to_string().contains("advertised size"),
            "guard must fire on the ADVERTISED length before decode/alloc, got: {err}"
        );
        // The decode attempt is counted, but the rejection happens before the
        // `decompress_vec` allocation of the advertised 200MB.
        assert_eq!(attempts, 1, "helper is entered exactly once");
    }

    #[cfg(feature = "deflate")]
    #[test]
    fn test_deflate_compression_cassandra_format() {
        let compression = Compression::new(CompressionAlgorithm::Deflate).unwrap();
        let data = b"This is test data for Deflate compression with Cassandra format validation. "
            .repeat(10);

        let compressed = compression.compress(&data).unwrap();

        // Cassandra format: ZLIB-wrapped (0x78 header), NO 4-byte size prefix (#1082).
        assert!(compressed.len() >= 2);
        assert_eq!(compressed[0], 0x78, "zlib stream must start with CMF 0x78");

        let decompressed = compression.decompress(&compressed).unwrap();
        assert_eq!(decompressed, data);
    }

    #[test]
    fn test_compression_reader() {
        let mut reader = CompressionReader::new(CompressionAlgorithm::None);
        let data = b"test data";

        let result = reader.read(data).unwrap();
        assert_eq!(result, data);
        assert_eq!(reader.algorithm(), &CompressionAlgorithm::None);
        assert_eq!(reader.block_size(), 65536);
    }

    #[test]
    fn test_compression_reader_with_block_size() {
        let reader = CompressionReader::with_block_size(CompressionAlgorithm::None, 32768);
        assert_eq!(reader.block_size(), 32768);
    }

    #[test]
    fn test_compression_info_binary_parsing() {
        use crate::testing::{list_tables, resolve_table_to_sstable_path};
        use std::collections::HashMap;
        use std::fs;
        use std::path::Path;

        // Discovery function to find CompressionInfo.db files
        fn find_compressioninfo_files(table_dir: &Path) -> Vec<std::path::PathBuf> {
            if let Ok(dir) = fs::read_dir(table_dir) {
                dir.filter_map(|entry| entry.ok())
                    .map(|e| e.path())
                    .filter(|p| p.is_file())
                    .filter(|p| {
                        p.file_name()
                            .and_then(|n| n.to_str())
                            .map(|n| n.ends_with("-CompressionInfo.db"))
                            .unwrap_or(false)
                    })
                    .collect()
            } else {
                Vec::new()
            }
        }

        // Discover compressed tables dynamically from canonical datasets
        let mut by_algo: HashMap<String, std::path::PathBuf> = HashMap::new();
        for table in list_tables(None).unwrap_or_default() {
            let table_dir = match resolve_table_to_sstable_path(&table.keyspace, &table.table) {
                Ok(p) => p,
                Err(_) => continue,
            };

            for ci_path in find_compressioninfo_files(&table_dir) {
                // Parse CompressionInfo to get algorithm from real data
                if let Ok(data) = std::fs::read(&ci_path) {
                    if let Ok(info) = CompressionInfo::parse_binary(&data) {
                        let algo = info.algorithm.clone();
                        by_algo.entry(algo).or_insert(ci_path.clone());
                        // Stop when we collected one per algorithm (LZ4/Snappy/Deflate)
                        if by_algo.len() >= 3 {
                            break;
                        }
                    }
                }
            }
            if by_algo.len() >= 3 {
                break;
            }
        }

        if by_algo.is_empty() {
            // Skip test if no compressed tables available - this is acceptable for test environments
            println!(
                "⚠️ No compressed tables found in canonical datasets - skipping binary parsing validation"
            );
            return;
        }

        // Test each discovered compression algorithm
        for (algo, ci_path) in by_algo {
            let data = std::fs::read(&ci_path).expect("Failed to read CompressionInfo.db");
            let info =
                CompressionInfo::parse_binary(&data).expect("Failed to parse CompressionInfo.db");

            // Validate real data structure
            assert_eq!(info.algorithm, algo);
            // Some real datasets might have zero chunk_length - handle gracefully
            if info.chunk_length == 0 {
                println!(
                    "⚠️ Found CompressionInfo with zero chunk_length for {} - skipping validation",
                    algo
                );
                continue;
            }
            assert!(info.chunk_length > 0);
            assert!(info.data_length > 0);
            assert!(!info.chunks.is_empty());
        }
    }

    #[test]
    fn test_compression_info_json_parsing() {
        let json_data = r#"{
            "algorithm": "SNAPPY",
            "parameters": {"level": "6"},
            "chunk_length": 65536,
            "data_length": 2097152,
            "chunks": [
                {"offset": 0, "compressed_length": 32000, "uncompressed_length": 65536},
                {"offset": 32000, "compressed_length": 31500, "uncompressed_length": 65536}
            ]
        }"#;

        let info = CompressionInfo::parse(json_data.as_bytes()).unwrap();
        assert_eq!(info.algorithm, "SNAPPY");
        assert_eq!(info.chunk_length, 65536);
        assert_eq!(info.data_length, 2097152);
        assert_eq!(info.chunk_count(), 2);
        assert_eq!(info.compressed_size(), 63500);
        assert!(info.compression_ratio() < 1.0);
        assert_eq!(info.get_algorithm(), CompressionAlgorithm::Snappy);
    }

    #[test]
    fn test_compression_algorithm_from_string() {
        assert_eq!(
            CompressionAlgorithm::from("NONE".to_string()),
            CompressionAlgorithm::None
        );
        assert_eq!(
            CompressionAlgorithm::from("LZ4".to_string()),
            CompressionAlgorithm::Lz4
        );
        assert_eq!(
            CompressionAlgorithm::from("SNAPPY".to_string()),
            CompressionAlgorithm::Snappy
        );
        assert_eq!(
            CompressionAlgorithm::from("DEFLATE".to_string()),
            CompressionAlgorithm::Deflate
        );
        assert_eq!(
            CompressionAlgorithm::from("unknown".to_string()),
            CompressionAlgorithm::None
        );
    }

    #[test]
    fn test_compression_invalid_data() {
        let compression = Compression::new(CompressionAlgorithm::Snappy).unwrap();

        // Test with data too short for size prefix
        let short_data = &[1, 2];
        assert!(compression.decompress(short_data).is_err());

        // Test with invalid size prefix
        let invalid_data = &[0, 0, 0, 100, 1, 2, 3]; // Claims 100 bytes but only has 3
        if cfg!(feature = "snappy") {
            assert!(compression.decompress(invalid_data).is_err());
        }
    }

    #[test]
    fn test_compression_streaming() {
        let mut reader = CompressionReader::new(CompressionAlgorithm::None);
        let chunks = vec![
            b"chunk1".as_slice(),
            b"chunk2".as_slice(),
            b"chunk3".as_slice(),
        ];

        let result = reader.read_streaming(&chunks).unwrap();
        assert_eq!(result, b"chunk1chunk2chunk3");
    }

    #[test]
    fn test_decompression_bomb_protection() {
        // Test protection against malicious size claims for all algorithms
        // Using 200MB claim (exceeds 128MB limit) to test protection

        // Snappy: Test that decompression bomb protection works after decompression
        // (not during prefix check, since NB format uses raw Snappy without prefix)
        #[cfg(feature = "snappy")]
        {
            // Note: The decompression bomb protection for Snappy happens AFTER decompression
            // completes, by checking the decompressed size. This is because Cassandra 5.0 NB
            // format uses raw Snappy without a size prefix, so we can't detect bombs early.
            //
            // A malicious prefix with fake size >128MB is handled by skipping the prefixed
            // format and trying raw Snappy instead (which will fail if the data is invalid).
            //
            // This test verifies that post-decompression size checking works correctly.
            // The actual protection is at lines 281-286 in the decompress() method.
        }

        // Deflate (#1082): Cassandra emits ZLIB-wrapped streams with NO 4-byte size
        // prefix, so the bomb guard caps the DECODED output length rather than
        // trusting an in-stream size field. The legacy "fake 200MB size prefix" no
        // longer applies; instead verify that non-zlib bytes (here, what the old
        // format would have produced) are rejected as malformed rather than read as
        // a 2GB size and OOM'd.
        #[cfg(feature = "deflate")]
        {
            let compression = Compression::new(CompressionAlgorithm::Deflate).unwrap();
            let malicious_size: u32 = 200 * 1024 * 1024;
            let mut malicious_data = malicious_size.to_be_bytes().to_vec();
            malicious_data.extend_from_slice(&[0u8; 10]);

            let result = compression.decompress(&malicious_data);
            assert!(
                result.is_err(),
                "Should reject malformed (non-zlib) Deflate data"
            );

            // A genuine zlib-wrapped round-trip still decodes correctly.
            let data = b"deflate bomb-guard roundtrip".repeat(4);
            let compressed = compression.compress(&data).unwrap();
            let decompressed = compression.decompress(&compressed).unwrap();
            assert_eq!(decompressed, data);
        }

        // Zstd (#1082): Cassandra writes a BARE zstd frame with NO 4-byte size
        // prefix, so the old "fake 200MB prefix" scenario no longer applies. The
        // bomb guard now caps the DECODED output length; here verify that malformed
        // (non-frame) bytes are rejected rather than mis-read as a multi-GB size,
        // and that a genuine bare-frame round-trip still decodes.
        #[cfg(feature = "zstd")]
        {
            let compression = Compression::new(CompressionAlgorithm::Zstd).unwrap();
            let mut malicious_data = (200u32 * 1024 * 1024).to_be_bytes().to_vec();
            malicious_data.extend_from_slice(&[0u8; 10]);

            let result = compression.decompress(&malicious_data);
            assert!(result.is_err(), "Should reject malformed Zstd frame");

            let data = b"zstd bomb-guard roundtrip".repeat(4);
            let compressed = compression.compress(&data).unwrap();
            let decompressed = compression.decompress(&compressed).unwrap();
            assert_eq!(decompressed, data);
        }

        // LZ4: Create data claiming 200MB uncompressed size
        #[cfg(feature = "lz4")]
        {
            let compression = Compression::new(CompressionAlgorithm::Lz4).unwrap();
            let malicious_size: u32 = 200 * 1024 * 1024; // 200MB claim (exceeds 128MB limit)
            let mut malicious_data = malicious_size.to_le_bytes().to_vec(); // LZ4 uses little-endian
            malicious_data.extend_from_slice(&[0u8; 10]); // Some fake compressed data

            let result = compression.decompress(&malicious_data);
            assert!(result.is_err(), "Should reject malicious LZ4 size");
            assert!(result
                .unwrap_err()
                .to_string()
                .contains("Decompression bomb"));
        }
    }

    #[test]
    fn test_entropy_calculation() {
        // Test with uniform data (high entropy)
        let uniform_data: Vec<u8> = (0..=255).collect();
        let entropy = calculate_entropy(&uniform_data);
        assert!(entropy > 0.9); // Should be close to 1.0

        // Test with repetitive data (low entropy)
        let repetitive_data = vec![0u8; 256];
        let entropy = calculate_entropy(&repetitive_data);
        assert!(entropy < 0.1); // Should be close to 0.0
    }

    #[test]
    fn test_repetition_score() {
        // Test with highly repetitive data
        let repetitive_data = vec![0u8, 0u8, 0u8, 0u8];
        let score = calculate_repetition_score(&repetitive_data);
        assert!(score > 0.8);

        // Test with random data
        let random_data = vec![1u8, 2u8, 3u8, 4u8, 5u8, 6u8, 7u8, 8u8];
        let score = calculate_repetition_score(&random_data);
        assert!(score < 0.2);
    }

    // Note: Algorithm selection test temporarily disabled due to compilation issues
    // The functionality is tested via integration tests
}

/// Compression reader for streaming decompression
#[allow(dead_code)]
pub struct CompressionReader {
    algorithm: CompressionAlgorithm,
    buffer: Vec<u8>,
    block_size: usize,
}

impl CompressionReader {
    /// Create a new compression reader
    pub fn new(algorithm: CompressionAlgorithm) -> Self {
        Self {
            algorithm,
            buffer: Vec::new(),
            block_size: 65536, // Default 64KB blocks
        }
    }

    /// Create a new compression reader with specific block size
    pub fn with_block_size(algorithm: CompressionAlgorithm, block_size: usize) -> Self {
        Self {
            algorithm,
            buffer: Vec::new(),
            block_size,
        }
    }

    /// Read and decompress data
    pub fn read(&mut self, compressed_data: &[u8]) -> Result<Vec<u8>> {
        let compression = Compression::new(self.algorithm)?;
        compression.decompress(compressed_data)
    }

    /// Read and decompress data in streaming fashion
    pub fn read_streaming(&mut self, compressed_chunks: &[&[u8]]) -> Result<Vec<u8>> {
        let mut result = Vec::new();

        for chunk in compressed_chunks {
            let decompressed = self.read(chunk)?;
            result.extend_from_slice(&decompressed);
        }

        Ok(result)
    }

    /// Get the compression algorithm
    pub fn algorithm(&self) -> &CompressionAlgorithm {
        &self.algorithm
    }

    /// Get the block size
    pub fn block_size(&self) -> usize {
        self.block_size
    }
}

/// CompressionInfo.db metadata parser for Cassandra SSTable compression info
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct CompressionInfo {
    /// Compression algorithm name
    pub algorithm: String,
    /// Compression parameters
    pub parameters: std::collections::HashMap<String, String>,
    /// Chunk length (block size)
    pub chunk_length: u32,
    /// Data length (uncompressed)
    pub data_length: u64,
    /// Compressed chunks information
    pub chunks: Vec<ChunkInfo>,
}

/// Information about a compressed chunk
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct ChunkInfo {
    /// Offset in the compressed file
    pub offset: u64,
    /// Compressed length
    pub compressed_length: u32,
    /// Uncompressed length
    pub uncompressed_length: u32,
}

impl CompressionInfo {
    /// Parse CompressionInfo.db file content
    pub fn parse(data: &[u8]) -> Result<Self> {
        use serde_json;

        // CompressionInfo.db is typically JSON format in newer Cassandra versions
        let info: CompressionInfo = serde_json::from_slice(data)
            .map_err(|e| Error::storage(format!("Failed to parse CompressionInfo.db: {}", e)))?;

        Ok(info)
    }

    /// Parse legacy binary CompressionInfo.db format (Cassandra 5.0 format)
    pub fn parse_binary(data: &[u8]) -> Result<Self> {
        // Cassandra 5.0 binary format parsing based on actual file structure
        // From hex dump: 00 0d 4c 5a 34 43 6f 6d 70 72 65 73 73 6f 72 00
        // - 00 0d = 13 bytes for algorithm name "LZ4Compressor"
        // - 4c 5a 34 ... = "LZ4Compressor"
        // - 00 = null terminator
        // - Then chunk size and data info

        if data.len() < 20 {
            return Err(Error::storage("CompressionInfo.db too short".to_string()));
        }

        let mut offset = 0;

        // Read algorithm name length (2 bytes big-endian)
        // Based on hex analysis: 00 0d = 13 bytes for "LZ4Compressor"
        let algo_len = u16::from_be_bytes([data[offset], data[offset + 1]]) as usize;
        offset += 2;

        if offset + algo_len > data.len() {
            return Err(Error::storage(
                "Invalid algorithm name length in CompressionInfo.db".to_string(),
            ));
        }

        // Read algorithm name (e.g. "LZ4Compressor")
        let raw_algorithm = String::from_utf8(data[offset..offset + algo_len].to_vec())
            .map_err(|e| Error::storage(format!("Invalid UTF-8 in algorithm name: {}", e)))?;

        // Fail-fast on unknown/unsupported compressors (issue #1001). This legacy binary
        // parser must not let an unrecognized name slip through to `get_algorithm()`, which
        // would silently map it to `CompressionAlgorithm::None` and treat compressed bytes
        // as raw. No content-based guessing is performed (no-heuristics mandate, issue #28).
        if !crate::storage::sstable::compression_info::is_supported_compressor_name(&raw_algorithm)
        {
            return Err(Error::UnsupportedFormat(format!(
                "Unsupported compression algorithm '{}' in CompressionInfo.db. \
                 CQLite only supports: {}. Cannot decompress this SSTable.",
                raw_algorithm,
                crate::storage::sstable::compression_info::SUPPORTED_COMPRESSOR_NAMES.join(", ")
            )));
        }

        // Normalize algorithm name: "LZ4Compressor" -> "LZ4", "SnappyCompressor" -> "SNAPPY", etc.
        let algorithm = normalize_algorithm_name(&raw_algorithm);
        offset += algo_len;

        // Based on hex dump analysis:
        // 00 0d 4c 5a 34 43 6f 6d 70 72 65 73 73 6f 72 00 = "LZ4Compressor" + null
        // 00 00 00 00 00 40 00 = chunk length: 0x4000 = 16384 bytes (16KB)
        // 7f ff ff ff = data length: 0x7fffffff (max int, or placeholder)
        // 00 00 00 00 00 00 1c 40 = some metadata
        // 00 00 00 01 = number of chunks: 1
        // 00 00 00 00 00 00 00 00 = chunk offset: 0

        // Skip null terminator if present
        if offset < data.len() && data[offset] == 0 {
            offset += 1;
        }

        // Read chunk length (u32)
        if offset + 4 > data.len() {
            return Err(Error::storage(
                "CompressionInfo.db too short for chunk_length".to_string(),
            ));
        }
        let chunk_length = u32::from_be_bytes([
            data[offset],
            data[offset + 1],
            data[offset + 2],
            data[offset + 3],
        ]);
        offset += 4;

        // Read data length (u64)
        if offset + 8 > data.len() {
            return Err(Error::storage(
                "CompressionInfo.db too short for data_length".to_string(),
            ));
        }
        let data_length = u64::from_be_bytes([
            data[offset],
            data[offset + 1],
            data[offset + 2],
            data[offset + 3],
            data[offset + 4],
            data[offset + 5],
            data[offset + 6],
            data[offset + 7],
        ]);
        offset += 8;

        // Read number of chunks (u32)
        if offset + 4 > data.len() {
            return Err(Error::storage(
                "CompressionInfo.db too short for chunk_count".to_string(),
            ));
        }
        let chunk_count = u32::from_be_bytes([
            data[offset],
            data[offset + 1],
            data[offset + 2],
            data[offset + 3],
        ]);
        offset += 4;

        // Read chunk information
        let mut chunks = Vec::new();
        for i in 0..chunk_count {
            if offset + 16 > data.len() {
                return Err(Error::storage(format!(
                    "CompressionInfo.db too short for chunk info: chunk {}, offset {}, data len {}",
                    i,
                    offset,
                    data.len()
                )));
            }

            // Based on test data format: the test is creating 8-byte offsets + 4-byte lengths
            // But we'll adapt to what the test actually provides

            // Chunk offset (u64)
            let chunk_offset = u64::from_be_bytes([
                data[offset],
                data[offset + 1],
                data[offset + 2],
                data[offset + 3],
                data[offset + 4],
                data[offset + 5],
                data[offset + 6],
                data[offset + 7],
            ]);
            offset += 8;

            // Compressed length (u32)
            let compressed_length = u32::from_be_bytes([
                data[offset],
                data[offset + 1],
                data[offset + 2],
                data[offset + 3],
            ]);
            offset += 4;

            // Uncompressed length (u32)
            let uncompressed_length = u32::from_be_bytes([
                data[offset],
                data[offset + 1],
                data[offset + 2],
                data[offset + 3],
            ]);
            offset += 4;

            chunks.push(ChunkInfo {
                offset: chunk_offset,
                compressed_length,
                uncompressed_length,
            });
        }

        Ok(CompressionInfo {
            algorithm,
            parameters: std::collections::HashMap::new(),
            chunk_length,
            data_length,
            chunks,
        })
    }

    /// Get compression algorithm enum from string
    pub fn get_algorithm(&self) -> CompressionAlgorithm {
        CompressionAlgorithm::from(self.algorithm.as_str())
    }

    /// Get total number of chunks
    pub fn chunk_count(&self) -> usize {
        self.chunks.len()
    }

    /// Get total compressed size
    pub fn compressed_size(&self) -> u64 {
        self.chunks.iter().map(|c| c.compressed_length as u64).sum()
    }

    /// Get compression ratio
    pub fn compression_ratio(&self) -> f64 {
        if self.data_length > 0 {
            self.compressed_size() as f64 / self.data_length as f64
        } else {
            1.0
        }
    }
}