query-forge 0.3.0

use std::path::Path;
use std::{collections::HashSet, fmt::Write as _, fs, sync::OnceLock};

use anyhow::{Context, Result, anyhow, bail};
use calamine::{Data, Reader, open_workbook_auto};
use chrono::{NaiveDate, NaiveDateTime};
use parquet::{
    file::reader::{FileReader, SerializedFileReader},
    record::Field,
};
use regex::Regex;
use roxmltree::{Document, Node};
use rusqlite::{
    Connection, params_from_iter,
    types::{Value, ValueRef},
};
use rust_xlsxwriter::Workbook;
use serde_json::Value as JsonValue;

#[derive(Debug, Clone, PartialEq)]
pub enum QueryValue {
    Null,
    Integer(i64),
    Real(f64),
    Text(String),
}

#[derive(Debug, Clone, PartialEq)]
pub struct QueryResult {
    pub columns: Vec<String>,
    pub rows: Vec<Vec<QueryValue>>,
}

#[derive(Debug, Clone, PartialEq)]
pub struct QueryParam {
    pub name: String,
    pub value: QueryValue,
}

#[derive(Debug, Clone, Copy)]
pub struct WorkbookInput<'a> {
    pub path: &'a Path,
    pub sheet_name: Option<&'a str>,
    pub table_name: Option<&'a str>,
}

#[derive(Debug, Clone)]
pub struct TypeInferenceOptions {
    pub infer_types: bool,
    pub decimal_comma: bool,
    pub date_format: Option<String>,
    pub null_values: Vec<String>,
    pub true_values: Vec<String>,
    pub false_values: Vec<String>,
}

impl Default for TypeInferenceOptions {
    fn default() -> Self {
        Self {
            infer_types: true,
            decimal_comma: false,
            date_format: None,
            null_values: vec![String::new()],
            true_values: vec!["true".to_owned()],
            false_values: vec!["false".to_owned()],
        }
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub enum HeaderCase {
    Snake,
}

#[derive(Debug, Clone, Default)]
pub struct InputNormalizationOptions {
    pub trim: bool,
    pub skip_empty_rows: bool,
    pub normalize_headers: bool,
    pub header_case: Option<HeaderCase>,
    pub dedupe_headers: bool,
}

/// Controls how JSON input files are parsed into rows.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum JsonMode {
    /// Each element of a top-level JSON array becomes a row (default).
    ///
    /// If the root is an object the object itself becomes a single row.
    #[default]
    Array,
    /// Each key-value pair of a JSON object becomes a row with "key" and "value" columns.
    ///
    /// Useful for flat configuration or metadata objects.
    Object,
    /// Recursively flatten nested JSON objects and arrays into a single row per top-level
    /// element, using dotted key paths (e.g. `address.city`) for nested fields.
    Flatten,
}

/// Controls how XML input files are parsed into rows.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum XmlMode {
    /// Detect and extract tabular rows from the XML structure (default).
    ///
    /// Looks first for direct `<row>` children, then for any element whose direct
    /// children are all leaf elements (no deeper nesting).
    #[default]
    Rows,
    /// Collect every leaf text element in the document as a separate row.
    ///
    /// Each row has two columns: `tag` (element name) and `value` (text content).
    Descendants,
    /// Extract XML element attributes as columns.
    ///
    /// Every element that carries at least one XML attribute becomes a row; attribute
    /// names map to column names. Text content of the element is also included as a
    /// `value` column when present.
    Attributes,
}

/// Options that control how JSON and XML inputs are extracted into rows.
#[derive(Debug, Clone, Default)]
pub struct ExtractionOptions {
    pub json_mode: JsonMode,
    pub xml_mode: XmlMode,
}

/// The SQL-compatible type inferred for a column.
#[derive(Debug, Clone, PartialEq)]
pub enum InferredType {
    Integer,
    Real,
    Text,
    /// Column contains a mix of numeric and text values.
    Mixed,
}

impl InferredType {
    pub fn as_str(&self) -> &'static str {
        match self {
            InferredType::Integer => "INTEGER",
            InferredType::Real => "REAL",
            InferredType::Text => "TEXT",
            InferredType::Mixed => "MIXED",
        }
    }
}

/// Per-column information returned by inspection commands.
#[derive(Debug, Clone)]
pub struct ColumnInfo {
    pub table_name: String,
    pub column_name: String,
    pub inferred_type: InferredType,
    pub nullable: bool,
}

/// Optional per-column statistics (enabled by `--stats`).
#[derive(Debug, Clone)]
pub struct ColumnStats {
    pub distinct_count: usize,
    pub min_value: Option<String>,
    pub max_value: Option<String>,
}

/// Summary of a single loaded table, returned by `load_table_summaries`.
#[derive(Debug, Clone)]
pub struct TableSummary {
    /// The logical SQL table name (explicit or auto-generated).
    pub table_name: String,
    /// The original file path as provided on the command line.
    pub source_path: String,
    /// The sheet/tag/key/index selector, if any.
    pub source_selector: Option<String>,
    pub row_count: usize,
    pub columns: Vec<ColumnInfo>,
    /// Up to `sample` rows of actual data (column values in column order).
    pub sample_rows: Vec<Vec<QueryValue>>,
    /// Non-fatal warnings (e.g. mixed-type columns, duplicate header normalisations).
    pub warnings: Vec<String>,
    /// Per-column statistics; `None` when `--stats` is not requested.
    pub column_stats: Option<Vec<ColumnStats>>,
}

/// Load metadata for every input without running a SQL query.
///
/// Returns one [`TableSummary`] per input in declaration order.
pub fn load_table_summaries(
    workbook_inputs: &[WorkbookInput<'_>],
    sample: usize,
    include_stats: bool,
) -> Result<Vec<TableSummary>> {
    let mut summaries = Vec::with_capacity(workbook_inputs.len());
    let inference_options = TypeInferenceOptions::default();

    for (index, workbook_input) in workbook_inputs.iter().enumerate() {
        let sheet = load_input(
            workbook_input.path,
            workbook_input.sheet_name,
            &inference_options,
            &ExtractionOptions::default(),
            true,
        )?;

        let table_name = workbook_input
            .table_name
            .map(str::to_owned)
            .unwrap_or_else(|| {
                if index == 0 {
                    "table".to_owned()
                } else {
                    format!("table{}", index + 1)
                }
            });

        let columns = infer_column_infos(&table_name, &sheet);
        let mut warnings = collect_warnings(&table_name, &sheet, &columns);

        let sample_rows = sheet.rows.iter().take(sample).cloned().collect();

        let column_stats = if include_stats {
            Some(compute_column_stats(&sheet))
        } else {
            None
        };

        // Warn about empty tables (non-fatal for inspection).
        if sheet.rows.is_empty() {
            warnings.push(format!(
                "table '{}' loaded from '{}' contains no data rows",
                table_name,
                workbook_input.path.display()
            ));
        }

        summaries.push(TableSummary {
            table_name,
            source_path: workbook_input.path.display().to_string(),
            source_selector: workbook_input.sheet_name.map(str::to_owned),
            row_count: sheet.rows.len(),
            columns,
            sample_rows,
            warnings,
            column_stats,
        });
    }

    Ok(summaries)
}

/// Infer the SQL-compatible type and nullability for each column in `sheet`.
fn infer_column_infos(table_name: &str, sheet: &SheetData) -> Vec<ColumnInfo> {
    sheet
        .columns
        .iter()
        .enumerate()
        .map(|(col_idx, col_name)| {
            let mut has_integer = false;
            let mut has_real = false;
            let mut has_text = false;
            let mut has_null = false;

            for row in &sheet.rows {
                match row.get(col_idx).unwrap_or(&QueryValue::Null) {
                    QueryValue::Null => has_null = true,
                    QueryValue::Integer(_) => has_integer = true,
                    QueryValue::Real(_) => has_real = true,
                    QueryValue::Text(_) => has_text = true,
                }
            }

            let inferred_type = match (has_integer, has_real, has_text) {
                (true, false, false) => InferredType::Integer,
                (false, false, true) => InferredType::Text,
                (false, false, false) => InferredType::Text,
                // Integer+Real and Real-only both map to REAL: integers are
                // promotable to real without loss, so REAL is the wider type.
                (_, true, false) => InferredType::Real,
                _ => InferredType::Mixed,
            };

            ColumnInfo {
                table_name: table_name.to_owned(),
                column_name: col_name.clone(),
                inferred_type,
                nullable: has_null,
            }
        })
        .collect()
}

/// Collect non-fatal warnings for a loaded table.
fn collect_warnings(table_name: &str, sheet: &SheetData, columns: &[ColumnInfo]) -> Vec<String> {
    let mut warnings = Vec::new();

    for col_info in columns {
        if col_info.inferred_type == InferredType::Mixed {
            warnings.push(format!(
                "column '{}' in table '{}' has mixed types (INTEGER/REAL and TEXT values)",
                col_info.column_name, table_name
            ));
        }
    }

    // Detect duplicate original column names that were de-duplicated by normalisation.
    let mut seen_names: std::collections::HashMap<String, usize> = std::collections::HashMap::new();
    for col in &sheet.columns {
        *seen_names.entry(col.clone()).or_insert(0) += 1;
    }
    for (name, count) in &seen_names {
        if *count > 1 {
            warnings.push(format!(
                "column name '{}' appears {} times in table '{}'; duplicates were renamed",
                name, count, table_name
            ));
        }
    }

    warnings
}

/// Compute per-column distinct counts and min/max values.
fn compute_column_stats(sheet: &SheetData) -> Vec<ColumnStats> {
    sheet
        .columns
        .iter()
        .enumerate()
        .map(|(col_idx, _)| {
            let mut distinct: std::collections::HashSet<String> = std::collections::HashSet::new();
            // Track min/max as f64 to handle both Integer and Real columns
            // uniformly without reparsing strings.
            let mut min_num: Option<f64> = None;
            let mut max_num: Option<f64> = None;
            // Remember whether the extremes came from an integer so we can
            // render them without a spurious decimal point.
            let mut min_as_int: Option<i64> = None;
            let mut max_as_int: Option<i64> = None;

            for row in &sheet.rows {
                let value = row.get(col_idx).unwrap_or(&QueryValue::Null);
                if matches!(value, QueryValue::Null) {
                    continue;
                }
                distinct.insert(display_value(value));

                let (as_f64, as_int) = match value {
                    QueryValue::Integer(n) => (*n as f64, Some(*n)),
                    QueryValue::Real(n) => (*n, None),
                    _ => continue,
                };

                if min_num.map_or(true, |m| as_f64 < m) {
                    min_num = Some(as_f64);
                    min_as_int = as_int;
                }
                if max_num.map_or(true, |m| as_f64 > m) {
                    max_num = Some(as_f64);
                    max_as_int = as_int;
                }
            }

            let min_value =
                min_num.map(|v| min_as_int.map_or_else(|| v.to_string(), |i| i.to_string()));
            let max_value =
                max_num.map(|v| max_as_int.map_or_else(|| v.to_string(), |i| i.to_string()));

            ColumnStats {
                distinct_count: distinct.len(),
                min_value,
                max_value,
            }
        })
        .collect()
}

pub fn run_query(
    workbook_path: &Path,
    sheet_name: Option<&str>,
    query: &str,
    has_headers: bool,
) -> Result<QueryResult> {
    run_query_with_params(workbook_path, sheet_name, query, &[], has_headers)
}

pub fn run_query_with_params(
    workbook_path: &Path,
    sheet_name: Option<&str>,
    query: &str,
    params: &[QueryParam],
    has_headers: bool,
) -> Result<QueryResult> {
    run_query_with_params_multi(&[workbook_path], sheet_name, query, params, has_headers)
}

pub fn run_query_with_params_multi(
    workbook_paths: &[&Path],
    sheet_name: Option<&str>,
    query: &str,
    params: &[QueryParam],
    has_headers: bool,
) -> Result<QueryResult> {
    let workbook_inputs = workbook_paths
        .iter()
        .map(|path| WorkbookInput {
            path,
            sheet_name,
            table_name: None,
        })
        .collect::<Vec<_>>();

    run_query_with_params_multi_inputs(&workbook_inputs, query, params, has_headers)
}

pub fn run_query_with_params_multi_inputs(
    workbook_inputs: &[WorkbookInput<'_>],
    query: &str,
    params: &[QueryParam],
    has_headers: bool,
) -> Result<QueryResult> {
    run_query_with_params_multi_inputs_and_options(
        workbook_inputs,
        query,
        params,
        &TypeInferenceOptions::default(),
        has_headers,
    )
}

pub fn run_query_with_params_multi_inputs_and_options(
    workbook_inputs: &[WorkbookInput<'_>],
    query: &str,
    params: &[QueryParam],
    inference_options: &TypeInferenceOptions,
    has_headers: bool,
) -> Result<QueryResult> {
    run_query_with_params_multi_inputs_and_options_and_normalization(
        workbook_inputs,
        query,
        params,
        inference_options,
        &InputNormalizationOptions::default(),
        &ExtractionOptions::default(),
        has_headers,
    )
}

pub fn run_query_with_params_multi_inputs_and_options_and_normalization(
    workbook_inputs: &[WorkbookInput<'_>],
    query: &str,
    params: &[QueryParam],
    inference_options: &TypeInferenceOptions,
    normalization_options: &InputNormalizationOptions,
    extraction_options: &ExtractionOptions,
    has_headers: bool,
) -> Result<QueryResult> {
    if workbook_inputs.is_empty() {
        bail!("at least one workbook input is required");
    }

    let connection =
        Connection::open_in_memory().context("failed to create in-memory SQLite database")?;
    let mut registered_sheet_views = HashSet::new();

    for (index, workbook_input) in workbook_inputs.iter().enumerate() {
        let mut sheet = load_input(
            workbook_input.path,
            workbook_input.sheet_name,
            inference_options,
            extraction_options,
            has_headers,
        )?;
        apply_input_normalization(&mut sheet, normalization_options);
        let table_name = workbook_input
            .table_name
            .map(str::to_owned)
            .unwrap_or_else(|| {
                if index == 0 {
                    "table".to_owned()
                } else {
                    format!("table{}", index + 1)
                }
            });

        register_sheet(
            &connection,
            &sheet,
            &table_name,
            &mut registered_sheet_views,
        )?;
    }

    execute_query(&connection, query, params)
}

pub fn render_text(result: &QueryResult) -> String {
    if result.columns.is_empty() {
        return String::new();
    }

    let mut column_widths = result
        .columns
        .iter()
        .map(|column| column.len())
        .collect::<Vec<_>>();
    let mut rendered_rows = Vec::with_capacity(result.rows.len());

    for row in &result.rows {
        let mut rendered_row = Vec::with_capacity(result.columns.len());
        for index in 0..result.columns.len() {
            let rendered_value = row.get(index).map(display_value).unwrap_or_default();
            column_widths[index] = column_widths[index].max(rendered_value.len());
            rendered_row.push(rendered_value);
        }
        rendered_rows.push(rendered_row);
    }

    let mut output = String::new();
    write_aligned_row(&mut output, &result.columns, &column_widths);
    output.push('\n');
    output.push_str(
        &column_widths
            .iter()
            .map(|width| "-".repeat(*width))
            .collect::<Vec<_>>()
            .join("-+-"),
    );

    for row in rendered_rows {
        output.push('\n');
        write_aligned_row(&mut output, &row, &column_widths);
    }

    output
}

fn write_aligned_row(output: &mut String, values: &[String], column_widths: &[usize]) {
    for (index, value) in values.iter().enumerate() {
        if index > 0 {
            output.push_str(" | ");
        }

        let _ = write!(output, "{value:<width$}", width = column_widths[index]);
    }
}

pub fn render_csv(result: &QueryResult) -> String {
    let mut output = String::new();
    output.push_str(
        &result
            .columns
            .iter()
            .map(|column| escape_csv_field(column))
            .collect::<Vec<_>>()
            .join(","),
    );

    for row in &result.rows {
        output.push('\n');
        output.push_str(
            &row.iter()
                .map(display_value)
                .map(|value| escape_csv_field(&value))
                .collect::<Vec<_>>()
                .join(","),
        );
    }

    output
}

pub fn render_jsonl(result: &QueryResult) -> String {
    let mut output = String::new();

    for (row_index, row) in result.rows.iter().enumerate() {
        if row_index > 0 {
            output.push('\n');
        }
        output.push('{');
        for (column_index, column) in result.columns.iter().enumerate() {
            if column_index > 0 {
                output.push(',');
            }
            output.push_str(&escape_json_string(column));
            output.push(':');
            output.push_str(&to_json_value(
                row.get(column_index).unwrap_or(&QueryValue::Null),
            ));
        }
        output.push('}');
    }

    output
}

pub fn render_json(result: &QueryResult) -> String {
    let mut output = String::from("[");

    for (row_index, row) in result.rows.iter().enumerate() {
        if row_index > 0 {
            output.push(',');
        }
        output.push('{');
        for (column_index, column) in result.columns.iter().enumerate() {
            if column_index > 0 {
                output.push(',');
            }
            output.push_str(&escape_json_string(column));
            output.push(':');
            output.push_str(&to_json_value(
                row.get(column_index).unwrap_or(&QueryValue::Null),
            ));
        }
        output.push('}');
    }

    output.push(']');
    output
}

pub fn render_markdown(result: &QueryResult) -> String {
    let mut output = String::new();

    output.push('|');
    for column in &result.columns {
        output.push(' ');
        output.push_str(&escape_markdown_cell(column));
        output.push(' ');
        output.push('|');
    }
    output.push('\n');

    output.push('|');
    for _ in &result.columns {
        output.push_str(" --- |");
    }

    for row in &result.rows {
        output.push('\n');
        output.push('|');
        for column_index in 0..result.columns.len() {
            let value = row.get(column_index).unwrap_or(&QueryValue::Null);
            output.push(' ');
            output.push_str(&escape_markdown_cell(&display_value(value)));
            output.push(' ');
            output.push('|');
        }
    }

    output
}

pub fn render_xml(result: &QueryResult) -> String {
    let mut output = String::from("<?xml version=\"1.0\" encoding=\"UTF-8\"?>\n<data>\n");

    for row in &result.rows {
        output.push_str("  <row>\n");
        for (column_index, column) in result.columns.iter().enumerate() {
            let value = row.get(column_index).unwrap_or(&QueryValue::Null);
            let xml_column = escape_xml_tag(column);
            let xml_value = escape_xml_text(&display_value(value));
            output.push_str(&format!(
                "    <{}>{}</{}>
",
                xml_column, xml_value, xml_column
            ));
        }
        output.push_str("  </row>\n");
    }

    output.push_str("</data>");
    output
}

pub fn write_xlsx(result: &QueryResult, output_path: &Path) -> Result<()> {
    let mut workbook = Workbook::new();
    let worksheet = workbook.add_worksheet();

    for (column, header) in result.columns.iter().enumerate() {
        worksheet.write_string(0, column as u16, header)?;
    }

    for (row_index, row) in result.rows.iter().enumerate() {
        for (column_index, value) in row.iter().enumerate() {
            let excel_row = (row_index + 1) as u32;
            let excel_column = column_index as u16;

            match value {
                QueryValue::Null => {}
                QueryValue::Integer(number) => {
                    worksheet.write_number(excel_row, excel_column, *number as f64)?;
                }
                QueryValue::Real(number) => {
                    worksheet.write_number(excel_row, excel_column, *number)?;
                }
                QueryValue::Text(text) => {
                    worksheet.write_string(excel_row, excel_column, text)?;
                }
            }
        }
    }

    workbook
        .save(output_path)
        .with_context(|| format!("failed to write {}", output_path.display()))
}
pub fn write_parquet(result: &QueryResult, output_path: &Path) -> Result<()> {
    use std::sync::Arc;

    use parquet::basic::{ConvertedType, Repetition, Type as PhysicalType};
    use parquet::column::writer::ColumnWriter;
    use parquet::data_type::ByteArray;
    use parquet::file::properties::WriterProperties;
    use parquet::file::writer::SerializedFileWriter;
    use parquet::schema::types::Type as ParquetType;

    #[derive(Clone, Copy)]
    enum ColKind {
        Int64,
        Double,
        Bytes,
    }

    // Infer the physical type for each column from its values.
    let col_kinds: Vec<ColKind> = (0..result.columns.len())
        .map(|ci| {
            let mut has_real = false;
            let mut has_text = false;
            for row in &result.rows {
                match row.get(ci).unwrap_or(&QueryValue::Null) {
                    QueryValue::Real(_) => has_real = true,
                    QueryValue::Text(_) => has_text = true,
                    _ => {}
                }
            }
            if has_text {
                ColKind::Bytes
            } else if has_real {
                ColKind::Double
            } else {
                ColKind::Int64
            }
        })
        .collect();

    // Build the Parquet schema.
    let fields: Vec<Arc<ParquetType>> = result
        .columns
        .iter()
        .zip(&col_kinds)
        .map(|(name, kind)| {
            let physical = match kind {
                ColKind::Int64 => PhysicalType::INT64,
                ColKind::Double => PhysicalType::DOUBLE,
                ColKind::Bytes => PhysicalType::BYTE_ARRAY,
            };
            let mut builder = ParquetType::primitive_type_builder(name, physical)
                .with_repetition(Repetition::OPTIONAL);
            if matches!(kind, ColKind::Bytes) {
                builder = builder.with_converted_type(ConvertedType::UTF8);
            }
            Arc::new(
                builder
                    .build()
                    .with_context(|| format!("failed to build Parquet field '{name}'"))
                    .expect("valid field"),
            )
        })
        .collect();

    let schema = Arc::new(
        ParquetType::group_type_builder("schema")
            .with_fields(fields)
            .build()
            .context("failed to build Parquet schema")?,
    );

    let props = Arc::new(WriterProperties::builder().build());
    let file = fs::File::create(output_path)
        .with_context(|| format!("failed to create {}", output_path.display()))?;
    let mut file_writer = SerializedFileWriter::new(file, schema, props)
        .context("failed to initialize Parquet writer")?;

    let mut rg = file_writer
        .next_row_group()
        .context("failed to start Parquet row group")?;

    for (col_idx, kind) in col_kinds.iter().enumerate() {
        let def_levels: Vec<i16> = result
            .rows
            .iter()
            .map(|row| match row.get(col_idx).unwrap_or(&QueryValue::Null) {
                QueryValue::Null => 0,
                _ => 1,
            })
            .collect();

        let Some(mut col_writer) = rg
            .next_column()
            .context("failed to open Parquet column writer")?
        else {
            break;
        };

        match (kind, col_writer.untyped()) {
            (ColKind::Int64, ColumnWriter::Int64ColumnWriter(w)) => {
                let values: Vec<i64> = result
                    .rows
                    .iter()
                    .filter_map(|row| match row.get(col_idx).unwrap_or(&QueryValue::Null) {
                        QueryValue::Integer(v) => Some(*v),
                        _ => None,
                    })
                    .collect();
                w.write_batch(&values, Some(&def_levels), None)?;
            }
            (ColKind::Double, ColumnWriter::DoubleColumnWriter(w)) => {
                let values: Vec<f64> = result
                    .rows
                    .iter()
                    .filter_map(|row| match row.get(col_idx).unwrap_or(&QueryValue::Null) {
                        QueryValue::Real(v) => Some(*v),
                        QueryValue::Integer(v) => Some(*v as f64),
                        _ => None,
                    })
                    .collect();
                w.write_batch(&values, Some(&def_levels), None)?;
            }
            (ColKind::Bytes, ColumnWriter::ByteArrayColumnWriter(w)) => {
                let values: Vec<ByteArray> = result
                    .rows
                    .iter()
                    .filter_map(|row| match row.get(col_idx).unwrap_or(&QueryValue::Null) {
                        QueryValue::Text(s) => Some(ByteArray::from(s.as_bytes().to_vec())),
                        QueryValue::Integer(v) => Some(ByteArray::from(v.to_string().into_bytes())),
                        QueryValue::Real(v) => Some(ByteArray::from(v.to_string().into_bytes())),
                        _ => None,
                    })
                    .collect();
                w.write_batch(&values, Some(&def_levels), None)?;
            }
            _ => unreachable!("column kind and writer type must agree"),
        }

        col_writer.close()?;
    }

    rg.close()?;
    file_writer.close()?;

    Ok(())
}

#[derive(Debug)]
struct SheetData {
    original_name: String,
    columns: Vec<String>,
    rows: Vec<Vec<QueryValue>>,
}

fn load_input(
    input_path: &Path,
    requested_sheet: Option<&str>,
    inference_options: &TypeInferenceOptions,
    extraction_options: &ExtractionOptions,
    has_headers: bool,
) -> Result<SheetData> {
    if input_path
        .extension()
        .map(|extension| extension.to_string_lossy())
        .is_some_and(|extension| extension.eq_ignore_ascii_case("parquet"))
    {
        return load_parquet_sheet(input_path, requested_sheet, inference_options);
    }

    if input_path
        .extension()
        .map(|extension| extension.to_string_lossy())
        .is_some_and(|extension| extension.eq_ignore_ascii_case("xml"))
    {
        return load_xml_sheet(input_path, requested_sheet, inference_options, extraction_options);
    }

    if input_path
        .extension()
        .map(|extension| extension.to_string_lossy())
        .is_some_and(|extension| extension.eq_ignore_ascii_case("csv"))
    {
        return load_csv_sheet(input_path, requested_sheet, inference_options, has_headers);
    }

    if input_path
        .extension()
        .map(|extension| extension.to_string_lossy())
        .is_some_and(|extension| extension.eq_ignore_ascii_case("jsonl"))
    {
        return load_jsonl_sheet(input_path, requested_sheet, inference_options);
    }

    if input_path
        .extension()
        .map(|extension| extension.to_string_lossy())
        .is_some_and(|extension| extension.eq_ignore_ascii_case("json"))
    {
        return load_json_sheet(input_path, requested_sheet, inference_options, extraction_options);
    }

    if input_path
        .extension()
        .map(|extension| extension.to_string_lossy())
        .is_some_and(|extension| {
            extension.eq_ignore_ascii_case("md") || extension.eq_ignore_ascii_case("markdown")
        })
    {
        return load_markdown_sheet(input_path, requested_sheet, inference_options, has_headers);
    }

    load_xlsx_sheet(input_path, requested_sheet, inference_options, has_headers)
}

fn load_parquet_sheet(
    parquet_path: &Path,
    requested_sheet: Option<&str>,
    inference_options: &TypeInferenceOptions,
) -> Result<SheetData> {
    if let Some(selector) = requested_sheet {
        bail!(
            "Parquet input {} does not support selector '{selector}'. Remove ':{selector}' from --input for Parquet files.",
            parquet_path.display()
        );
    }

    let reader = SerializedFileReader::try_from(parquet_path)
        .with_context(|| format!("failed to open {}", parquet_path.display()))?;
    let mut row_iter = reader
        .get_row_iter(None)
        .with_context(|| format!("failed to read rows from {}", parquet_path.display()))?;

    let first_row = row_iter
        .next()
        .transpose()
        .with_context(|| format!("failed to read first row from {}", parquet_path.display()))?;

    let Some(first_row) = first_row else {
        bail!("Parquet input {} is empty", parquet_path.display());
    };

    let columns = normalize_text_headers(
        &first_row
            .get_column_iter()
            .map(|(name, _)| name.to_owned())
            .collect::<Vec<_>>(),
    );

    let mut rows = vec![parquet_row_to_values(first_row, inference_options)];
    for row in row_iter {
        let row =
            row.with_context(|| format!("failed to read row from {}", parquet_path.display()))?;
        rows.push(parquet_row_to_values(row, inference_options));
    }

    Ok(SheetData {
        original_name: "parquet".to_owned(),
        columns,
        rows,
    })
}

fn load_csv_sheet(
    csv_path: &Path,
    requested_sheet: Option<&str>,
    inference_options: &TypeInferenceOptions,
    has_headers: bool,
) -> Result<SheetData> {
    if let Some(selector) = requested_sheet {
        bail!(
            "CSV input {} does not support selector '{selector}'. Remove ':{selector}' from --input for CSV files.",
            csv_path.display()
        );
    }

    let mut reader = csv::ReaderBuilder::new()
        .has_headers(has_headers)
        .from_path(csv_path)
        .with_context(|| format!("failed to open {}", csv_path.display()))?;

    let mut records = Vec::new();
    for record in reader.records() {
        let record = record
            .with_context(|| format!("failed to read CSV record from {}", csv_path.display()))?;
        records.push(record.iter().map(str::to_owned).collect::<Vec<_>>());
    }

    let columns = if has_headers {
        let headers = reader
            .headers()
            .with_context(|| format!("failed to read headers from {}", csv_path.display()))?
            .iter()
            .map(str::to_owned)
            .collect::<Vec<_>>();
        normalize_text_headers(&headers)
    } else {
        let width = records.iter().map(Vec::len).max().unwrap_or(0);
        (1..=width).map(|index| format!("column{index}")).collect()
    };

    if columns.is_empty() {
        bail!("CSV input {} is empty", csv_path.display());
    }

    let rows = records
        .into_iter()
        .map(|record| {
            (0..columns.len())
                .map(|index| {
                    let value = record.get(index).map(String::as_str).unwrap_or("");
                    parse_scalar_value(value, inference_options)
                })
                .collect::<Vec<_>>()
        })
        .filter(|row| row.iter().any(|value| !matches!(value, QueryValue::Null)))
        .collect::<Vec<_>>();

    Ok(SheetData {
        original_name: "csv".to_owned(),
        columns,
        rows,
    })
}

fn load_jsonl_sheet(
    jsonl_path: &Path,
    requested_sheet: Option<&str>,
    inference_options: &TypeInferenceOptions,
) -> Result<SheetData> {
    if let Some(selector) = requested_sheet {
        bail!(
            "JSONL input {} does not support selector '{selector}'. Remove ':{selector}' from --input for JSONL files.",
            jsonl_path.display()
        );
    }

    let content = fs::read_to_string(jsonl_path)
        .with_context(|| format!("failed to read {}", jsonl_path.display()))?;

    let mut rows_maps = Vec::<Vec<(String, QueryValue)>>::new();
    for (line_index, raw_line) in content.lines().enumerate() {
        let line = raw_line.trim();
        if line.is_empty() {
            continue;
        }

        let json_value = serde_json::from_str::<JsonValue>(line).with_context(|| {
            format!(
                "failed to parse JSONL line {} in {}",
                line_index + 1,
                jsonl_path.display()
            )
        })?;

        let JsonValue::Object(object) = json_value else {
            bail!(
                "JSONL line {} in {} is not an object",
                line_index + 1,
                jsonl_path.display()
            );
        };

        rows_maps.push(
            object
                .into_iter()
                .map(|(key, value)| (key, json_to_query_value(value, inference_options)))
                .collect(),
        );
    }

    if rows_maps.is_empty() {
        bail!("JSONL input {} is empty", jsonl_path.display());
    }

    Ok(rows_maps_to_sheet_data(rows_maps, "jsonl".to_owned()))
}

fn load_json_sheet(
    json_path: &Path,
    requested_sheet: Option<&str>,
    inference_options: &TypeInferenceOptions,
    extraction_options: &ExtractionOptions,
) -> Result<SheetData> {
    let content = fs::read_to_string(json_path)
        .with_context(|| format!("failed to read {}", json_path.display()))?;
    let root = serde_json::from_str::<JsonValue>(&content)
        .with_context(|| format!("failed to parse JSON {}", json_path.display()))?;

    let scope = if let Some(sheet_key) = requested_sheet {
        let JsonValue::Object(mut object) = root else {
            bail!(
                "JSON input {} uses selector '{sheet_key}', but key selection requires a top-level object. Remove the selector or provide a JSON object at the root.",
                json_path.display()
            );
        };

        let mut available_keys = object.keys().cloned().collect::<Vec<_>>();
        available_keys.sort();

        object.remove(sheet_key).ok_or_else(|| {
            let available_suffix = if available_keys.is_empty() {
                " The top-level object has no keys.".to_owned()
            } else {
                format!(" Available keys: {}.", available_keys.join(", "))
            };
            anyhow!(
                "JSON key '{sheet_key}' not found in {}.{available_suffix}",
                json_path.display()
            )
        })?
    } else {
        root
    };

    let rows_maps = match extraction_options.json_mode {
        JsonMode::Array => json_scope_to_rows(scope, inference_options),
        JsonMode::Object => json_scope_to_rows_object(scope, inference_options),
        JsonMode::Flatten => json_scope_to_rows_flatten(scope, inference_options),
    };
    if rows_maps.is_empty() {
        if let Some(sheet_key) = requested_sheet {
            bail!(
                "JSON selector '{sheet_key}' in {} resolved to an empty table (no rows)",
                json_path.display()
            );
        }
        bail!("JSON input {} is empty", json_path.display());
    }

    Ok(rows_maps_to_sheet_data(
        rows_maps,
        requested_sheet.unwrap_or("json").to_owned(),
    ))
}

fn load_markdown_sheet(
    markdown_path: &Path,
    requested_sheet: Option<&str>,
    inference_options: &TypeInferenceOptions,
    has_headers: bool,
) -> Result<SheetData> {
    let content = fs::read_to_string(markdown_path)
        .with_context(|| format!("failed to read {}", markdown_path.display()))?;
    let tables = parse_markdown_tables(&content);
    if tables.is_empty() {
        bail!(
            "Markdown input {} does not contain any table",
            markdown_path.display()
        );
    }

    let table_index = if let Some(key) = requested_sheet {
        let index = key
            .trim()
            .parse::<usize>()
            .map_err(|_| {
                anyhow!(
                    "invalid Markdown table selector '{key}' for {}. Use a 1-based numeric index such as ':1' or ':2'.",
                    markdown_path.display()
                )
            })?;
        if index == 0 {
            bail!(
                "invalid Markdown table selector '{key}' for {}. Table indexes are 1-based (:1, :2, ...).",
                markdown_path.display()
            );
        }
        index
    } else {
        1
    };

    let table_count = tables.len();
    let Some(table) = tables.into_iter().nth(table_index - 1) else {
        bail!(
            "Markdown table {} not found in {}. Found {} table(s); choose an index between 1 and {}.",
            table_index,
            markdown_path.display(),
            table_count,
            table_count
        );
    };

    let mut rows = table.rows;
    let columns = if has_headers {
        normalize_text_headers(&table.headers)
    } else {
        rows.insert(0, table.headers);
        let width = rows.iter().map(Vec::len).max().unwrap_or(0);
        (1..=width).map(|index| format!("column{index}")).collect()
    };

    if columns.is_empty() {
        bail!("Markdown table {} is empty", table_index);
    }

    let data_rows = rows
        .into_iter()
        .map(|row| {
            (0..columns.len())
                .map(|index| {
                    let value = row.get(index).map(String::as_str).unwrap_or("");
                    parse_scalar_value(value, inference_options)
                })
                .collect::<Vec<_>>()
        })
        .filter(|row| row.iter().any(|value| !matches!(value, QueryValue::Null)))
        .collect::<Vec<_>>();

    if data_rows.is_empty() {
        bail!(
            "Markdown table {} in {} is empty (no data rows)",
            table_index,
            markdown_path.display()
        );
    }

    Ok(SheetData {
        original_name: format!("table{table_index}"),
        columns,
        rows: data_rows,
    })
}

fn load_xlsx_sheet(
    workbook_path: &Path,
    requested_sheet: Option<&str>,
    inference_options: &TypeInferenceOptions,
    has_headers: bool,
) -> Result<SheetData> {
    let mut workbook = open_workbook_auto(workbook_path)
        .with_context(|| format!("failed to open {}", workbook_path.display()))?;

    let sheet_name = match requested_sheet {
        Some(name) => name.to_owned(),
        None => workbook
            .sheet_names()
            .first()
            .cloned()
            .ok_or_else(|| anyhow!("workbook does not contain any sheets"))?,
    };

    let range = workbook
        .worksheet_range(&sheet_name)
        .with_context(|| format!("failed to read sheet {sheet_name}"))?;

    if range.width() == 0 {
        bail!("sheet {sheet_name} is empty");
    }

    let mut rows = range.rows();
    let columns = if has_headers {
        let header_row = rows
            .next()
            .ok_or_else(|| anyhow!("sheet {sheet_name} is empty"))?;
        normalize_headers(header_row, range.width())
    } else {
        (1..=range.width())
            .map(|index| format!("column{index}"))
            .collect()
    };

    let data_rows = rows
        .map(|row| {
            (0..columns.len())
                .map(|index| convert_cell(row.get(index).unwrap_or(&Data::Empty)))
                .map(|value| apply_inference_overrides(value, inference_options))
                .collect::<Vec<_>>()
        })
        .filter(|row| row.iter().any(|value| !matches!(value, QueryValue::Null)))
        .collect();

    Ok(SheetData {
        original_name: sheet_name,
        columns,
        rows: data_rows,
    })
}

fn load_xml_sheet(
    xml_path: &Path,
    requested_sheet: Option<&str>,
    inference_options: &TypeInferenceOptions,
    extraction_options: &ExtractionOptions,
) -> Result<SheetData> {
    let xml_content = fs::read_to_string(xml_path)
        .with_context(|| format!("failed to read {}", xml_path.display()))?;
    let document = Document::parse(&xml_content)
        .with_context(|| format!("failed to parse XML {}", xml_path.display()))?;

    let root = document.root_element();
    let scope = if let Some(sheet_tag) = requested_sheet {
        root.descendants()
            .find(|node| node.is_element() && node.tag_name().name() == sheet_tag)
            .ok_or_else(|| anyhow!("XML tag '{sheet_tag}' not found"))?
    } else {
        root
    };

    let records = match extraction_options.xml_mode {
        XmlMode::Rows => {
            let mut records = collect_xml_records(scope, inference_options);
            if records.is_empty() {
                let fallback = xml_row_from_children(scope, inference_options);
                if fallback.is_empty() {
                    let scope_name = requested_sheet.unwrap_or(scope.tag_name().name());
                    bail!("XML scope '{scope_name}' does not contain tabular data");
                }
                records.push(fallback);
            }
            records
        }
        XmlMode::Descendants => {
            let records = collect_xml_descendants(scope, inference_options);
            if records.is_empty() {
                let scope_name = requested_sheet.unwrap_or(scope.tag_name().name());
                bail!(
                    "XML scope '{scope_name}' does not contain any leaf text elements (--xml-mode descendants)"
                );
            }
            records
        }
        XmlMode::Attributes => {
            let records = collect_xml_attributes_records(scope, inference_options);
            if records.is_empty() {
                let scope_name = requested_sheet.unwrap_or(scope.tag_name().name());
                bail!(
                    "XML scope '{scope_name}' does not contain any elements with attributes (--xml-mode attributes)"
                );
            }
            records
        }
    };

    let mut columns = Vec::new();
    for row in &records {
        for (column, _) in row {
            if !columns.iter().any(|existing| existing == column) {
                columns.push(column.clone());
            }
        }
    }

    let rows = records
        .into_iter()
        .map(|row| {
            columns
                .iter()
                .map(|column| {
                    row.iter()
                        .find(|(key, _)| key == column)
                        .map(|(_, value)| value.clone())
                        .unwrap_or(QueryValue::Null)
                })
                .collect::<Vec<_>>()
        })
        .collect::<Vec<_>>();

    let original_name = requested_sheet
        .map(str::to_owned)
        .unwrap_or_else(|| scope.tag_name().name().to_owned());

    Ok(SheetData {
        original_name,
        columns,
        rows,
    })
}

fn collect_xml_records(
    scope: Node<'_, '_>,
    inference_options: &TypeInferenceOptions,
) -> Vec<Vec<(String, QueryValue)>> {
    let direct_row_nodes = scope
        .children()
        .filter(|node| node.is_element() && node.tag_name().name().eq_ignore_ascii_case("row"))
        .collect::<Vec<_>>();
    if !direct_row_nodes.is_empty() {
        return direct_row_nodes
            .into_iter()
            .map(|row| xml_row_from_children(row, inference_options))
            .filter(|row| !row.is_empty())
            .collect();
    }

    scope
        .descendants()
        .filter(|node| node.is_element() && *node != scope)
        .filter(|node| is_xml_record_candidate(*node))
        .map(|row| xml_row_from_children(row, inference_options))
        .filter(|row| !row.is_empty())
        .collect()
}

fn is_xml_record_candidate(node: Node<'_, '_>) -> bool {
    let children = node
        .children()
        .filter(|child| child.is_element())
        .collect::<Vec<_>>();

    !children.is_empty()
        && children
            .iter()
            .all(|child| !child.children().any(|inner| inner.is_element()))
}

fn xml_row_from_children(
    node: Node<'_, '_>,
    inference_options: &TypeInferenceOptions,
) -> Vec<(String, QueryValue)> {
    node.children()
        .filter(|child| child.is_element())
        .map(|child| {
            let column = child.tag_name().name().to_owned();
            let value = xml_text_to_query_value(&xml_text_content(child), inference_options);
            (column, value)
        })
        .collect()
}

fn xml_text_content(node: Node<'_, '_>) -> String {
    node.text().unwrap_or_default().trim().to_owned()
}

fn xml_text_to_query_value(raw: &str, inference_options: &TypeInferenceOptions) -> QueryValue {
    parse_scalar_value(raw, inference_options)
}

/// `descendants` mode: collect every leaf text element (element with no child elements)
/// as a separate row.  Each row has two columns: `tag` (element name) and `value`
/// (text content).
fn collect_xml_descendants(
    scope: Node<'_, '_>,
    inference_options: &TypeInferenceOptions,
) -> Vec<Vec<(String, QueryValue)>> {
    scope
        .descendants()
        .filter(|node| node.is_element() && *node != scope)
        .filter(|node| !node.children().any(|child| child.is_element()))
        .map(|node| {
            let tag = node.tag_name().name().to_owned();
            let value = xml_text_to_query_value(&xml_text_content(node), inference_options);
            vec![
                ("tag".to_owned(), QueryValue::Text(tag)),
                ("value".to_owned(), value),
            ]
        })
        .collect()
}

/// `attributes` mode: every element that carries at least one XML attribute becomes a
/// row.  Attribute names map to column names; text content is included as a `value`
/// column when non-empty.
fn collect_xml_attributes_records(
    scope: Node<'_, '_>,
    inference_options: &TypeInferenceOptions,
) -> Vec<Vec<(String, QueryValue)>> {
    scope
        .descendants()
        .filter(|node| node.is_element() && *node != scope)
        .filter(|node| node.attributes().count() > 0)
        .map(|node| {
            let mut row: Vec<(String, QueryValue)> = node
                .attributes()
                .map(|attr| {
                    let column = attr.name().to_owned();
                    let value = xml_text_to_query_value(attr.value(), inference_options);
                    (column, value)
                })
                .collect();
            let text = xml_text_content(node);
            if !text.is_empty() {
                row.push((
                    "value".to_owned(),
                    xml_text_to_query_value(&text, inference_options),
                ));
            }
            row
        })
        .filter(|row| !row.is_empty())
        .collect()
}

fn parse_scalar_value(raw: &str, inference_options: &TypeInferenceOptions) -> QueryValue {
    let trimmed = raw.trim();
    if matches_token(trimmed, &inference_options.null_values) {
        return QueryValue::Null;
    }

    if !inference_options.infer_types {
        return QueryValue::Text(raw.to_owned());
    }

    if matches_token(trimmed, &inference_options.true_values) {
        return QueryValue::Integer(1);
    }

    if matches_token(trimmed, &inference_options.false_values) {
        return QueryValue::Integer(0);
    }

    if let Ok(value) = trimmed.parse::<i64>() {
        return QueryValue::Integer(value);
    }

    if let Some(value) = parse_decimal_value(trimmed, inference_options.decimal_comma) {
        return QueryValue::Real(value);
    }

    if let Some(value) = parse_date_value(trimmed, inference_options.date_format.as_deref()) {
        return QueryValue::Text(value);
    }

    QueryValue::Text(raw.to_owned())
}

fn matches_token(raw: &str, candidates: &[String]) -> bool {
    candidates
        .iter()
        .any(|candidate| raw.eq_ignore_ascii_case(candidate.trim()))
}

fn parse_decimal_value(raw: &str, decimal_comma: bool) -> Option<f64> {
    if let Ok(value) = raw.parse::<f64>() {
        return Some(value);
    }

    if !decimal_comma {
        return None;
    }

    let compact = raw.replace(' ', "");
    if !compact.contains(',') {
        return None;
    }

    let normalized = if compact.contains('.') {
        compact.replace('.', "").replace(',', ".")
    } else {
        compact.replace(',', ".")
    };

    normalized.parse::<f64>().ok()
}

fn parse_date_value(raw: &str, date_format: Option<&str>) -> Option<String> {
    let format = date_format?;

    if let Ok(value) = NaiveDate::parse_from_str(raw, format) {
        return Some(value.format("%Y-%m-%d").to_string());
    }

    if let Ok(value) = NaiveDateTime::parse_from_str(raw, format) {
        return Some(value.format("%Y-%m-%d %H:%M:%S").to_string());
    }

    None
}

fn apply_inference_overrides(
    value: QueryValue,
    inference_options: &TypeInferenceOptions,
) -> QueryValue {
    if inference_options.infer_types {
        return value;
    }

    match value {
        QueryValue::Null => QueryValue::Null,
        QueryValue::Integer(number) => QueryValue::Text(number.to_string()),
        QueryValue::Real(number) => QueryValue::Text(number.to_string()),
        QueryValue::Text(text) => QueryValue::Text(text),
    }
}

fn parquet_row_to_values(
    row: parquet::record::Row,
    inference_options: &TypeInferenceOptions,
) -> Vec<QueryValue> {
    row.into_columns()
        .into_iter()
        .map(|(_, field)| parquet_field_to_query_value(field, inference_options))
        .collect()
}

fn parquet_field_to_query_value(
    field: Field,
    inference_options: &TypeInferenceOptions,
) -> QueryValue {
    apply_inference_overrides(
        match field {
            Field::Null => QueryValue::Null,
            Field::Bool(value) => QueryValue::Integer(i64::from(value)),
            Field::Byte(value) => QueryValue::Integer(i64::from(value)),
            Field::Short(value) => QueryValue::Integer(i64::from(value)),
            Field::Int(value) => QueryValue::Integer(i64::from(value)),
            Field::Long(value) => QueryValue::Integer(value),
            Field::UByte(value) => QueryValue::Integer(i64::from(value)),
            Field::UShort(value) => QueryValue::Integer(i64::from(value)),
            Field::UInt(value) => QueryValue::Integer(i64::from(value)),
            Field::ULong(value) => i64::try_from(value)
                .map(QueryValue::Integer)
                .unwrap_or_else(|_| QueryValue::Text(value.to_string())),
            Field::Float16(value) => QueryValue::Real(f64::from(value)),
            Field::Float(value) => QueryValue::Real(f64::from(value)),
            Field::Double(value) => QueryValue::Real(value),
            Field::Decimal(value) => QueryValue::Text(format!("{value:?}")),
            Field::Str(value) => QueryValue::Text(value),
            Field::Bytes(value) => QueryValue::Text(String::from_utf8_lossy(value.data()).into()),
            Field::Date(value) => QueryValue::Integer(i64::from(value)),
            Field::TimestampMillis(value) => QueryValue::Integer(value),
            Field::TimestampMicros(value) => QueryValue::Integer(value),
            Field::Group(value) => QueryValue::Text(value.to_string()),
            Field::ListInternal(value) => QueryValue::Text(format!("{value:?}")),
            Field::MapInternal(value) => QueryValue::Text(format!("{value:?}")),
        },
        inference_options,
    )
}

fn apply_input_normalization(sheet: &mut SheetData, options: &InputNormalizationOptions) {
    if options.trim {
        for column in &mut sheet.columns {
            *column = column.trim().to_owned();
        }

        for row in &mut sheet.rows {
            for value in row {
                if let QueryValue::Text(text) = value {
                    let trimmed = text.trim();
                    if trimmed.is_empty() {
                        *value = QueryValue::Null;
                    } else {
                        *text = trimmed.to_owned();
                    }
                }
            }
        }
    }

    if options.normalize_headers {
        sheet.columns = sheet
            .columns
            .iter()
            .enumerate()
            .map(|(index, header)| {
                normalize_header_value(header, index + 1, options.header_case.as_ref())
            })
            .collect();
    }

    if options.dedupe_headers {
        dedupe_headers(&mut sheet.columns);
    }

    if options.skip_empty_rows {
        sheet.rows.retain(|row| !is_row_empty(row));
    }
}

fn normalize_header_value(header: &str, index: usize, header_case: Option<&HeaderCase>) -> String {
    let normalized = header
        .trim()
        .chars()
        .map(|character| {
            if character.is_alphanumeric() || character == '_' {
                character
            } else {
                '_'
            }
        })
        .collect::<String>();
    let collapsed = collapse_underscores(&normalized)
        .trim_matches('_')
        .to_owned();

    let mut value = if collapsed.is_empty() {
        format!("column{index}")
    } else {
        collapsed
    };

    if matches!(header_case, Some(HeaderCase::Snake)) {
        value = value.to_ascii_lowercase();
    }

    value
}

fn collapse_underscores(value: &str) -> String {
    let mut output = String::with_capacity(value.len());
    let mut previous_was_underscore = false;

    for character in value.chars() {
        if character == '_' {
            if !previous_was_underscore {
                output.push(character);
            }
            previous_was_underscore = true;
        } else {
            output.push(character);
            previous_was_underscore = false;
        }
    }

    output
}

fn dedupe_headers(headers: &mut [String]) {
    let mut seen = std::collections::HashMap::new();
    let mut assigned = HashSet::new();

    for header in headers {
        let base = header.clone();
        let count = seen
            .entry(base.clone())
            .and_modify(|count| *count += 1)
            .or_insert(1usize);

        let mut candidate = if *count == 1 {
            base.clone()
        } else {
            format!("{base}_{count}")
        };

        while assigned.contains(&candidate) {
            *count += 1;
            candidate = format!("{base}_{count}");
        }

        assigned.insert(candidate.clone());
        *header = candidate;
    }
}

fn is_row_empty(row: &[QueryValue]) -> bool {
    row.iter().all(|value| match value {
        QueryValue::Null => true,
        QueryValue::Text(text) => text.trim().is_empty(),
        QueryValue::Integer(_) | QueryValue::Real(_) => false,
    })
}

fn normalize_text_headers(headers: &[String]) -> Vec<String> {
    let mut seen = std::collections::HashMap::new();

    headers
        .iter()
        .enumerate()
        .map(|(index, header)| {
            let base = if header.trim().is_empty() {
                format!("column{}", index + 1)
            } else {
                header.clone()
            };

            let count = seen
                .entry(base.clone())
                .and_modify(|count| *count += 1)
                .or_insert(1usize);

            if *count == 1 {
                base
            } else {
                format!("{base}_{count}")
            }
        })
        .collect()
}

fn rows_maps_to_sheet_data(
    rows_maps: Vec<Vec<(String, QueryValue)>>,
    original_name: String,
) -> SheetData {
    let mut columns = Vec::<String>::new();
    for row in &rows_maps {
        for (column, _) in row {
            if !columns.iter().any(|existing| existing == column) {
                columns.push(column.clone());
            }
        }
    }

    let rows = rows_maps
        .into_iter()
        .map(|row| {
            columns
                .iter()
                .map(|column| {
                    row.iter()
                        .find(|(key, _)| key == column)
                        .map(|(_, value)| value.clone())
                        .unwrap_or(QueryValue::Null)
                })
                .collect::<Vec<_>>()
        })
        .collect::<Vec<_>>();

    SheetData {
        original_name,
        columns,
        rows,
    }
}

#[derive(Debug)]
struct MarkdownTable {
    headers: Vec<String>,
    rows: Vec<Vec<String>>,
}

fn parse_markdown_tables(content: &str) -> Vec<MarkdownTable> {
    let lines = content.lines().collect::<Vec<_>>();
    let mut tables = Vec::new();
    let mut index = 0usize;

    while index + 1 < lines.len() {
        if !looks_like_markdown_row(lines[index]) || !is_markdown_separator(lines[index + 1]) {
            index += 1;
            continue;
        }

        let headers = parse_markdown_row(lines[index]);
        if headers.is_empty() {
            index += 1;
            continue;
        }

        let mut rows = Vec::new();
        index += 2;
        while index < lines.len() && looks_like_markdown_row(lines[index]) {
            if is_markdown_separator(lines[index]) {
                break;
            }
            let row = parse_markdown_row(lines[index]);
            if row.iter().any(|value| !value.trim().is_empty()) {
                rows.push(row);
            }
            index += 1;
        }

        tables.push(MarkdownTable { headers, rows });
    }

    tables
}

fn looks_like_markdown_row(line: &str) -> bool {
    line.contains('|')
}

fn is_markdown_separator(line: &str) -> bool {
    let parts = split_markdown_cells(line);
    if parts.is_empty() {
        return false;
    }

    parts.iter().all(|part| {
        let cell = part.trim();
        if cell.is_empty() {
            return false;
        }

        let inner = cell.trim_matches(':');
        !inner.is_empty() && inner.chars().all(|character| character == '-')
    })
}

fn parse_markdown_row(line: &str) -> Vec<String> {
    split_markdown_cells(line)
        .into_iter()
        .map(|cell| cell.replace("\\|", "|").trim().to_owned())
        .collect()
}

fn split_markdown_cells(line: &str) -> Vec<String> {
    let trimmed = line.trim();
    if trimmed.is_empty() {
        return Vec::new();
    }

    let without_prefix = trimmed.strip_prefix('|').unwrap_or(trimmed);
    let content = without_prefix.strip_suffix('|').unwrap_or(without_prefix);

    content.split('|').map(str::to_owned).collect()
}

fn json_scope_to_rows(
    scope: JsonValue,
    inference_options: &TypeInferenceOptions,
) -> Vec<Vec<(String, QueryValue)>> {
    match scope {
        JsonValue::Array(items) => items
            .into_iter()
            .map(|item| json_item_to_row(item, inference_options))
            .collect::<Vec<_>>(),
        JsonValue::Object(object) => vec![
            object
                .into_iter()
                .map(|(key, value)| (key, json_to_query_value(value, inference_options)))
                .collect(),
        ],
        scalar => vec![vec![(
            "value".to_owned(),
            json_to_query_value(scalar, inference_options),
        )]],
    }
}

fn json_item_to_row(
    item: JsonValue,
    inference_options: &TypeInferenceOptions,
) -> Vec<(String, QueryValue)> {
    match item {
        JsonValue::Object(object) => object
            .into_iter()
            .map(|(key, value)| (key, json_to_query_value(value, inference_options)))
            .collect(),
        scalar => vec![(
            "value".to_owned(),
            json_to_query_value(scalar, inference_options),
        )],
    }
}

fn json_to_query_value(value: JsonValue, inference_options: &TypeInferenceOptions) -> QueryValue {
    match value {
        JsonValue::Null => QueryValue::Null,
        JsonValue::Bool(flag) => {
            apply_inference_overrides(QueryValue::Integer(i64::from(flag)), inference_options)
        }
        JsonValue::Number(number) => {
            if let Some(integer) = number.as_i64() {
                apply_inference_overrides(QueryValue::Integer(integer), inference_options)
            } else if let Some(real) = number.as_f64() {
                apply_inference_overrides(QueryValue::Real(real), inference_options)
            } else {
                QueryValue::Text(number.to_string())
            }
        }
        JsonValue::String(text) => parse_scalar_value(&text, inference_options),
        JsonValue::Array(_) | JsonValue::Object(_) => QueryValue::Text(value.to_string()),
    }
}

/// `object` mode: each key-value pair of a JSON object becomes a separate row with
/// columns `key` and `value`.  When the scope is an array each element is processed
/// independently.  Non-object, non-array scalars produce a single row.
fn json_scope_to_rows_object(
    scope: JsonValue,
    inference_options: &TypeInferenceOptions,
) -> Vec<Vec<(String, QueryValue)>> {
    match scope {
        JsonValue::Object(object) => object
            .into_iter()
            .map(|(key, value)| {
                vec![
                    ("key".to_owned(), QueryValue::Text(key)),
                    (
                        "value".to_owned(),
                        json_to_query_value(value, inference_options),
                    ),
                ]
            })
            .collect(),
        JsonValue::Array(items) => items
            .into_iter()
            .flat_map(|item| json_scope_to_rows_object(item, inference_options))
            .collect(),
        scalar => vec![vec![
            ("key".to_owned(), QueryValue::Text("value".to_owned())),
            (
                "value".to_owned(),
                json_to_query_value(scalar, inference_options),
            ),
        ]],
    }
}

/// `flatten` mode: recursively expand nested JSON objects and arrays into flat key-value
/// pairs using dotted paths (`address.city`, `tags.0`, …).  Arrays produce one row per
/// top-level element; a non-array root produces a single row.
fn json_scope_to_rows_flatten(
    scope: JsonValue,
    inference_options: &TypeInferenceOptions,
) -> Vec<Vec<(String, QueryValue)>> {
    match scope {
        JsonValue::Array(items) => items
            .into_iter()
            .map(|item| {
                let mut row = Vec::new();
                flatten_json_value("", item, &mut row, inference_options);
                row
            })
            .filter(|row| !row.is_empty())
            .collect(),
        _ => {
            let mut row = Vec::new();
            flatten_json_value("", scope, &mut row, inference_options);
            if row.is_empty() { vec![] } else { vec![row] }
        }
    }
}

/// Recursively walks `value`, emitting one `(key, scalar)` pair per leaf into
/// `result`.  Nested object keys are joined with `.`; array indices are appended
/// with `.N`.
fn flatten_json_value(
    prefix: &str,
    value: JsonValue,
    result: &mut Vec<(String, QueryValue)>,
    inference_options: &TypeInferenceOptions,
) {
    match value {
        JsonValue::Object(obj) => {
            for (k, v) in obj {
                let key = if prefix.is_empty() {
                    k
                } else {
                    format!("{prefix}.{k}")
                };
                flatten_json_value(&key, v, result, inference_options);
            }
        }
        JsonValue::Array(arr) => {
            for (i, v) in arr.into_iter().enumerate() {
                let key = if prefix.is_empty() {
                    i.to_string()
                } else {
                    format!("{prefix}.{i}")
                };
                flatten_json_value(&key, v, result, inference_options);
            }
        }
        _ => {
            let key = if prefix.is_empty() {
                "value".to_owned()
            } else {
                prefix.to_owned()
            };
            result.push((key, json_to_query_value(value, inference_options)));
        }
    }
}

fn normalize_headers(header_row: &[Data], width: usize) -> Vec<String> {
    let mut seen = std::collections::HashMap::new();

    (0..width)
        .map(|index| {
            let base = header_row
                .get(index)
                .map(cell_to_string)
                .filter(|value| !value.trim().is_empty())
                .unwrap_or_else(|| format!("column{}", index + 1));

            let count = seen
                .entry(base.clone())
                .and_modify(|count| *count += 1)
                .or_insert(1usize);

            if *count == 1 {
                base
            } else {
                format!("{base}_{count}")
            }
        })
        .collect()
}

fn convert_cell(cell: &Data) -> QueryValue {
    match cell {
        Data::Empty => QueryValue::Null,
        Data::Int(value) => QueryValue::Integer(*value),
        Data::Float(value) => QueryValue::Real(*value),
        Data::String(value) => QueryValue::Text(value.clone()),
        Data::Bool(value) => QueryValue::Integer(i64::from(*value)),
        Data::DateTime(value) => QueryValue::Text(value.to_string()),
        Data::DateTimeIso(value) => QueryValue::Text(value.clone()),
        Data::DurationIso(value) => QueryValue::Text(value.clone()),
        Data::Error(value) => QueryValue::Text(value.to_string()),
    }
}

fn cell_to_string(cell: &Data) -> String {
    match convert_cell(cell) {
        QueryValue::Null => String::new(),
        QueryValue::Integer(value) => value.to_string(),
        QueryValue::Real(value) => value.to_string(),
        QueryValue::Text(value) => value,
    }
}

fn register_sheet(
    connection: &Connection,
    sheet: &SheetData,
    table_name: &str,
    registered_views: &mut HashSet<String>,
) -> Result<()> {
    let columns = sheet
        .columns
        .iter()
        .map(|column| quote_identifier(column))
        .collect::<Vec<_>>()
        .join(", ");

    connection
        .execute(
            &format!("CREATE TABLE {} ({columns})", quote_identifier(table_name)),
            [],
        )
        .context("failed to create sheet table")?;

    let table1_key = normalize_view_key("table1");
    if table_name == "table" && !registered_views.contains(&table1_key) {
        connection
            .execute(
                &format!(
                    "CREATE VIEW {} AS SELECT * FROM {}",
                    quote_identifier("table1"),
                    quote_identifier("table")
                ),
                [],
            )
            .context("failed to register alias view table1 for first input")?;
        registered_views.insert(table1_key);
    }

    let sanitized_sheet_name = sanitize_table_name(&sheet.original_name);
    let sanitized_sheet_key = normalize_view_key(&sanitized_sheet_name);
    if sanitized_sheet_name != table_name && !registered_views.contains(&sanitized_sheet_key) {
        connection
            .execute(
                &format!(
                    "CREATE VIEW {} AS SELECT * FROM {}",
                    quote_identifier(&sanitized_sheet_name),
                    quote_identifier(table_name)
                ),
                [],
            )
            .with_context(|| {
                format!("failed to register view for sheet {}", sheet.original_name)
            })?;
        registered_views.insert(sanitized_sheet_key);
    }

    if sheet.rows.is_empty() {
        return Ok(());
    }

    let placeholders = vec!["?"; sheet.columns.len()].join(", ");
    let insert_sql = format!(
        "INSERT INTO {} VALUES ({placeholders})",
        quote_identifier(table_name)
    );
    let mut statement = connection
        .prepare(&insert_sql)
        .context("failed to prepare insert statement")?;

    for row in &sheet.rows {
        let values = row.iter().map(to_sql_value).collect::<Vec<_>>();
        statement
            .execute(params_from_iter(values))
            .context("failed to insert sheet row")?;
    }

    Ok(())
}

fn execute_query(
    connection: &Connection,
    query: &str,
    params: &[QueryParam],
) -> Result<QueryResult> {
    let normalized_query = normalize_query_for_reserved_identifiers(query);
    let mut statement = connection
        .prepare(&normalized_query)
        .map_err(|error| enrich_query_prepare_error(connection, query, error))?;

    if statement.column_count() == 0 {
        bail!("query must return rows");
    }

    let columns = statement
        .column_names()
        .into_iter()
        .map(str::to_owned)
        .collect::<Vec<_>>();

    bind_query_params(&mut statement, params)?;

    let column_count = statement.column_count();
    let mut rows = statement.raw_query();
    let mut result_rows = Vec::new();

    while let Some(row) = rows.next().context("failed to fetch query row")? {
        let mut values = Vec::with_capacity(column_count);

        for index in 0..column_count {
            values.push(match row.get_ref(index)? {
                ValueRef::Null => QueryValue::Null,
                ValueRef::Integer(value) => QueryValue::Integer(value),
                ValueRef::Real(value) => QueryValue::Real(value),
                ValueRef::Text(value) => {
                    QueryValue::Text(String::from_utf8_lossy(value).into_owned())
                }
                ValueRef::Blob(value) => QueryValue::Text(format_blob(value)),
            });
        }

        result_rows.push(values);
    }

    Ok(QueryResult {
        columns,
        rows: result_rows,
    })
}

fn enrich_query_prepare_error(
    connection: &Connection,
    query: &str,
    error: rusqlite::Error,
) -> anyhow::Error {
    let raw_error = error.to_string();

    if let Some(table) = raw_error.strip_prefix("no such table: ").map(str::trim) {
        let table = simplify_sqlite_missing_identifier(table);
        let available_tables = list_loaded_table_names(connection);
        let available_suffix = if available_tables.is_empty() {
            String::new()
        } else {
            format!(" Available tables/views: {}.", available_tables.join(", "))
        };

        return anyhow!(
            "query references unknown table '{table}'.{available_suffix} Check your table names or run `qf tables --input ...` to inspect available tables.\nQuery: {query}"
        );
    }

    if let Some(column) = raw_error.strip_prefix("no such column: ").map(str::trim) {
        let column = simplify_sqlite_missing_identifier(column);
        return anyhow!(
            "query references unknown column '{column}'. Check your column names or run `qf schema --input ...` to inspect available columns.\nQuery: {query}"
        );
    }

    anyhow!(error).context(format!("failed to prepare query: {query}"))
}

fn simplify_sqlite_missing_identifier(identifier: &str) -> &str {
    identifier
        .split_once(" in ")
        .map(|(name, _)| name)
        .unwrap_or(identifier)
        .trim()
}

fn list_loaded_table_names(connection: &Connection) -> Vec<String> {
    let mut statement = match connection.prepare(
        "SELECT name FROM sqlite_master WHERE type IN ('table', 'view') AND name NOT LIKE 'sqlite_%' ORDER BY name",
    ) {
        Ok(statement) => statement,
        Err(_) => return Vec::new(),
    };

    let names = match statement.query_map([], |row| row.get::<_, String>(0)) {
        Ok(names) => names,
        Err(_) => return Vec::new(),
    };

    names.filter_map(|name| name.ok()).collect()
}

fn normalize_query_for_reserved_identifiers(query: &str) -> String {
    static RESERVED_TABLE_RE: OnceLock<Regex> = OnceLock::new();

    let re = RESERVED_TABLE_RE.get_or_init(|| {
        Regex::new(r"(?i)\b(from|join|update|into)\s+table\b")
            .expect("reserved table regex should compile")
    });

    re.replace_all(query, |captures: &regex::Captures<'_>| {
        format!("{} \"table\"", &captures[1])
    })
    .into_owned()
}

fn bind_query_params(statement: &mut rusqlite::Statement<'_>, params: &[QueryParam]) -> Result<()> {
    for param in params {
        let mut bound = false;

        for prefix in [":", "@", "$"] {
            let parameter_name = format!("{prefix}{}", param.name);
            if let Some(index) = statement
                .parameter_index(&parameter_name)
                .with_context(|| format!("failed to inspect parameter {parameter_name}"))?
            {
                statement
                    .raw_bind_parameter(index, to_sql_value(&param.value))
                    .with_context(|| format!("failed to bind parameter {parameter_name}"))?;
                bound = true;
                break;
            }
        }

        if !bound {
            bail!("query does not contain parameter :{}", param.name);
        }
    }

    Ok(())
}

fn to_sql_value(value: &QueryValue) -> Value {
    match value {
        QueryValue::Null => Value::Null,
        QueryValue::Integer(value) => Value::Integer(*value),
        QueryValue::Real(value) => Value::Real(*value),
        QueryValue::Text(value) => Value::Text(value.clone()),
    }
}

fn display_value(value: &QueryValue) -> String {
    match value {
        QueryValue::Null => String::new(),
        QueryValue::Integer(value) => value.to_string(),
        QueryValue::Real(value) => value.to_string(),
        QueryValue::Text(value) => value.clone(),
    }
}

fn escape_csv_field(value: &str) -> String {
    if value.contains([',', '"', '\n', '\r']) {
        format!("\"{}\"", value.replace('"', "\"\""))
    } else {
        value.to_owned()
    }
}

fn escape_json_string(value: &str) -> String {
    let mut escaped = String::from("\"");
    for character in value.chars() {
        match character {
            '"' => escaped.push_str("\\\""),
            '\\' => escaped.push_str("\\\\"),
            '\n' => escaped.push_str("\\n"),
            '\r' => escaped.push_str("\\r"),
            '\t' => escaped.push_str("\\t"),
            '\u{08}' => escaped.push_str("\\b"),
            '\u{0C}' => escaped.push_str("\\f"),
            c if c <= '\u{1F}' => {
                let _ = write!(&mut escaped, "\\u{:04x}", c as u32);
            }
            c => escaped.push(c),
        }
    }
    escaped.push('"');
    escaped
}

fn to_json_value(value: &QueryValue) -> String {
    match value {
        QueryValue::Null => "null".to_owned(),
        QueryValue::Integer(number) => number.to_string(),
        QueryValue::Real(number) if number.is_finite() => number.to_string(),
        QueryValue::Real(_) => "null".to_owned(),
        QueryValue::Text(text) => escape_json_string(text),
    }
}

fn escape_markdown_cell(value: &str) -> String {
    value.replace('|', "\\|").replace('\n', "<br>")
}

fn escape_xml_text(value: &str) -> String {
    value
        .replace('&', "&amp;")
        .replace('<', "&lt;")
        .replace('>', "&gt;")
        .replace('"', "&quot;")
        .replace('\'', "&apos;")
}

fn escape_xml_tag(value: &str) -> String {
    let sanitized = value
        .chars()
        .map(|character| {
            if character.is_alphanumeric() || character == '_' || character == '-' {
                character
            } else {
                '_'
            }
        })
        .collect::<String>();

    if sanitized.is_empty() {
        "element".to_owned()
    } else if sanitized.chars().next().unwrap_or('x').is_numeric() {
        format!("_{sanitized}")
    } else {
        sanitized
    }
}

fn format_blob(value: &[u8]) -> String {
    let mut formatted = String::from("0x");
    for byte in value {
        let _ = write!(&mut formatted, "{byte:02x}");
    }

    formatted
}

fn quote_identifier(identifier: &str) -> String {
    format!("\"{}\"", identifier.replace('"', "\"\""))
}

fn sanitize_table_name(name: &str) -> String {
    let sanitized = name
        .chars()
        .map(|character| {
            if character.is_alphanumeric() || character == '_' {
                character
            } else {
                '_'
            }
        })
        .collect::<String>()
        .trim_matches('_')
        .to_string();

    if sanitized.is_empty() {
        "table_view".to_owned()
    } else {
        sanitized
    }
}

fn normalize_view_key(name: &str) -> String {
    name.to_ascii_lowercase()
}

#[cfg(test)]
mod tests {
    use std::{
        fs,
        path::PathBuf,
        time::{SystemTime, UNIX_EPOCH},
    };

    use super::*;

    fn temp_path(name: &str) -> PathBuf {
        let unique = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .expect("time should move forward")
            .as_nanos();
        std::env::temp_dir().join(format!("query-forge-{name}-{unique}.xlsx"))
    }

    fn temp_xml_path(name: &str) -> PathBuf {
        let unique = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .expect("time should move forward")
            .as_nanos();
        std::env::temp_dir().join(format!("query-forge-{name}-{unique}.xml"))
    }

    fn temp_csv_path(name: &str) -> PathBuf {
        let unique = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .expect("time should move forward")
            .as_nanos();
        std::env::temp_dir().join(format!("query-forge-{name}-{unique}.csv"))
    }

    fn temp_jsonl_path(name: &str) -> PathBuf {
        let unique = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .expect("time should move forward")
            .as_nanos();
        std::env::temp_dir().join(format!("query-forge-{name}-{unique}.jsonl"))
    }

    fn temp_json_path(name: &str) -> PathBuf {
        let unique = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .expect("time should move forward")
            .as_nanos();
        std::env::temp_dir().join(format!("query-forge-{name}-{unique}.json"))
    }

    fn temp_markdown_path(name: &str) -> PathBuf {
        let unique = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .expect("time should move forward")
            .as_nanos();
        std::env::temp_dir().join(format!("query-forge-{name}-{unique}.md"))
    }

    fn temp_parquet_path(name: &str) -> PathBuf {
        let unique = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .expect("time should move forward")
            .as_nanos();
        std::env::temp_dir().join(format!("query-forge-{name}-{unique}.parquet"))
    }

    fn write_test_workbook(path: &Path) -> Result<()> {
        let mut workbook = Workbook::new();
        let worksheet = workbook.add_worksheet();

        worksheet.write_string(0, 0, "name")?;
        worksheet.write_string(0, 1, "price")?;
        worksheet.write_string(0, 2, "active")?;
        worksheet.write_string(1, 0, "Keyboard")?;
        worksheet.write_number(1, 1, 12.5)?;
        worksheet.write_boolean(1, 2, true)?;
        worksheet.write_string(2, 0, "Cable")?;
        worksheet.write_number(2, 1, 5.0)?;
        worksheet.write_boolean(2, 2, false)?;

        workbook.save(path)?;
        Ok(())
    }

    fn write_test_workbook_on_sheet(path: &Path, sheet_name: &str) -> Result<()> {
        let mut workbook = Workbook::new();
        let worksheet = workbook.add_worksheet();
        worksheet.set_name(sheet_name)?;

        worksheet.write_string(0, 0, "name")?;
        worksheet.write_string(0, 1, "price")?;
        worksheet.write_string(1, 0, "Keyboard")?;
        worksheet.write_number(1, 1, 12.5)?;

        workbook.save(path)?;
        Ok(())
    }

    fn write_test_xml(path: &Path) -> Result<()> {
        fs::write(
            path,
            r#"<?xml version="1.0" encoding="UTF-8"?>
<data>
    <row>
        <name>Keyboard</name>
        <price>12.5</price>
        <active>true</active>
    </row>
    <row>
        <name>Cable</name>
        <price>5</price>
        <active>false</active>
    </row>
</data>
"#,
        )?;

        Ok(())
    }

    fn write_test_xml_with_sections(path: &Path) -> Result<()> {
        fs::write(
            path,
            r#"<?xml version="1.0" encoding="UTF-8"?>
<root>
    <Inventory>
        <row>
            <name>Keyboard</name>
            <price>12.5</price>
        </row>
    </Inventory>
    <Archive>
        <row>
            <name>Legacy Cable</name>
            <price>3.5</price>
        </row>
    </Archive>
</root>
"#,
        )?;

        Ok(())
    }

    fn write_test_csv(path: &Path) -> Result<()> {
        fs::write(
            path,
            "name,price,active\nKeyboard,12.5,true\nCable,5,false\n",
        )?;

        Ok(())
    }

    fn write_test_csv_with_custom_inference_values(path: &Path) -> Result<()> {
        fs::write(
            path,
            "name,amount,flag,date,note\nKeyboard,\"1,5\",YES,31/12/2025,N/A\nCable,\"2,0\",NO,01/01/2026,ok\n",
        )?;

        Ok(())
    }

    fn write_test_csv_no_headers(path: &Path) -> Result<()> {
        fs::write(path, "Keyboard,12.5,true\nCable,5,false\n")?;

        Ok(())
    }

    fn write_test_csv_for_normalization(path: &Path) -> Result<()> {
        fs::write(
            path,
            "First Name,First-Name, Notes \n Alice ,Alice, hello \n  ,  ,   \n",
        )?;

        Ok(())
    }

    fn write_test_jsonl(path: &Path) -> Result<()> {
        fs::write(
            path,
            "{\"name\":\"Keyboard\",\"price\":12.5,\"active\":true}\n{\"name\":\"Cable\",\"price\":5,\"active\":false}\n",
        )?;

        Ok(())
    }

    fn write_test_json(path: &Path) -> Result<()> {
        fs::write(
            path,
            "[{\"name\":\"Keyboard\",\"price\":12.5,\"active\":true},{\"name\":\"Cable\",\"price\":5,\"active\":false}]",
        )?;

        Ok(())
    }

    fn write_test_json_with_sections(path: &Path) -> Result<()> {
        fs::write(
            path,
            "{\"Inventory\":[{\"name\":\"Keyboard\",\"price\":12.5}],\"Archive\":[{\"name\":\"Legacy Cable\",\"price\":3.5}]}",
        )?;

        Ok(())
    }

    fn write_test_markdown_with_tables(path: &Path) -> Result<()> {
        fs::write(
            path,
            r#"# Inventory Report

| name | price | active |
| --- | ---: | :---: |
| Keyboard | 12.5 | true |
| Cable | 5 | false |

## Archive

| name | price |
| --- | --- |
| Legacy Cable | 3.5 |
"#,
        )?;

        Ok(())
    }

    fn write_test_markdown_with_headers_only(path: &Path) -> Result<()> {
        fs::write(
            path,
            r#"| name | price |
| --- | --- |
"#,
        )?;

        Ok(())
    }

    #[test]
    fn normalizes_duplicate_and_blank_headers() {
        let headers = vec![
            Data::String("name".into()),
            Data::String(String::new()),
            Data::String("name".into()),
        ];

        assert_eq!(
            normalize_headers(&headers, 3),
            vec!["name", "column2", "name_2"]
        );
    }

    #[test]
    fn applies_input_normalization_options_before_query() -> Result<()> {
        let csv_path = temp_csv_path("csv-normalization");
        write_test_csv_for_normalization(&csv_path)?;
        let options = InputNormalizationOptions {
            trim: true,
            skip_empty_rows: true,
            normalize_headers: true,
            header_case: Some(HeaderCase::Snake),
            dedupe_headers: true,
        };

        let inputs = [WorkbookInput {
            path: &csv_path,
            sheet_name: None,
            table_name: None,
        }];
        let result = run_query_with_params_multi_inputs_and_options_and_normalization(
            &inputs,
            "SELECT first_name, first_name_2, notes FROM table",
            &[],
            &TypeInferenceOptions::default(),
            &options,
            &ExtractionOptions::default(),
            true,
        )?;

        assert_eq!(result.columns, vec!["first_name", "first_name_2", "notes"]);
        assert_eq!(
            result.rows,
            vec![vec![
                QueryValue::Text("Alice".into()),
                QueryValue::Text("Alice".into()),
                QueryValue::Text("hello".into())
            ]]
        );

        fs::remove_file(csv_path)?;
        Ok(())
    }

    #[test]
    fn normalizes_special_character_headers_to_column_name() {
        assert_eq!(
            normalize_header_value(" !!! ", 3, Some(&HeaderCase::Snake)),
            "column3"
        );
    }

    #[test]
    fn dedupe_headers_handles_generated_name_collisions() {
        let mut headers = vec![
            "name".to_owned(),
            "name".to_owned(),
            "name_2".to_owned(),
            "name".to_owned(),
        ];

        dedupe_headers(&mut headers);

        assert_eq!(headers, vec!["name", "name_2", "name_2_2", "name_3"]);
    }

    #[test]
    fn executes_sql_query_against_sheet() -> Result<()> {
        let workbook_path = temp_path("query");
        write_test_workbook(&workbook_path)?;

        let result = run_query(
            &workbook_path,
            Some("Sheet1"),
            "SELECT name, price FROM table WHERE price > 10 ORDER BY price DESC",
            true,
        )?;

        assert_eq!(result.columns, vec!["name", "price"]);
        assert_eq!(
            result.rows,
            vec![vec![
                QueryValue::Text("Keyboard".into()),
                QueryValue::Real(12.5)
            ]]
        );

        fs::remove_file(workbook_path)?;
        Ok(())
    }

    #[test]
    fn executes_sql_query_against_sheet1_alias() -> Result<()> {
        let workbook_path = temp_path("query-sheet1-alias");
        write_test_workbook(&workbook_path)?;

        let result = run_query(
            &workbook_path,
            Some("Sheet1"),
            "SELECT name, price FROM table1 WHERE price > 10 ORDER BY price DESC",
            true,
        )?;

        assert_eq!(result.columns, vec!["name", "price"]);
        assert_eq!(
            result.rows,
            vec![vec![
                QueryValue::Text("Keyboard".into()),
                QueryValue::Real(12.5)
            ]]
        );

        fs::remove_file(workbook_path)?;
        Ok(())
    }

    #[test]
    fn executes_sql_query_with_named_parameter() -> Result<()> {
        let workbook_path = temp_path("query-with-param");
        write_test_workbook(&workbook_path)?;

        let result = run_query_with_params(
            &workbook_path,
            Some("Sheet1"),
            "SELECT name, price FROM table WHERE price > :min_price ORDER BY price DESC",
            &[QueryParam {
                name: "min_price".to_owned(),
                value: QueryValue::Real(10.0),
            }],
            true,
        )?;

        assert_eq!(
            result,
            QueryResult {
                columns: vec!["name".to_owned(), "price".to_owned()],
                rows: vec![vec![
                    QueryValue::Text("Keyboard".to_owned()),
                    QueryValue::Real(12.5),
                ]],
            }
        );

        fs::remove_file(&workbook_path)?;
        Ok(())
    }

    #[test]
    fn executes_query_against_multiple_workbooks() -> Result<()> {
        let workbook_path_1 = temp_path("multi-1");
        let workbook_path_2 = temp_path("multi-2");
        write_test_workbook(&workbook_path_1)?;
        write_test_workbook(&workbook_path_2)?;

        let result = run_query_with_params_multi(
            &[workbook_path_1.as_path(), workbook_path_2.as_path()],
            Some("Sheet1"),
            "SELECT COUNT(*) AS total_rows FROM table UNION ALL SELECT COUNT(*) AS total_rows FROM table2",
            &[],
            true,
        )?;

        assert_eq!(result.columns, vec!["total_rows"]);
        assert_eq!(
            result.rows,
            vec![vec![QueryValue::Integer(2)], vec![QueryValue::Integer(2)]]
        );

        fs::remove_file(&workbook_path_1)?;
        fs::remove_file(&workbook_path_2)?;
        Ok(())
    }

    #[test]
    fn executes_query_against_multiple_workbooks_with_distinct_sheet_names() -> Result<()> {
        let workbook_path_1 = temp_path("multi-sheet-1");
        let workbook_path_2 = temp_path("multi-sheet-2");
        write_test_workbook_on_sheet(&workbook_path_1, "Consuntivo")?;
        write_test_workbook_on_sheet(&workbook_path_2, "WKL")?;

        let result = run_query_with_params_multi_inputs(
            &[
                WorkbookInput {
                    path: workbook_path_1.as_path(),
                    sheet_name: Some("Consuntivo"),
                    table_name: None,
                },
                WorkbookInput {
                    path: workbook_path_2.as_path(),
                    sheet_name: Some("WKL"),
                    table_name: None,
                },
            ],
            "SELECT COUNT(*) AS total_rows FROM table UNION ALL SELECT COUNT(*) AS total_rows FROM table2",
            &[],
            true,
        )?;

        assert_eq!(result.columns, vec!["total_rows"]);
        assert_eq!(
            result.rows,
            vec![vec![QueryValue::Integer(1)], vec![QueryValue::Integer(1)]]
        );

        fs::remove_file(&workbook_path_1)?;
        fs::remove_file(&workbook_path_2)?;
        Ok(())
    }

    #[test]
    fn executes_query_with_explicit_table_names() -> Result<()> {
        let sales_path = temp_path("explicit-name-sales");
        let costs_path = temp_path("explicit-name-costs");
        write_test_workbook_on_sheet(&sales_path, "Sheet1")?;
        write_test_workbook_on_sheet(&costs_path, "Sheet1")?;

        let result = run_query_with_params_multi_inputs(
            &[
                WorkbookInput {
                    path: sales_path.as_path(),
                    sheet_name: Some("Sheet1"),
                    table_name: Some("sales"),
                },
                WorkbookInput {
                    path: costs_path.as_path(),
                    sheet_name: Some("Sheet1"),
                    table_name: Some("costs"),
                },
            ],
            "SELECT COUNT(*) AS total_rows FROM sales UNION ALL SELECT COUNT(*) AS total_rows FROM costs",
            &[],
            true,
        )?;

        assert_eq!(result.columns, vec!["total_rows"]);
        assert_eq!(
            result.rows,
            vec![vec![QueryValue::Integer(1)], vec![QueryValue::Integer(1)]]
        );

        fs::remove_file(&sales_path)?;
        fs::remove_file(&costs_path)?;
        Ok(())
    }

    #[test]
    fn executes_sql_query_against_whole_xml_file() -> Result<()> {
        let xml_path = temp_xml_path("xml-whole");
        write_test_xml(&xml_path)?;

        let result = run_query(
            &xml_path,
            None,
            "SELECT name, price FROM table WHERE active = 1 ORDER BY price DESC",
            true,
        )?;

        assert_eq!(result.columns, vec!["name", "price"]);
        assert_eq!(
            result.rows,
            vec![vec![
                QueryValue::Text("Keyboard".into()),
                QueryValue::Real(12.5)
            ]]
        );

        fs::remove_file(xml_path)?;
        Ok(())
    }

    #[test]
    fn executes_sql_query_against_xml_sheet_tag() -> Result<()> {
        let xml_path = temp_xml_path("xml-sheet-tag");
        write_test_xml_with_sections(&xml_path)?;

        let result = run_query(
            &xml_path,
            Some("Archive"),
            "SELECT name, price FROM table",
            true,
        )?;

        assert_eq!(result.columns, vec!["name", "price"]);
        assert_eq!(
            result.rows,
            vec![vec![
                QueryValue::Text("Legacy Cable".into()),
                QueryValue::Real(3.5)
            ]]
        );

        fs::remove_file(xml_path)?;
        Ok(())
    }

    #[test]
    fn executes_query_against_heterogeneous_xlsx_and_xml_inputs() -> Result<()> {
        let workbook_path = temp_path("mixed-xlsx");
        let xml_path = temp_xml_path("mixed-xml");
        write_test_workbook(&workbook_path)?;
        write_test_xml(&xml_path)?;

        let result = run_query_with_params_multi_inputs(
            &[
                WorkbookInput {
                    path: workbook_path.as_path(),
                    sheet_name: Some("Sheet1"),
                    table_name: None,
                },
                WorkbookInput {
                    path: xml_path.as_path(),
                    sheet_name: None,
                    table_name: None,
                },
            ],
            "SELECT COUNT(*) AS total_rows FROM table UNION ALL SELECT COUNT(*) AS total_rows FROM table2",
            &[],
            true,
        )?;

        assert_eq!(result.columns, vec!["total_rows"]);
        assert_eq!(
            result.rows,
            vec![vec![QueryValue::Integer(2)], vec![QueryValue::Integer(2)]]
        );

        fs::remove_file(&workbook_path)?;
        fs::remove_file(&xml_path)?;
        Ok(())
    }

    #[test]
    fn executes_sql_query_against_csv_file() -> Result<()> {
        let csv_path = temp_csv_path("csv");
        write_test_csv(&csv_path)?;

        let result = run_query(
            &csv_path,
            None,
            "SELECT name, price FROM table WHERE active = 1 ORDER BY price DESC",
            true,
        )?;

        assert_eq!(result.columns, vec!["name", "price"]);
        assert_eq!(
            result.rows,
            vec![vec![
                QueryValue::Text("Keyboard".into()),
                QueryValue::Real(12.5)
            ]]
        );

        fs::remove_file(csv_path)?;
        Ok(())
    }

    #[test]
    fn supports_configurable_type_inference_options() -> Result<()> {
        let csv_path = temp_csv_path("csv-custom-inference");
        write_test_csv_with_custom_inference_values(&csv_path)?;

        let options = TypeInferenceOptions {
            infer_types: true,
            decimal_comma: true,
            date_format: Some("%d/%m/%Y".to_owned()),
            null_values: vec!["N/A".to_owned()],
            true_values: vec!["YES".to_owned()],
            false_values: vec!["NO".to_owned()],
        };

        let workbook_inputs = [WorkbookInput {
            path: csv_path.as_path(),
            sheet_name: None,
            table_name: None,
        }];
        let result = run_query_with_params_multi_inputs_and_options(
            &workbook_inputs,
            "SELECT amount, flag, date, note FROM table ORDER BY amount",
            &[],
            &options,
            true,
        )?;

        assert_eq!(result.columns, vec!["amount", "flag", "date", "note"]);
        assert_eq!(
            result.rows,
            vec![
                vec![
                    QueryValue::Real(1.5),
                    QueryValue::Integer(1),
                    QueryValue::Text("2025-12-31".into()),
                    QueryValue::Null
                ],
                vec![
                    QueryValue::Real(2.0),
                    QueryValue::Integer(0),
                    QueryValue::Text("2026-01-01".into()),
                    QueryValue::Text("ok".into())
                ]
            ]
        );

        fs::remove_file(csv_path)?;
        Ok(())
    }

    #[test]
    fn supports_all_text_mode() -> Result<()> {
        let csv_path = temp_csv_path("csv-all-text");
        write_test_csv(&csv_path)?;

        let options = TypeInferenceOptions {
            infer_types: false,
            ..TypeInferenceOptions::default()
        };
        let workbook_inputs = [WorkbookInput {
            path: csv_path.as_path(),
            sheet_name: None,
            table_name: None,
        }];
        let result = run_query_with_params_multi_inputs_and_options(
            &workbook_inputs,
            "SELECT name, price, active FROM table ORDER BY name DESC",
            &[],
            &options,
            true,
        )?;

        assert_eq!(
            result.rows,
            vec![
                vec![
                    QueryValue::Text("Keyboard".into()),
                    QueryValue::Text("12.5".into()),
                    QueryValue::Text("true".into())
                ],
                vec![
                    QueryValue::Text("Cable".into()),
                    QueryValue::Text("5".into()),
                    QueryValue::Text("false".into())
                ]
            ]
        );

        fs::remove_file(csv_path)?;
        Ok(())
    }

    #[test]
    fn executes_sql_query_against_csv_without_headers() -> Result<()> {
        let csv_path = temp_csv_path("csv-no-headers");
        write_test_csv_no_headers(&csv_path)?;

        let result = run_query(
            &csv_path,
            None,
            "SELECT column1, column2 FROM table WHERE column3 = 1 ORDER BY column2 DESC",
            false,
        )?;

        assert_eq!(result.columns, vec!["column1", "column2"]);
        assert_eq!(
            result.rows,
            vec![vec![
                QueryValue::Text("Keyboard".into()),
                QueryValue::Real(12.5)
            ]]
        );

        fs::remove_file(csv_path)?;
        Ok(())
    }

    #[test]
    fn executes_sql_query_against_jsonl_file() -> Result<()> {
        let jsonl_path = temp_jsonl_path("jsonl");
        write_test_jsonl(&jsonl_path)?;

        let result = run_query(
            &jsonl_path,
            None,
            "SELECT name, price FROM table WHERE active = 1 ORDER BY price DESC",
            true,
        )?;

        assert_eq!(result.columns, vec!["name", "price"]);
        assert_eq!(
            result.rows,
            vec![vec![
                QueryValue::Text("Keyboard".into()),
                QueryValue::Real(12.5)
            ]]
        );

        fs::remove_file(jsonl_path)?;
        Ok(())
    }

    #[test]
    fn executes_sql_query_against_json_file() -> Result<()> {
        let json_path = temp_json_path("json");
        write_test_json(&json_path)?;

        let result = run_query(
            &json_path,
            None,
            "SELECT name, price FROM table WHERE active = 1 ORDER BY price DESC",
            true,
        )?;

        assert_eq!(result.columns, vec!["name", "price"]);
        assert_eq!(
            result.rows,
            vec![vec![
                QueryValue::Text("Keyboard".into()),
                QueryValue::Real(12.5)
            ]]
        );

        fs::remove_file(json_path)?;
        Ok(())
    }

    #[test]
    fn executes_sql_query_against_json_sheet_key() -> Result<()> {
        let json_path = temp_json_path("json-sheet-key");
        write_test_json_with_sections(&json_path)?;

        let result = run_query(
            &json_path,
            Some("Archive"),
            "SELECT name, price FROM table",
            true,
        )?;

        assert_eq!(result.columns, vec!["name", "price"]);
        assert_eq!(
            result.rows,
            vec![vec![
                QueryValue::Text("Legacy Cable".into()),
                QueryValue::Real(3.5)
            ]]
        );

        fs::remove_file(json_path)?;
        Ok(())
    }

    #[test]
    fn executes_sql_query_against_first_markdown_table_by_default() -> Result<()> {
        let markdown_path = temp_markdown_path("markdown-default");
        write_test_markdown_with_tables(&markdown_path)?;

        let result = run_query(
            &markdown_path,
            None,
            "SELECT name, price FROM table WHERE active = 1",
            true,
        )?;

        assert_eq!(result.columns, vec!["name", "price"]);
        assert_eq!(
            result.rows,
            vec![vec![
                QueryValue::Text("Keyboard".into()),
                QueryValue::Real(12.5)
            ]]
        );

        fs::remove_file(markdown_path)?;
        Ok(())
    }

    #[test]
    fn reports_actionable_error_for_csv_selector() -> Result<()> {
        let csv_path = temp_csv_path("csv-selector-error");
        write_test_csv(&csv_path)?;

        let error = run_query(&csv_path, Some("Sheet1"), "SELECT name FROM table", true)
            .expect_err("CSV selector should fail");
        let message = error.to_string();
        assert!(message.contains("does not support selector 'Sheet1'"));
        assert!(message.contains("Remove ':Sheet1'"));

        fs::remove_file(csv_path)?;
        Ok(())
    }

    #[test]
    fn reports_actionable_error_for_jsonl_selector() -> Result<()> {
        let jsonl_path = temp_jsonl_path("jsonl-selector-error");
        write_test_jsonl(&jsonl_path)?;

        let error = run_query(&jsonl_path, Some("Records"), "SELECT name FROM table", true)
            .expect_err("JSONL selector should fail");
        let message = error.to_string();
        assert!(message.contains("does not support selector 'Records'"));
        assert!(message.contains("Remove ':Records'"));

        fs::remove_file(jsonl_path)?;
        Ok(())
    }

    #[test]
    fn reports_json_key_error_with_available_keys() -> Result<()> {
        let json_path = temp_json_path("json-missing-key");
        write_test_json_with_sections(&json_path)?;

        let error = run_query(&json_path, Some("Missing"), "SELECT name FROM table", true)
            .expect_err("missing JSON key should fail");
        let message = error.to_string();
        assert!(message.contains("JSON key 'Missing' not found"));
        assert!(message.contains("Available keys: Archive, Inventory"));

        fs::remove_file(json_path)?;
        Ok(())
    }

    #[test]
    fn reports_invalid_markdown_selector_with_guidance() -> Result<()> {
        let markdown_path = temp_markdown_path("markdown-invalid-selector");
        write_test_markdown_with_tables(&markdown_path)?;

        let error = run_query(&markdown_path, Some("abc"), "SELECT name FROM table", true)
            .expect_err("non-numeric markdown selector should fail");
        let message = error.to_string();
        assert!(message.contains("invalid Markdown table selector 'abc'"));
        assert!(message.contains("':1'"));

        fs::remove_file(markdown_path)?;
        Ok(())
    }

    #[test]
    fn reports_empty_markdown_table_as_actionable_error() -> Result<()> {
        let markdown_path = temp_markdown_path("markdown-empty-table");
        write_test_markdown_with_headers_only(&markdown_path)?;

        let error = run_query(&markdown_path, None, "SELECT name FROM table", true)
            .expect_err("empty markdown table should fail");
        let message = error.to_string();
        assert!(message.contains("is empty (no data rows)"));

        fs::remove_file(markdown_path)?;
        Ok(())
    }

    #[test]
    fn reports_unknown_table_with_inspection_hint() -> Result<()> {
        let workbook_path = temp_path("unknown-table-query");
        write_test_workbook(&workbook_path)?;

        let error = run_query(
            &workbook_path,
            Some("Sheet1"),
            "SELECT * FROM missing_table",
            true,
        )
        .expect_err("unknown table should fail");
        let message = error.to_string();
        assert!(message.contains("unknown table 'missing_table'"));
        assert!(message.contains("qf tables --input"));

        fs::remove_file(workbook_path)?;
        Ok(())
    }

    #[test]
    fn reports_unknown_column_with_schema_hint() -> Result<()> {
        let workbook_path = temp_path("unknown-column-query");
        write_test_workbook(&workbook_path)?;

        let error = run_query(
            &workbook_path,
            Some("Sheet1"),
            "SELECT missing_column FROM table",
            true,
        )
        .expect_err("unknown column should fail");
        let message = error.to_string();
        assert!(message.contains("unknown column 'missing_column'"));
        assert!(message.contains("qf schema --input"));

        fs::remove_file(workbook_path)?;
        Ok(())
    }

    #[test]
    fn executes_sql_query_against_markdown_table_by_numeric_key() -> Result<()> {
        let markdown_path = temp_markdown_path("markdown-key");
        write_test_markdown_with_tables(&markdown_path)?;

        let result = run_query(
            &markdown_path,
            Some("2"),
            "SELECT name, price FROM table",
            true,
        )?;

        assert_eq!(result.columns, vec!["name", "price"]);
        assert_eq!(
            result.rows,
            vec![vec![
                QueryValue::Text("Legacy Cable".into()),
                QueryValue::Real(3.5)
            ]]
        );

        fs::remove_file(markdown_path)?;
        Ok(())
    }

    #[test]
    fn executes_query_against_heterogeneous_xlsx_xml_csv_jsonl_json_markdown_inputs() -> Result<()>
    {
        let workbook_path = temp_path("mixed4-xlsx");
        let xml_path = temp_xml_path("mixed4-xml");
        let csv_path = temp_csv_path("mixed4-csv");
        let jsonl_path = temp_jsonl_path("mixed4-jsonl");
        let json_path = temp_json_path("mixed4-json");
        let markdown_path = temp_markdown_path("mixed4-markdown");
        write_test_workbook(&workbook_path)?;
        write_test_xml(&xml_path)?;
        write_test_csv(&csv_path)?;
        write_test_jsonl(&jsonl_path)?;
        write_test_json(&json_path)?;
        write_test_markdown_with_tables(&markdown_path)?;

        let result = run_query_with_params_multi_inputs(
            &[
                WorkbookInput {
                    path: workbook_path.as_path(),
                    sheet_name: Some("Sheet1"),
                    table_name: None,
                },
                WorkbookInput {
                    path: xml_path.as_path(),
                    sheet_name: None,
                    table_name: None,
                },
                WorkbookInput {
                    path: csv_path.as_path(),
                    sheet_name: None,
                    table_name: None,
                },
                WorkbookInput {
                    path: jsonl_path.as_path(),
                    sheet_name: None,
                    table_name: None,
                },
                WorkbookInput {
                    path: json_path.as_path(),
                    sheet_name: None,
                    table_name: None,
                },
                WorkbookInput {
                    path: markdown_path.as_path(),
                    sheet_name: None,
                    table_name: None,
                },
            ],
            "SELECT COUNT(*) AS total_rows FROM table UNION ALL SELECT COUNT(*) AS total_rows FROM table2 UNION ALL SELECT COUNT(*) AS total_rows FROM table3 UNION ALL SELECT COUNT(*) AS total_rows FROM table4 UNION ALL SELECT COUNT(*) AS total_rows FROM table5 UNION ALL SELECT COUNT(*) AS total_rows FROM table6",
            &[],
            true,
        )?;

        assert_eq!(result.columns, vec!["total_rows"]);
        assert_eq!(
            result.rows,
            vec![
                vec![QueryValue::Integer(2)],
                vec![QueryValue::Integer(2)],
                vec![QueryValue::Integer(2)],
                vec![QueryValue::Integer(2)],
                vec![QueryValue::Integer(2)],
                vec![QueryValue::Integer(2)]
            ]
        );

        fs::remove_file(&workbook_path)?;
        fs::remove_file(&xml_path)?;
        fs::remove_file(&csv_path)?;
        fs::remove_file(&jsonl_path)?;
        fs::remove_file(&json_path)?;
        fs::remove_file(&markdown_path)?;
        Ok(())
    }

    #[test]
    fn writes_query_result_to_xlsx() -> Result<()> {
        let output_path = temp_path("output");
        let result = QueryResult {
            columns: vec!["item".into(), "total".into()],
            rows: vec![vec![
                QueryValue::Text("Mouse".into()),
                QueryValue::Integer(3),
            ]],
        };

        write_xlsx(&result, &output_path)?;

        let written = run_query(
            &output_path,
            Some("Sheet1"),
            "SELECT item, total FROM table",
            true,
        )?;

        assert_eq!(written.columns, vec!["item", "total"]);
        assert_eq!(
            written.rows,
            vec![vec![
                QueryValue::Text("Mouse".into()),
                QueryValue::Real(3.0)
            ]]
        );

        fs::remove_file(output_path)?;
        Ok(())
    }

    #[test]
    fn writes_query_result_to_parquet() -> Result<()> {
        let output_path = temp_parquet_path("output");
        let result = QueryResult {
            columns: vec!["item".into(), "qty".into(), "price".into()],
            rows: vec![
                vec![
                    QueryValue::Text("Mouse".into()),
                    QueryValue::Integer(3),
                    QueryValue::Real(9.99),
                ],
                vec![
                    QueryValue::Text("Keyboard".into()),
                    QueryValue::Integer(1),
                    QueryValue::Null,
                ],
            ],
        };

        write_parquet(&result, &output_path)?;

        // Verify the file exists, is non-empty, and has the Parquet magic bytes (PAR1).
        let bytes = fs::read(&output_path)?;
        assert!(bytes.len() > 8, "parquet file should have content");
        assert_eq!(&bytes[..4], b"PAR1", "parquet file should start with PAR1");
        assert_eq!(
            &bytes[bytes.len() - 4..],
            b"PAR1",
            "parquet file should end with PAR1"
        );

        fs::remove_file(output_path)?;
        Ok(())
    }

    #[test]
    fn executes_sql_query_against_parquet_file() -> Result<()> {
        let parquet_path = temp_parquet_path("input");
        let result = QueryResult {
            columns: vec!["name".into(), "price".into(), "active".into()],
            rows: vec![
                vec![
                    QueryValue::Text("Keyboard".into()),
                    QueryValue::Real(12.5),
                    QueryValue::Integer(1),
                ],
                vec![
                    QueryValue::Text("Cable".into()),
                    QueryValue::Real(5.0),
                    QueryValue::Integer(0),
                ],
            ],
        };
        write_parquet(&result, &parquet_path)?;

        let queried = run_query(
            &parquet_path,
            None,
            "SELECT name, price FROM table WHERE active = 1 ORDER BY price DESC",
            true,
        )?;

        assert_eq!(queried.columns, vec!["name", "price"]);
        assert_eq!(
            queried.rows,
            vec![vec![
                QueryValue::Text("Keyboard".into()),
                QueryValue::Real(12.5)
            ]]
        );

        fs::remove_file(parquet_path)?;
        Ok(())
    }

    #[test]
    fn reports_actionable_error_for_parquet_selector() -> Result<()> {
        let parquet_path = temp_parquet_path("parquet-selector-error");
        let result = QueryResult {
            columns: vec!["name".into()],
            rows: vec![vec![QueryValue::Text("Keyboard".into())]],
        };
        write_parquet(&result, &parquet_path)?;

        let error = run_query(
            &parquet_path,
            Some("Sheet1"),
            "SELECT name FROM table",
            true,
        )
        .expect_err("Parquet selector should fail");
        let message = error.to_string();
        assert!(message.contains("does not support selector 'Sheet1'"));
        assert!(message.contains("Remove ':Sheet1'"));

        fs::remove_file(parquet_path)?;
        Ok(())
    }

    #[test]
    fn renders_csv_with_escaping() {
        let result = QueryResult {
            columns: vec!["name".into(), "notes".into()],
            rows: vec![vec![
                QueryValue::Text("Mouse".into()),
                QueryValue::Text("line1,line2".into()),
            ]],
        };

        assert_eq!(render_csv(&result), "name,notes\nMouse,\"line1,line2\"");
    }

    #[test]
    fn renders_text_with_aligned_columns() {
        let result = QueryResult {
            columns: vec!["mese".into(), "totale_ore".into()],
            rows: vec![
                vec![
                    QueryValue::Text("2026-01-01".into()),
                    QueryValue::Real(10.5),
                ],
                vec![
                    QueryValue::Text("2026-12-01".into()),
                    QueryValue::Integer(2),
                ],
            ],
        };

        assert_eq!(
            render_text(&result),
            "mese       | totale_ore\n-----------+-----------\n2026-01-01 | 10.5      \n2026-12-01 | 2         "
        );
    }

    #[test]
    fn renders_jsonl() {
        let result = QueryResult {
            columns: vec!["item".into(), "stock".into(), "note".into()],
            rows: vec![vec![
                QueryValue::Text("Desk".into()),
                QueryValue::Integer(8),
                QueryValue::Null,
            ]],
        };

        assert_eq!(
            render_jsonl(&result),
            "{\"item\":\"Desk\",\"stock\":8,\"note\":null}"
        );
    }

    #[test]
    fn renders_json() {
        let result = QueryResult {
            columns: vec!["item".into(), "stock".into(), "note".into()],
            rows: vec![vec![
                QueryValue::Text("Desk".into()),
                QueryValue::Integer(8),
                QueryValue::Null,
            ]],
        };

        assert_eq!(
            render_json(&result),
            "[{\"item\":\"Desk\",\"stock\":8,\"note\":null}]"
        );
    }

    #[test]
    fn renders_markdown() {
        let result = QueryResult {
            columns: vec!["category".into(), "total".into()],
            rows: vec![vec![
                QueryValue::Text("electronics".into()),
                QueryValue::Integer(47),
            ]],
        };

        assert_eq!(
            render_markdown(&result),
            "| category | total |\n| --- | --- |\n| electronics | 47 |"
        );
    }

    // ── JSON extraction modes ─────────────────────────────────────────────────

    fn write_test_json_object(path: &Path) -> Result<()> {
        fs::write(
            path,
            r#"{"name":"Alice","age":30,"city":"Rome"}"#,
        )?;
        Ok(())
    }

    fn write_test_json_nested(path: &Path) -> Result<()> {
        fs::write(
            path,
            r#"[{"user":{"name":"Alice","address":{"city":"Rome"}},"score":10},{"user":{"name":"Bob","address":{"city":"Paris"}},"score":20}]"#,
        )?;
        Ok(())
    }

    #[test]
    fn json_object_mode_turns_keys_into_rows() -> Result<()> {
        let json_path = temp_json_path("json-object-mode");
        write_test_json_object(&json_path)?;

        let opts = ExtractionOptions {
            json_mode: JsonMode::Object,
            ..ExtractionOptions::default()
        };
        let inputs = [WorkbookInput {
            path: &json_path,
            sheet_name: None,
            table_name: None,
        }];
        let result = run_query_with_params_multi_inputs_and_options_and_normalization(
            &inputs,
            "SELECT key, value FROM table ORDER BY key",
            &[],
            &TypeInferenceOptions::default(),
            &InputNormalizationOptions::default(),
            &opts,
            true,
        )?;

        assert_eq!(result.columns, vec!["key", "value"]);
        // Three key-value rows, sorted by key.
        assert_eq!(result.rows.len(), 3);
        assert_eq!(result.rows[0][0], QueryValue::Text("age".into()));
        assert_eq!(result.rows[0][1], QueryValue::Integer(30));
        assert_eq!(result.rows[1][0], QueryValue::Text("city".into()));
        assert_eq!(result.rows[1][1], QueryValue::Text("Rome".into()));
        assert_eq!(result.rows[2][0], QueryValue::Text("name".into()));
        assert_eq!(result.rows[2][1], QueryValue::Text("Alice".into()));

        fs::remove_file(json_path)?;
        Ok(())
    }

    #[test]
    fn json_flatten_mode_expands_nested_objects() -> Result<()> {
        let json_path = temp_json_path("json-flatten-mode");
        write_test_json_nested(&json_path)?;

        let opts = ExtractionOptions {
            json_mode: JsonMode::Flatten,
            ..ExtractionOptions::default()
        };
        let inputs = [WorkbookInput {
            path: &json_path,
            sheet_name: None,
            table_name: None,
        }];
        let result = run_query_with_params_multi_inputs_and_options_and_normalization(
            &inputs,
            "SELECT \"user.name\", \"user.address.city\", score FROM table ORDER BY score",
            &[],
            &TypeInferenceOptions::default(),
            &InputNormalizationOptions::default(),
            &opts,
            true,
        )?;

        assert!(result.columns.contains(&"user.name".to_owned()));
        assert!(result.columns.contains(&"user.address.city".to_owned()));
        assert_eq!(result.rows.len(), 2);
        // First row: Alice / Rome / 10
        let alice_col = result.columns.iter().position(|c| c == "user.name").unwrap();
        let city_col = result.columns.iter().position(|c| c == "user.address.city").unwrap();
        let score_col = result.columns.iter().position(|c| c == "score").unwrap();
        assert_eq!(result.rows[0][alice_col], QueryValue::Text("Alice".into()));
        assert_eq!(result.rows[0][city_col], QueryValue::Text("Rome".into()));
        assert_eq!(result.rows[0][score_col], QueryValue::Integer(10));

        fs::remove_file(json_path)?;
        Ok(())
    }

    #[test]
    fn json_array_mode_is_default_behavior() -> Result<()> {
        let json_path = temp_json_path("json-array-mode-default");
        write_test_json(&json_path)?;

        let opts = ExtractionOptions {
            json_mode: JsonMode::Array,
            ..ExtractionOptions::default()
        };
        let inputs = [WorkbookInput {
            path: &json_path,
            sheet_name: None,
            table_name: None,
        }];
        let result = run_query_with_params_multi_inputs_and_options_and_normalization(
            &inputs,
            "SELECT name, price FROM table WHERE active = 1",
            &[],
            &TypeInferenceOptions::default(),
            &InputNormalizationOptions::default(),
            &opts,
            true,
        )?;

        assert_eq!(result.columns, vec!["name", "price"]);
        assert_eq!(
            result.rows,
            vec![vec![
                QueryValue::Text("Keyboard".into()),
                QueryValue::Real(12.5)
            ]]
        );

        fs::remove_file(json_path)?;
        Ok(())
    }

    // ── XML extraction modes ──────────────────────────────────────────────────

    fn write_test_xml_with_attributes(path: &Path) -> Result<()> {
        fs::write(
            path,
            r#"<?xml version="1.0" encoding="UTF-8"?>
<products>
    <product id="1" name="Keyboard" price="12.5" active="true"/>
    <product id="2" name="Cable" price="5" active="false"/>
</products>
"#,
        )?;
        Ok(())
    }

    fn write_test_xml_leaf_elements(path: &Path) -> Result<()> {
        fs::write(
            path,
            r#"<?xml version="1.0" encoding="UTF-8"?>
<config>
    <host>localhost</host>
    <port>5432</port>
    <database>mydb</database>
</config>
"#,
        )?;
        Ok(())
    }

    #[test]
    fn xml_attributes_mode_extracts_attributes_as_columns() -> Result<()> {
        let xml_path = temp_xml_path("xml-attributes-mode");
        write_test_xml_with_attributes(&xml_path)?;

        let opts = ExtractionOptions {
            xml_mode: XmlMode::Attributes,
            ..ExtractionOptions::default()
        };
        let inputs = [WorkbookInput {
            path: &xml_path,
            sheet_name: None,
            table_name: None,
        }];
        let result = run_query_with_params_multi_inputs_and_options_and_normalization(
            &inputs,
            "SELECT id, name, price FROM table ORDER BY id",
            &[],
            &TypeInferenceOptions::default(),
            &InputNormalizationOptions::default(),
            &opts,
            true,
        )?;

        assert_eq!(result.rows.len(), 2);
        let id_col = result.columns.iter().position(|c| c == "id").unwrap();
        let name_col = result.columns.iter().position(|c| c == "name").unwrap();
        assert_eq!(result.rows[0][id_col], QueryValue::Integer(1));
        assert_eq!(result.rows[0][name_col], QueryValue::Text("Keyboard".into()));
        assert_eq!(result.rows[1][id_col], QueryValue::Integer(2));
        assert_eq!(result.rows[1][name_col], QueryValue::Text("Cable".into()));

        fs::remove_file(xml_path)?;
        Ok(())
    }

    #[test]
    fn xml_descendants_mode_collects_leaf_elements_as_tag_value_rows() -> Result<()> {
        let xml_path = temp_xml_path("xml-descendants-mode");
        write_test_xml_leaf_elements(&xml_path)?;

        let opts = ExtractionOptions {
            xml_mode: XmlMode::Descendants,
            ..ExtractionOptions::default()
        };
        let inputs = [WorkbookInput {
            path: &xml_path,
            sheet_name: None,
            table_name: None,
        }];
        let result = run_query_with_params_multi_inputs_and_options_and_normalization(
            &inputs,
            "SELECT tag, value FROM table ORDER BY tag",
            &[],
            &TypeInferenceOptions::default(),
            &InputNormalizationOptions::default(),
            &opts,
            true,
        )?;

        assert_eq!(result.columns, vec!["tag", "value"]);
        assert_eq!(result.rows.len(), 3);
        // Sorted alphabetically: database, host, port
        assert_eq!(result.rows[0][0], QueryValue::Text("database".into()));
        assert_eq!(result.rows[0][1], QueryValue::Text("mydb".into()));
        assert_eq!(result.rows[1][0], QueryValue::Text("host".into()));
        assert_eq!(result.rows[1][1], QueryValue::Text("localhost".into()));
        assert_eq!(result.rows[2][0], QueryValue::Text("port".into()));
        assert_eq!(result.rows[2][1], QueryValue::Integer(5432));

        fs::remove_file(xml_path)?;
        Ok(())
    }

    #[test]
    fn xml_rows_mode_is_default_behavior() -> Result<()> {
        let xml_path = temp_xml_path("xml-rows-mode-default");
        write_test_xml(&xml_path)?;

        let opts = ExtractionOptions {
            xml_mode: XmlMode::Rows,
            ..ExtractionOptions::default()
        };
        let inputs = [WorkbookInput {
            path: &xml_path,
            sheet_name: None,
            table_name: None,
        }];
        let result = run_query_with_params_multi_inputs_and_options_and_normalization(
            &inputs,
            "SELECT name, price FROM table WHERE active = 1 ORDER BY price DESC",
            &[],
            &TypeInferenceOptions::default(),
            &InputNormalizationOptions::default(),
            &opts,
            true,
        )?;

        assert_eq!(result.columns, vec!["name", "price"]);
        assert_eq!(
            result.rows,
            vec![vec![
                QueryValue::Text("Keyboard".into()),
                QueryValue::Real(12.5)
            ]]
        );

        fs::remove_file(xml_path)?;
        Ok(())
    }
}