sql_cli/data/

stream_loader.rs

// Stream-based data loader that works with any Read source
// This allows the same code to handle files, HTTP responses, or any other data stream

use anyhow::{Context, Result};
use csv::ReaderBuilder;
use serde_json::Value as JsonValue;
use std::borrow::Cow;
use std::collections::{HashMap, HashSet};
use std::io::{BufRead, BufReader, Read};
use tracing::{debug, info};

use crate::data::advanced_csv_loader::StringInterner;
use crate::data::datatable::{DataColumn, DataRow, DataTable, DataType, DataValue};

/// Options controlling how a CSV stream is parsed.
///
/// Default is RFC-4180 style: comma delimiter, header row required. Surfaces
/// (CLI flag, `READ_CSV(path, '|')`, WEB CTE `DELIMITER`) populate this struct
/// at the edge; internal layers just pass it through.
#[derive(Debug, Clone)]
pub struct CsvReadOptions {
    pub delimiter: u8,
    pub has_headers: bool,
}

impl Default for CsvReadOptions {
    fn default() -> Self {
        Self {
            delimiter: b',',
            has_headers: true,
        }
    }
}

/// Pick a default delimiter from a path's extension.
///
/// `.tsv` → tab, `.psv` → pipe; everything else (including stdin `-`) → comma.
/// Case-insensitive. This is only the *auto-detect* layer — explicit overrides
/// from the CLI flag, `READ_CSV` 2nd arg, or WEB CTE `DELIMITER` win over this.
pub fn detect_delimiter_from_path(path: &str) -> u8 {
    let lower = path.to_ascii_lowercase();
    if lower.ends_with(".tsv") {
        b'\t'
    } else if lower.ends_with(".psv") {
        b'|'
    } else {
        b','
    }
}

/// Parse a user-supplied delimiter string into a single byte.
///
/// Accepts:
///   - a single ASCII character (e.g. `","`, `"|"`, `";"`)
///   - the two-character escapes `"\t"`, `"\n"`, `"\r"` (typing literal tabs
///     in SQL strings or shell args is awkward, so this is the canonical form)
///
/// Rejects multi-character strings, non-ASCII, and empty strings with a clear
/// error. Caller is expected to wrap the error with context if needed.
pub fn parse_delimiter_arg(s: &str) -> anyhow::Result<u8> {
    match s {
        "\\t" | "\t" => return Ok(b'\t'),
        "\\n" => return Ok(b'\n'),
        "\\r" => return Ok(b'\r'),
        _ => {}
    }
    let bytes = s.as_bytes();
    if bytes.len() == 1 && bytes[0].is_ascii() {
        return Ok(bytes[0]);
    }
    Err(anyhow::anyhow!(
        "delimiter must be a single ASCII character (or '\\t', '\\n', '\\r'); got {:?}",
        s
    ))
}

/// Resolve which delimiter to use for a given path.
///
/// Precedence (highest first):
///   1. `explicit` override (typically from a CLI flag or 2nd arg)
///   2. extension auto-detect (`.tsv` → tab, `.psv` → pipe)
///   3. comma
pub fn resolve_delimiter(path: &str, explicit: Option<u8>) -> u8 {
    explicit.unwrap_or_else(|| detect_delimiter_from_path(path))
}

/// Delete the whitespace padding that "pretty printed" CSVs put around *quoted*
/// fields, so the parser can see they were quoted at all.
///
/// RFC 4180 only recognises a quote that is the *first* byte of a field, so a
/// human-aligned file like
///
/// ```text
/// "Index", "Name", "Day"
///  1, "George Washington",  22
/// ```
///
/// parses the second column's header as the literal text ` "Name"` — padding,
/// quotes and all — which is why `SELECT Name` then fails with "column not
/// found". Worse, a comma *inside* a padded field splits it in two, because the
/// parser never saw the field as quoted.
///
/// Nobody writing a file like that intends the quotes to end up in their column
/// names, so we delete the run of spaces/tabs in front of an opening quote (and
/// the mirror run after the closing quote). That turns the field back into a
/// properly quoted one and the parser handles it from there — embedded
/// delimiters included.
///
/// Deliberately narrow: **unquoted fields are left exactly as they are.** In a
/// file written by a machine, ` 22` and `22` mean the same thing, but there is
/// no way to tell that apart from a value whose spaces are real — an unquoted
/// `  David  ` is the only way some producers can express a padded string, and
/// `data/test_simple_strings.csv` relies on it surviving the load. Whitespace
/// next to a quote carries no such ambiguity: the quotes already delimit the
/// value, so anything outside them is alignment. Numeric-looking unquoted fields
/// get their padding handled at type-inference time instead, where trimming
/// can't destroy a string.
///
/// Bytes inside a quoted field are copied verbatim, so `"  padded  "` keeps its
/// spaces — if you quoted the whitespace, you meant it. The delimiter itself is
/// never treated as padding, which keeps `.tsv` files intact.
///
/// Returns [`Cow::Borrowed`] when there was nothing to strip, so the common
/// RFC-4180 case costs one scan and no allocation.
pub fn strip_field_padding(input: &[u8], delimiter: u8) -> Cow<'_, [u8]> {
    let is_pad = |b: u8| (b == b' ' || b == b'\t') && b != delimiter;
    let ends_field =
        |b: Option<u8>| matches!(b, None | Some(b'\n') | Some(b'\r')) || b == Some(delimiter);

    let mut out: Option<Vec<u8>> = None;
    let mut i = 0;
    let mut at_field_start = true;
    let mut in_quotes = false;
    // Set when a quoted field has just closed, so the padding that follows it is
    // known to be alignment rather than part of an unquoted value.
    let mut after_quoted_field = false;

    while i < input.len() {
        let b = input[i];

        if in_quotes {
            // `""` is an escaped quote and stays inside the field.
            if b == b'"' {
                if input.get(i + 1) == Some(&b'"') {
                    if let Some(o) = out.as_mut() {
                        o.extend_from_slice(&input[i..i + 2]);
                    }
                    i += 2;
                    continue;
                }
                in_quotes = false;
                after_quoted_field = true;
            }
            if let Some(o) = out.as_mut() {
                o.push(b);
            }
            i += 1;
            continue;
        }

        if is_pad(b) {
            let mut j = i;
            while j < input.len() && is_pad(input[j]) {
                j += 1;
            }
            let next = input.get(j).copied();
            // Padding is only ours to remove when a quote sits on one side of it:
            // before an opening quote, or after a closing one.
            let before_open_quote = at_field_start && next == Some(b'"');
            let after_close_quote = after_quoted_field && ends_field(next);
            if before_open_quote || after_close_quote {
                out.get_or_insert_with(|| input[..i].to_vec());
                i = j;
                continue;
            }
            // Anything else belongs to an unquoted value — leave it alone.
            if let Some(o) = out.as_mut() {
                o.extend_from_slice(&input[i..j]);
            }
            i = j;
            at_field_start = false;
            continue;
        }

        if b == b'"' && at_field_start {
            in_quotes = true;
        }
        at_field_start = b == delimiter || b == b'\n' || b == b'\r';
        if at_field_start {
            after_quoted_field = false;
        }
        if let Some(o) = out.as_mut() {
            o.push(b);
        }
        i += 1;
    }

    out.map_or(Cow::Borrowed(input), Cow::Owned)
}

/// Human-readable form of a delimiter byte for diagnostic metadata.
fn delimiter_label(d: u8) -> String {
    match d {
        b'\t' => "\\t".to_string(),
        b'\n' => "\\n".to_string(),
        b'\r' => "\\r".to_string(),
        b => (b as char).to_string(),
    }
}

/// Column analysis results for determining interning strategy
#[derive(Debug)]
struct ColumnAnalysis {
    index: usize,
    _name: String,
    _cardinality: usize,
    _sample_size: usize,
    _unique_ratio: f64,
    is_categorical: bool,
    _avg_string_length: usize,
}

/// Advanced stream-based CSV loader with string interning
pub struct StreamCsvLoader {
    sample_size: usize,
    cardinality_threshold: f64,
    interners: HashMap<usize, StringInterner>,
}

impl StreamCsvLoader {
    pub fn new() -> Self {
        Self {
            sample_size: 1000,
            cardinality_threshold: 0.3,
            interners: HashMap::new(),
        }
    }

    /// Analyze columns to determine which should use string interning
    fn analyze_columns(
        &self,
        rows: &[Vec<String>],
        headers: &csv::StringRecord,
    ) -> Vec<ColumnAnalysis> {
        let mut analyses = Vec::new();

        for (col_idx, header) in headers.iter().enumerate() {
            let mut unique_values = HashSet::new();
            let mut total_length = 0;
            let mut non_empty_count = 0;

            // Sample rows to analyze cardinality
            for row in rows.iter().take(self.sample_size) {
                if let Some(value) = row.get(col_idx) {
                    if !value.is_empty() {
                        unique_values.insert(value.clone());
                        total_length += value.len();
                        non_empty_count += 1;
                    }
                }
            }

            let cardinality = unique_values.len();
            let sample_size = rows.len().min(self.sample_size);
            let unique_ratio = if sample_size > 0 {
                cardinality as f64 / sample_size as f64
            } else {
                1.0
            };

            let avg_string_length = if non_empty_count > 0 {
                total_length / non_empty_count
            } else {
                0
            };

            // Consider categorical if low cardinality ratio or short strings with repetition
            let is_categorical = unique_ratio < self.cardinality_threshold
                || (avg_string_length < 20 && cardinality < sample_size / 2);

            analyses.push(ColumnAnalysis {
                index: col_idx,
                _name: header.to_string(),
                _cardinality: cardinality,
                _sample_size: sample_size,
                _unique_ratio: unique_ratio,
                is_categorical,
                _avg_string_length: avg_string_length,
            });
        }

        analyses
    }

    /// Load CSV data with string interning from any Read source, using default
    /// comma-delimited options. Thin wrapper over [`load_csv_from_reader_with_opts`]
    /// kept so existing callers don't need to touch options.
    pub fn load_csv_from_reader<R: Read>(
        &mut self,
        reader: R,
        table_name: &str,
        source_type: &str,
        source_path: &str,
    ) -> Result<DataTable> {
        self.load_csv_from_reader_with_opts(
            reader,
            table_name,
            source_type,
            source_path,
            &CsvReadOptions::default(),
        )
    }

    /// Load CSV data with string interning, honouring caller-supplied options
    /// (delimiter, headers).
    pub fn load_csv_from_reader_with_opts<R: Read>(
        &mut self,
        mut reader: R,
        table_name: &str,
        source_type: &str,
        source_path: &str,
        opts: &CsvReadOptions,
    ) -> Result<DataTable> {
        info!(
            "Stream CSV load: Loading {} with optimizations (delimiter={})",
            source_path,
            delimiter_label(opts.delimiter)
        );

        // Read all data into memory
        let mut raw_buffer = Vec::new();
        reader.read_to_end(&mut raw_buffer)?;

        // Strip alignment padding up front so both passes below agree on where
        // fields start and end (the NULL pass indexes into these same bytes).
        let buffer = strip_field_padding(&raw_buffer, opts.delimiter);

        // First pass: Parse CSV with headers
        let mut csv_reader = ReaderBuilder::new()
            .has_headers(opts.has_headers)
            .delimiter(opts.delimiter)
            .from_reader(&buffer[..]);

        let headers = csv_reader.headers()?.clone();
        let mut table = DataTable::new(table_name);

        // Add metadata about the source
        table
            .metadata
            .insert("source_type".to_string(), source_type.to_string());
        table
            .metadata
            .insert("source_path".to_string(), source_path.to_string());
        table
            .metadata
            .insert("delimiter".to_string(), delimiter_label(opts.delimiter));

        // Create columns from headers
        for header in &headers {
            table.add_column(DataColumn::new(header));
        }

        // Collect all rows as strings
        let mut string_rows = Vec::new();
        for result in csv_reader.records() {
            let record = result?;
            let row: Vec<String> = record.iter().map(|s| s.to_string()).collect();
            string_rows.push(row);
        }

        // Analyze columns for string interning
        let analyses = self.analyze_columns(&string_rows, &headers);
        let categorical_columns: HashSet<usize> = analyses
            .iter()
            .filter(|a| a.is_categorical)
            .map(|a| a.index)
            .collect();

        info!(
            "Column analysis: {} of {} columns will use string interning",
            categorical_columns.len(),
            analyses.len()
        );

        // Initialize interners for categorical columns
        for col_idx in &categorical_columns {
            self.interners.insert(*col_idx, StringInterner::new());
        }

        // Second pass: Read raw lines for NULL detection
        let mut line_reader = BufReader::new(&buffer[..]);
        let mut raw_lines = Vec::new();
        let mut raw_line = String::new();

        // Skip header
        line_reader.read_line(&mut raw_line)?;
        raw_line.clear();

        // Read all raw lines
        for _ in 0..string_rows.len() {
            line_reader.read_line(&mut raw_line)?;
            raw_lines.push(raw_line.clone());
            raw_line.clear();
        }

        // Infer column types by sampling
        let mut column_types = vec![DataType::Null; headers.len()];
        let sample_size = string_rows.len().min(100);

        for row in string_rows.iter().take(sample_size) {
            for (col_idx, value) in row.iter().enumerate() {
                if !value.is_empty() {
                    let inferred = DataType::infer_from_string(value);
                    column_types[col_idx] = column_types[col_idx].merge(&inferred);
                }
            }
        }

        // Update column types
        for (col_idx, column) in table.columns.iter_mut().enumerate() {
            column.data_type = column_types[col_idx].clone();
        }

        // Convert strings to typed values and add rows
        for (row_idx, string_row) in string_rows.iter().enumerate() {
            let mut values = Vec::new();
            let raw_line = &raw_lines[row_idx];

            for (col_idx, value) in string_row.iter().enumerate() {
                let data_value = if value.is_empty() {
                    // Check if this is NULL (,,) vs empty string ("")
                    if is_null_field(raw_line, col_idx, opts.delimiter as char) {
                        DataValue::Null
                    } else if categorical_columns.contains(&col_idx) {
                        // Use interned string for empty categorical values
                        if let Some(interner) = self.interners.get_mut(&col_idx) {
                            DataValue::InternedString(interner.intern(""))
                        } else {
                            DataValue::String(String::new())
                        }
                    } else {
                        DataValue::String(String::new())
                    }
                } else if categorical_columns.contains(&col_idx)
                    && column_types[col_idx] == DataType::String
                {
                    // Use string interning for categorical columns
                    if let Some(interner) = self.interners.get_mut(&col_idx) {
                        DataValue::InternedString(interner.intern(value))
                    } else {
                        DataValue::from_string(value, &column_types[col_idx])
                    }
                } else {
                    DataValue::from_string(value, &column_types[col_idx])
                };
                values.push(data_value);
            }
            table
                .add_row(DataRow::new(values))
                .map_err(|e| anyhow::anyhow!(e))?;
        }

        // Print interner statistics
        for (col_idx, interner) in &self.interners {
            let stats = interner.stats();
            if stats.memory_saved_bytes > 0 {
                debug!(
                    "Column {} interning: {} unique strings, {} references, {} bytes saved",
                    headers.get(*col_idx).unwrap_or(&String::new()),
                    stats.unique_strings,
                    stats.total_references,
                    stats.memory_saved_bytes
                );
            }
        }

        // Update column statistics
        table.infer_column_types();

        Ok(table)
    }
}

/// Simple wrapper for loading CSV without advanced features. Defaults to
/// comma delimiter; for other delimiters use [`load_csv_from_reader_with_opts`].
pub fn load_csv_from_reader<R: Read>(
    reader: R,
    table_name: &str,
    source_type: &str,
    source_path: &str,
) -> Result<DataTable> {
    let mut loader = StreamCsvLoader::new();
    loader.load_csv_from_reader(reader, table_name, source_type, source_path)
}

/// As [`load_csv_from_reader`], but honouring caller-supplied [`CsvReadOptions`]
/// (delimiter, headers).
pub fn load_csv_from_reader_with_opts<R: Read>(
    reader: R,
    table_name: &str,
    source_type: &str,
    source_path: &str,
    opts: &CsvReadOptions,
) -> Result<DataTable> {
    let mut loader = StreamCsvLoader::new();
    loader.load_csv_from_reader_with_opts(reader, table_name, source_type, source_path, opts)
}

/// Parse JSON content as either a JSON array of objects or JSONL
/// (newline-delimited JSON, one object per line). The format is detected by
/// peeking at the first non-whitespace byte: `[` starts an array, anything
/// else is parsed line-by-line.
///
/// Empty and whitespace-only lines are skipped in JSONL mode. Parse errors
/// in JSONL mode include the source line number.
pub fn parse_json_records(content: &str) -> Result<Vec<JsonValue>> {
    let trimmed = content.trim_start();
    if trimmed.starts_with('[') {
        return serde_json::from_str(content).with_context(|| "Failed to parse JSON array");
    }

    let mut out = Vec::new();
    for (idx, raw_line) in content.lines().enumerate() {
        let line = raw_line.trim();
        if line.is_empty() {
            continue;
        }
        let value: JsonValue = serde_json::from_str(line)
            .with_context(|| format!("Failed to parse JSONL at line {}", idx + 1))?;
        out.push(value);
    }
    Ok(out)
}

/// Navigate to a sub-value of a JSON document using a dotted path.
///
/// This is the shared "find the rows" step used by both the WEB CTE
/// `JSON_PATH` clause and `READ_JSON(path, json_path)`. It only *locates* a
/// value; it deliberately does not filter, transform, or pluck scalars — that
/// is SQL's job (use a WHERE clause / column list once the rows are loaded).
/// Anything beyond locating the row set is out of scope; pipe through `jq`.
///
/// Supports two forms per dotted segment:
///   - `name`     — descend into an object key
///   - `name[]`   — descend into `name` (must be an array), then map the
///                  remainder of the path across every element. A bare `[]`
///                  projects over the current value when it is already an array.
///
/// Example for an Elasticsearch response:
///   `hits.hits[]._source`
/// returns an array of `_source` objects, one per hit — i.e. the `_source`
/// fields become the top-level row shape consumed by the loader.
///
/// An empty path (or one that is only dots) returns the value unchanged.
pub fn navigate_json_path(value: &JsonValue, path: &str) -> Result<JsonValue> {
    let parts: Vec<&str> = path.split('.').filter(|p| !p.is_empty()).collect();
    walk_json_path(value, &parts)
}

fn walk_json_path(value: &JsonValue, parts: &[&str]) -> Result<JsonValue> {
    let Some((head, tail)) = parts.split_first() else {
        return Ok(value.clone());
    };

    // Array projection: `name[]` (or bare `[]`) maps the rest of the path
    // across each element of an array.
    if let Some(name) = head.strip_suffix("[]") {
        let array_val = if name.is_empty() {
            value
        } else {
            value
                .get(name)
                .ok_or_else(|| anyhow::anyhow!("Path '{}' not found in JSON", name))?
        };
        let arr = array_val.as_array().ok_or_else(|| {
            anyhow::anyhow!(
                "Expected array at '{}' for [] projection, got {}",
                if name.is_empty() { "<root>" } else { name },
                json_kind(array_val)
            )
        })?;
        let mut projected = Vec::with_capacity(arr.len());
        for el in arr {
            projected.push(walk_json_path(el, tail)?);
        }
        return Ok(JsonValue::Array(projected));
    }

    let next = value
        .get(head)
        .ok_or_else(|| anyhow::anyhow!("Path '{}' not found in JSON", head))?;
    walk_json_path(next, tail)
}

/// Human-readable JSON type name, for error messages.
fn json_kind(value: &JsonValue) -> &'static str {
    match value {
        JsonValue::Null => "null",
        JsonValue::Bool(_) => "bool",
        JsonValue::Number(_) => "number",
        JsonValue::String(_) => "string",
        JsonValue::Array(_) => "array",
        JsonValue::Object(_) => "object",
    }
}

/// Compute the ordered union of object keys across the first `sample_size`
/// records. Order of first occurrence is preserved so the column layout is
/// stable. Non-object records are skipped.
pub fn collect_column_names(records: &[JsonValue], sample_size: usize) -> Vec<String> {
    let mut seen: HashSet<String> = HashSet::new();
    let mut names: Vec<String> = Vec::new();
    for record in records.iter().take(sample_size) {
        if let Some(obj) = record.as_object() {
            for key in obj.keys() {
                if seen.insert(key.clone()) {
                    names.push(key.clone());
                }
            }
        }
    }
    names
}

/// Load JSON data from any Read source into a DataTable.
///
/// Accepts either a JSON array of objects (`[{...}, {...}]`) or JSONL
/// (one JSON object per line). Format is auto-detected.
pub fn load_json_from_reader<R: Read>(
    mut reader: R,
    table_name: &str,
    source_type: &str,
    source_path: &str,
) -> Result<DataTable> {
    let mut json_str = String::new();
    reader.read_to_string(&mut json_str)?;

    let json_data: Vec<JsonValue> = parse_json_records(&json_str)?;

    if json_data.is_empty() {
        return Ok(DataTable::new(table_name));
    }

    // Schema is the union of keys across the first 100 records so heterogeneous
    // JSONL streams (where later records may carry fields the first one did
    // not) don't silently drop columns.
    let column_names = collect_column_names(&json_data, 100);
    if column_names.is_empty() {
        return Err(anyhow::anyhow!(
            "JSON data must contain objects (got non-object records)"
        ));
    }

    let mut table = DataTable::new(table_name);

    // Add metadata
    table
        .metadata
        .insert("source_type".to_string(), source_type.to_string());
    table
        .metadata
        .insert("source_path".to_string(), source_path.to_string());

    for name in &column_names {
        table.add_column(DataColumn::new(name));
    }

    // Collect values for type inference
    let mut string_rows = Vec::new();
    for json_obj in &json_data {
        if let Some(obj) = json_obj.as_object() {
            let mut row = Vec::new();
            for col_name in &column_names {
                let value = obj
                    .get(col_name)
                    .map(|v| json_value_to_string(v))
                    .unwrap_or_default();
                row.push(value);
            }
            string_rows.push(row);
        }
    }

    // Infer column types
    let mut column_types = vec![DataType::Null; column_names.len()];
    let sample_size = string_rows.len().min(100);

    for row in string_rows.iter().take(sample_size) {
        for (col_idx, value) in row.iter().enumerate() {
            if !value.is_empty() && value != "null" {
                let inferred = DataType::infer_from_string(value);
                column_types[col_idx] = column_types[col_idx].merge(&inferred);
            }
        }
    }

    // Update column types
    for (col_idx, column) in table.columns.iter_mut().enumerate() {
        column.data_type = column_types[col_idx].clone();
    }

    // Convert to typed values and add rows
    for string_row in &string_rows {
        let mut values = Vec::new();
        for (col_idx, value) in string_row.iter().enumerate() {
            let data_value = if value.is_empty() || value == "null" {
                DataValue::Null
            } else {
                DataValue::from_string(value, &column_types[col_idx])
            };
            values.push(data_value);
        }
        table
            .add_row(DataRow::new(values))
            .map_err(|e| anyhow::anyhow!(e))?;
    }

    // Update statistics
    table.infer_column_types();

    Ok(table)
}

/// Helper to convert JSON value to string for type inference
fn json_value_to_string(value: &JsonValue) -> String {
    match value {
        JsonValue::Null => String::new(),
        JsonValue::Bool(b) => b.to_string(),
        JsonValue::Number(n) => n.to_string(),
        JsonValue::String(s) => s.clone(),
        JsonValue::Array(arr) => format!("{:?}", arr),
        JsonValue::Object(obj) => format!("{:?}", obj),
    }
}

/// Helper to detect NULL fields in raw CSV lines. `delimiter` is the field
/// separator character used in the source (`,` for plain CSV, `\t` for TSV, etc.).
fn is_null_field(raw_line: &str, field_index: usize, delimiter: char) -> bool {
    let mut delim_count = 0;
    let mut in_quotes = false;
    let mut field_start = 0;
    let mut prev_char = ' ';

    for (i, ch) in raw_line.char_indices() {
        if ch == '"' && prev_char != '\\' {
            in_quotes = !in_quotes;
        } else if ch == delimiter && !in_quotes {
            if delim_count == field_index {
                // Found the field - check if it's empty
                return i == field_start
                    || (i == field_start + 1
                        && raw_line.chars().nth(field_start) == Some(delimiter));
            }
            delim_count += 1;
            field_start = i + 1;
        }
        prev_char = ch;
    }

    // Check last field
    if delim_count == field_index {
        let remaining = raw_line[field_start..].trim_end();
        return remaining.is_empty() || remaining.chars().next() == Some(delimiter);
    }

    false
}

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

    // ---- strip_field_padding tests ----

    fn stripped(input: &str, delimiter: u8) -> String {
        String::from_utf8(strip_field_padding(input.as_bytes(), delimiter).into_owned()).unwrap()
    }

    #[test]
    fn test_strip_padding_leaves_rfc4180_untouched() {
        let input = "Index,Name,Day\n1,\"George Washington\",22\n";
        // Nothing to strip => no allocation, bytes handed straight through.
        assert!(matches!(
            strip_field_padding(input.as_bytes(), b','),
            Cow::Borrowed(_)
        ));
        assert_eq!(stripped(input, b','), input);
    }

    #[test]
    fn test_strip_padding_unwraps_quotes_padded_after_delimiter() {
        // The president_birthdays.csv shape: quotes preceded by a space are
        // invisible to the parser until the padding is gone.
        let input = "\"Index\", \"Name\", \"Day\"\n 1, \"George Washington\",\t22\n";
        // Only the padding hugging a quote goes; ` 1` and `\t22` are unquoted and
        // stay put (type inference sees through their padding instead).
        assert_eq!(
            stripped(input, b','),
            "\"Index\",\"Name\",\"Day\"\n 1,\"George Washington\",\t22\n"
        );
    }

    #[test]
    fn test_strip_padding_preserves_whitespace_inside_quotes() {
        // Quoted padding is data, not alignment.
        let input = "a, \"  keep  \" ,b\n";
        assert_eq!(stripped(input, b','), "a,\"  keep  \",b\n");
    }

    #[test]
    fn test_strip_padding_leaves_unquoted_fields_alone() {
        // `data/test_simple_strings.csv` stores `  David  ` unquoted and tests
        // assert on those spaces, so an unquoted field is never touched here.
        let input = "4,  David  ,x\n";
        assert_eq!(stripped(input, b','), input);
    }

    #[test]
    fn test_strip_padding_preserves_escaped_quotes() {
        let input = " \"he said \"\"hi\"\"\" ,1\n";
        assert_eq!(stripped(input, b','), "\"he said \"\"hi\"\"\",1\n");
    }

    #[test]
    fn test_strip_padding_never_eats_a_tab_delimiter() {
        // In a .tsv the tab is the delimiter, so only spaces can be padding.
        let input = "a\t \"b\" \tc\n";
        assert_eq!(stripped(input, b'\t'), "a\t\"b\"\tc\n");
    }

    #[test]
    fn test_strip_padding_recovers_delimiter_inside_padded_quotes() {
        // Padding hid the quoting, so this used to split into three fields.
        let input = "1, \"Adams, John\", 2\n";
        let out = stripped(input, b',');
        assert_eq!(out, "1,\"Adams, John\", 2\n");

        let mut rdr = ReaderBuilder::new()
            .has_headers(false)
            .from_reader(out.as_bytes());
        let rec = rdr.records().next().unwrap().unwrap();
        assert_eq!(rec.len(), 3);
        assert_eq!(&rec[1], "Adams, John");
    }

    #[test]
    fn test_strip_padding_handles_crlf_and_end_of_line_pad() {
        let input = " \"a\" , \"b\" \r\n \"c\" , \"d\" \r\n";
        assert_eq!(stripped(input, b','), "\"a\",\"b\"\r\n\"c\",\"d\"\r\n");
    }

    #[test]
    fn test_padded_csv_loads_with_clean_headers_and_numeric_types() {
        let csv = "\"Index\", \"Name\", \"Year\"\n 1, \"George Washington\", 1732\n 2, \"John Adams\", 1735\n";
        let table = StreamCsvLoader::new()
            .load_csv_from_reader(Cursor::new(csv), "pres", "test", "<test>")
            .unwrap();

        let names: Vec<&str> = table.columns.iter().map(|c| c.name.as_str()).collect();
        assert_eq!(names, vec!["Index", "Name", "Year"]);
        // Padding used to make every column a string; Year should now be numeric.
        assert_eq!(table.columns[2].data_type, DataType::Integer);
        assert_eq!(
            table.rows[0].values[1],
            DataValue::String("George Washington".to_string())
        );
    }

    // ---- navigate_json_path tests ----

    #[test]
    fn test_navigate_json_path_descends_object_key() {
        // The TeamCity-style case: object with a nested array one level down.
        let doc = serde_json::json!({
            "count": 2,
            "project": [{"id": "a"}, {"id": "b"}]
        });
        let extracted = navigate_json_path(&doc, "project").unwrap();
        assert!(extracted.is_array());
        assert_eq!(extracted.as_array().unwrap().len(), 2);
    }

    #[test]
    fn test_navigate_json_path_nested_descent() {
        // TeamCity actually wraps as { projects: { project: [...] } }.
        let doc = serde_json::json!({
            "projects": {"project": [{"id": "a"}, {"id": "b"}, {"id": "c"}]}
        });
        let extracted = navigate_json_path(&doc, "projects.project").unwrap();
        assert_eq!(extracted.as_array().unwrap().len(), 3);
    }

    #[test]
    fn test_navigate_json_path_array_projection() {
        // Elasticsearch-style: project _source out of each hit.
        let doc = serde_json::json!({
            "hits": {"hits": [
                {"_source": {"id": 1}},
                {"_source": {"id": 2}}
            ]}
        });
        let extracted = navigate_json_path(&doc, "hits.hits[]._source").unwrap();
        let arr = extracted.as_array().unwrap();
        assert_eq!(arr.len(), 2);
        assert_eq!(arr[1]["id"], serde_json::json!(2));
    }

    #[test]
    fn test_navigate_json_path_bare_projection_over_root_array() {
        let doc = serde_json::json!([{"v": {"x": 1}}, {"v": {"x": 2}}]);
        let extracted = navigate_json_path(&doc, "[].v").unwrap();
        let arr = extracted.as_array().unwrap();
        assert_eq!(arr[0]["x"], serde_json::json!(1));
    }

    #[test]
    fn test_navigate_json_path_empty_path_is_identity() {
        let doc = serde_json::json!({"a": 1});
        let extracted = navigate_json_path(&doc, "").unwrap();
        assert_eq!(extracted, doc);
    }

    #[test]
    fn test_navigate_json_path_missing_key_errors() {
        let doc = serde_json::json!({"a": 1});
        let err = navigate_json_path(&doc, "b").unwrap_err();
        assert!(err.to_string().contains("not found"), "{}", err);
    }

    #[test]
    fn test_navigate_json_path_projection_on_non_array_errors() {
        let doc = serde_json::json!({"a": {"not": "an array"}});
        let err = navigate_json_path(&doc, "a[]").unwrap_err();
        let msg = err.to_string();
        assert!(msg.contains("Expected array"), "{}", msg);
        assert!(msg.contains("object"), "{}", msg);
    }

    #[test]
    fn test_csv_from_reader() {
        let csv_data = "id,name,value\n1,Alice,100\n2,Bob,200\n3,,300";
        let reader = Cursor::new(csv_data);

        let table =
            load_csv_from_reader(reader, "test", "stream", "memory").expect("Failed to load CSV");

        assert_eq!(table.name, "test");
        assert_eq!(table.column_count(), 3);
        assert_eq!(table.row_count(), 3);

        // Check that empty field is NULL
        let value = table.get_value(2, 1).unwrap();
        assert!(matches!(value, DataValue::Null));
    }

    #[test]
    fn test_json_from_reader() {
        let json_data = r#"[
            {"id": 1, "name": "Alice", "value": 100},
            {"id": 2, "name": "Bob", "value": 200},
            {"id": 3, "name": null, "value": 300}
        ]"#;
        let reader = Cursor::new(json_data);

        let table =
            load_json_from_reader(reader, "test", "stream", "memory").expect("Failed to load JSON");

        assert_eq!(table.name, "test");
        assert_eq!(table.column_count(), 3);
        assert_eq!(table.row_count(), 3);

        // Check that null is handled
        let value = table.get_value(2, 1).unwrap();
        assert!(matches!(value, DataValue::Null));
    }

    #[test]
    fn test_jsonl_from_reader() {
        let jsonl_data = "{\"id\":1,\"name\":\"Alice\"}\n{\"id\":2,\"name\":\"Bob\"}\n";
        let reader = Cursor::new(jsonl_data);

        let table = load_json_from_reader(reader, "test", "stream", "memory")
            .expect("Failed to load JSONL");

        assert_eq!(table.column_count(), 2);
        assert_eq!(table.row_count(), 2);
    }

    #[test]
    fn test_jsonl_heterogeneous_schema_unioned() {
        // Second record adds an "extra" field; loader should pick it up via the
        // union, and row 0 should have Null for it.
        let jsonl_data = "{\"id\":1}\n{\"id\":2,\"extra\":\"hi\"}\n";
        let reader = Cursor::new(jsonl_data);
        let table = load_json_from_reader(reader, "test", "stream", "memory").expect("load");
        assert_eq!(table.column_count(), 2);
        assert_eq!(table.row_count(), 2);
    }

    #[test]
    fn test_jsonl_skips_blank_lines() {
        let jsonl_data = "{\"id\":1}\n\n\n{\"id\":2}\n";
        let reader = Cursor::new(jsonl_data);
        let table = load_json_from_reader(reader, "test", "stream", "memory").expect("load");
        assert_eq!(table.row_count(), 2);
    }

    #[test]
    fn test_parse_json_records_array_form() {
        let recs = parse_json_records(r#"[{"a":1},{"a":2}]"#).unwrap();
        assert_eq!(recs.len(), 2);
    }

    #[test]
    fn test_parse_json_records_jsonl_form() {
        let recs = parse_json_records("{\"a\":1}\n{\"a\":2}\n").unwrap();
        assert_eq!(recs.len(), 2);
    }

    #[test]
    fn test_parse_json_records_jsonl_error_cites_line() {
        let err = parse_json_records("{\"a\":1}\nnot json\n").unwrap_err();
        assert!(err.to_string().contains("line 2"));
    }

    // ---- CsvReadOptions / delimiter detection ----

    #[test]
    fn test_csv_options_default_is_comma() {
        let opts = CsvReadOptions::default();
        assert_eq!(opts.delimiter, b',');
        assert!(opts.has_headers);
    }

    #[test]
    fn test_detect_delimiter_from_path() {
        assert_eq!(detect_delimiter_from_path("data.tsv"), b'\t');
        assert_eq!(detect_delimiter_from_path("data.TSV"), b'\t');
        assert_eq!(detect_delimiter_from_path("/tmp/foo.psv"), b'|');
        assert_eq!(detect_delimiter_from_path("data.PSV"), b'|');
        assert_eq!(detect_delimiter_from_path("data.csv"), b',');
        assert_eq!(detect_delimiter_from_path("noext"), b',');
        assert_eq!(detect_delimiter_from_path("-"), b',');
    }

    #[test]
    fn test_load_csv_with_pipe_delimiter() {
        let data = "id|name|score\n1|alice|10\n2|bob|20\n";
        let reader = Cursor::new(data);
        let opts = CsvReadOptions {
            delimiter: b'|',
            has_headers: true,
        };
        let table = load_csv_from_reader_with_opts(reader, "psv", "test", "memory", &opts)
            .expect("load failed");
        assert_eq!(table.column_count(), 3);
        assert_eq!(table.row_count(), 2);
        assert_eq!(table.get_value(0, 0).unwrap(), &DataValue::Integer(1));
        assert_eq!(
            table.get_value(1, 1).unwrap(),
            &DataValue::String("bob".to_string())
        );
    }

    #[test]
    fn test_load_csv_with_tab_delimiter() {
        let data = "id\tname\tscore\n1\talice\t10\n2\tbob\t20\n";
        let reader = Cursor::new(data);
        let opts = CsvReadOptions {
            delimiter: b'\t',
            has_headers: true,
        };
        let table = load_csv_from_reader_with_opts(reader, "tsv", "test", "memory", &opts)
            .expect("load failed");
        assert_eq!(table.column_count(), 3);
        assert_eq!(table.row_count(), 2);
        assert_eq!(table.get_value(0, 0).unwrap(), &DataValue::Integer(1));
    }

    #[test]
    fn test_metadata_records_delimiter() {
        // Comma -> stored as ","
        let table = load_csv_from_reader(Cursor::new("a,b\n1,2\n"), "t", "test", "memory").unwrap();
        assert_eq!(
            table.metadata.get("delimiter").map(String::as_str),
            Some(",")
        );

        // Tab -> stored as "\t"
        let opts = CsvReadOptions {
            delimiter: b'\t',
            has_headers: true,
        };
        let table = load_csv_from_reader_with_opts(
            Cursor::new("a\tb\n1\t2\n"),
            "t",
            "test",
            "memory",
            &opts,
        )
        .unwrap();
        assert_eq!(
            table.metadata.get("delimiter").map(String::as_str),
            Some("\\t")
        );
    }

    #[test]
    fn test_parse_delimiter_arg_accepts_single_char() {
        assert_eq!(parse_delimiter_arg(",").unwrap(), b',');
        assert_eq!(parse_delimiter_arg("|").unwrap(), b'|');
        assert_eq!(parse_delimiter_arg(";").unwrap(), b';');
    }

    #[test]
    fn test_parse_delimiter_arg_accepts_backslash_escapes() {
        assert_eq!(parse_delimiter_arg("\\t").unwrap(), b'\t');
        assert_eq!(parse_delimiter_arg("\t").unwrap(), b'\t');
        assert_eq!(parse_delimiter_arg("\\n").unwrap(), b'\n');
        assert_eq!(parse_delimiter_arg("\\r").unwrap(), b'\r');
    }

    #[test]
    fn test_parse_delimiter_arg_rejects_multi_char() {
        let err = parse_delimiter_arg("||").unwrap_err();
        assert!(err.to_string().contains("single ASCII character"));
    }

    #[test]
    fn test_parse_delimiter_arg_rejects_non_ascii() {
        let err = parse_delimiter_arg("ö").unwrap_err();
        assert!(err.to_string().contains("single ASCII character"));
    }

    #[test]
    fn test_resolve_delimiter_explicit_wins() {
        assert_eq!(resolve_delimiter("data.psv", Some(b',')), b',');
        assert_eq!(resolve_delimiter("data.tsv", Some(b';')), b';');
        assert_eq!(resolve_delimiter("data.csv", Some(b'|')), b'|');
    }

    #[test]
    fn test_resolve_delimiter_falls_back_to_extension() {
        assert_eq!(resolve_delimiter("data.psv", None), b'|');
        assert_eq!(resolve_delimiter("data.tsv", None), b'\t');
        assert_eq!(resolve_delimiter("data.csv", None), b',');
        assert_eq!(resolve_delimiter("data.dat", None), b',');
    }

    #[test]
    fn test_null_detection_works_with_pipe_delimiter() {
        // Middle column is unquoted-empty -> NULL, not empty string.
        let data = "id|name|score\n1||10\n";
        let opts = CsvReadOptions {
            delimiter: b'|',
            has_headers: true,
        };
        let table =
            load_csv_from_reader_with_opts(Cursor::new(data), "psv", "test", "memory", &opts)
                .expect("load failed");
        assert!(matches!(table.get_value(0, 1).unwrap(), DataValue::Null));
    }
}