dataprof 0.12.0

//! Cross-engine consistency test.
//!
//! Profiles the same values through file engines and column analysis (used by
//! database connectors), then asserts exact equality of serialized metrics (#547).
//! Raw numeric diagnostics allow relative tolerance 1e-9 or absolute tolerance 1e-12.

use std::io::Write;
use std::path::Path;

use dataprof::{ChunkSize, EngineType, Profiler, StopCondition, TruncationReason};
use dataprof::{
    ColumnStats, CsvParserConfig, DataType, MetricConfidence, QualityDimension, analyze_csv_file,
};
use serde_json::json;
use tempfile::NamedTempFile;

/// Frequency fields used to be present only on the database column-analysis
/// path (#709). Comparing file engines alone cannot expose that divergence.
#[test]
fn serialized_text_stats_match_column_analysis_and_csv_engines() {
    for values in [
        vec!["hello", "world", "test", "hello", "world"],
        vec!["東京", "café", "e\u{301}", "東京", "", "  ", "NULL"],
        vec!["", "  ", "NULL"],
    ] {
        let data: Vec<String> = values.iter().map(|value| (*value).to_owned()).collect();
        let mut csv = NamedTempFile::new().unwrap();
        writeln!(csv, "text,row").unwrap();
        for (row, value) in values.iter().enumerate() {
            // A second column keeps empty text cells from becoming blank lines.
            writeln!(csv, "{value},{row}").unwrap();
        }
        csv.flush().unwrap();

        let standard = analyze_csv_file(csv.path(), &CsvParserConfig::default()).unwrap();
        let expected = serde_json::to_value(&standard.column_profiles[0].stats).unwrap();
        assert!(expected["Text"].is_object(), "{values:?}");
        assert!(expected["Text"].get("most_frequent").is_none());
        assert!(expected["Text"].get("least_frequent").is_none());

        for column in [
            dataprof::analyze_column("text", &data),
            dataprof::analyze_column_fast("text", &data),
            dataprof::analyze_column_with_analysis_options(
                "text",
                &data,
                &dataprof::AnalysisOptions::default(),
            ),
        ] {
            assert_eq!(column.data_type, standard.column_profiles[0].data_type);
            assert_eq!(column.null_count, standard.column_profiles[0].null_count);
            assert_eq!(column.total_count, standard.column_profiles[0].total_count);
            assert_eq!(
                serde_json::to_value(&column.stats).unwrap(),
                expected,
                "column analysis, {values:?}"
            );
        }

        for engine in [
            EngineType::Auto,
            EngineType::Incremental,
            EngineType::Columnar,
        ] {
            let report = Profiler::new()
                .engine(engine)
                .analyze_file(csv.path())
                .unwrap();
            assert_eq!(
                serde_json::to_value(&report.column_profiles[0].stats).unwrap(),
                expected,
                "{engine:?}, {values:?}"
            );
        }
    }
}

/// New profiles leave frequencies unassessed, but stored Rust statistics must
/// retain measurements made by earlier versions, including measured-empty lists.
#[test]
fn historical_text_frequencies_survive_rust_round_trip() {
    for frequencies in [
        json!([{"value": "hello", "count": 2, "percentage": 100.0}]),
        json!([]),
    ] {
        let document = json!({"Text": {
            "min_length": 5,
            "max_length": 5,
            "avg_length": 5.0,
            "most_frequent": frequencies,
            "least_frequent": frequencies,
        }});
        let restored: ColumnStats = serde_json::from_value(document.clone()).unwrap();
        assert_eq!(serde_json::to_value(restored).unwrap(), document);
    }
}

/// 30k rows of *sorted* values — larger than the 10k per-column sample
/// reservoirs, so any engine that derives base statistics from its retained
/// sample (or samples a biased prefix) reports wildly wrong min/max/mean.
fn create_sorted_30k_csv() -> NamedTempFile {
    let mut f = NamedTempFile::new().unwrap();
    writeln!(f, "id,value").unwrap();
    for i in 0..30_000 {
        writeln!(f, "{},{:.2}", i, i as f64 / 2.0).unwrap();
    }
    f.flush().unwrap();
    f
}

/// The Arrow CSV reader must preserve raw cells and exclude non-finite
/// numbers before updating its numeric accumulator (#678).
#[test]
fn null_and_non_finite_numeric_tokens_match_across_csv_engines() {
    for (token, null_count) in [
        ("", 1),
        ("N/A", 0),
        ("NaN", 1),
        ("inf", 0),
        ("-inf", 0),
        ("1e400", 0),
    ] {
        let mut csv = NamedTempFile::new().unwrap();
        writeln!(csv, "amount,label").unwrap();
        for i in 0..10 {
            writeln!(csv, "{i}.5,row{i}").unwrap();
        }
        writeln!(csv, "{token},special").unwrap();
        csv.flush().unwrap();

        let standard = analyze_csv_file(csv.path(), &CsvParserConfig::default()).unwrap();
        let expected = &standard.column_profiles[0];
        assert_eq!(expected.data_type, DataType::Float, "{token:?}");
        assert_eq!(expected.total_count, 11, "{token:?}");
        assert_eq!(expected.null_count, null_count, "{token:?}");
        assert_eq!(expected.invalid_count, Some(1 - null_count), "{token:?}");
        let ColumnStats::Numeric(stats) = &expected.stats else {
            panic!("expected numeric statistics for {token:?}");
        };
        assert_eq!(stats.min, 0.5, "{token:?}");
        assert_eq!(stats.max, 9.5, "{token:?}");
        assert_eq!(stats.mean, 5.0, "{token:?}");

        for engine in [
            EngineType::Auto,
            EngineType::Incremental,
            EngineType::Columnar,
        ] {
            let report = Profiler::new()
                .engine(engine)
                .analyze_file(csv.path())
                .unwrap();
            assert_eq!(
                serde_json::to_value(&report.column_profiles).unwrap(),
                serde_json::to_value(&standard.column_profiles).unwrap(),
                "{engine:?}, token {token:?}"
            );
        }
    }
}

#[test]
fn test_header_only_csv_columns_match_across_engines_and_serialization() {
    let mut csv = NamedTempFile::new().unwrap();
    writeln!(csv, "name,age").unwrap();
    csv.flush().unwrap();

    let standard = analyze_csv_file(csv.path(), &CsvParserConfig::default())
        .expect("standard CSV analysis should succeed");
    let reports = [
        ("standard", standard),
        (
            "auto",
            Profiler::new()
                .engine(EngineType::Auto)
                .analyze_file(csv.path())
                .expect("auto analysis should succeed"),
        ),
        (
            "incremental",
            Profiler::new()
                .engine(EngineType::Incremental)
                .analyze_file(csv.path())
                .expect("incremental analysis should succeed"),
        ),
        (
            "columnar",
            Profiler::new()
                .engine(EngineType::Columnar)
                .analyze_file(csv.path())
                .expect("columnar analysis should succeed"),
        ),
    ];

    for (engine, report) in reports {
        assert_eq!(report.execution.rows_processed, 0, "{engine}");
        assert_eq!(
            report
                .column_profiles
                .iter()
                .map(|column| (column.name.as_str(), column.data_type.clone()))
                .collect::<Vec<_>>(),
            vec![("name", DataType::String), ("age", DataType::String)],
            "{engine}"
        );
        assert!(report.quality.is_some(), "{engine}");

        let serialized = serde_json::to_value(&report).expect("report should serialize");
        let serialized_columns = serialized["column_profiles"]
            .as_array()
            .expect("column_profiles should be an array");
        assert_eq!(
            serialized_columns
                .iter()
                .map(|column| column["name"].as_str().expect("column name"))
                .collect::<Vec<_>>(),
            vec!["name", "age"],
            "{engine}"
        );
        assert!(serialized["quality"].is_object(), "{engine}");
    }
}

/// A header-only CSV assesses no dimension on any path. The aggregate must
/// say so rather than averaging the empty set to 0.0, which reads as "this
/// data is terrible" (#571). All four paths carried the same 0.0 before this,
/// so they are pinned together.
#[test]
fn test_unassessable_csv_reports_no_overall_score_across_engines() {
    let mut csv = NamedTempFile::new().unwrap();
    writeln!(csv, "name,age").unwrap();
    csv.flush().unwrap();

    let reports = [
        (
            "standard",
            analyze_csv_file(csv.path(), &CsvParserConfig::default())
                .expect("standard CSV analysis should succeed"),
        ),
        (
            "auto",
            Profiler::new()
                .engine(EngineType::Auto)
                .analyze_file(csv.path())
                .expect("auto analysis should succeed"),
        ),
        (
            "incremental",
            Profiler::new()
                .engine(EngineType::Incremental)
                .analyze_file(csv.path())
                .expect("incremental analysis should succeed"),
        ),
        (
            "columnar",
            Profiler::new()
                .engine(EngineType::Columnar)
                .analyze_file(csv.path())
                .expect("columnar analysis should succeed"),
        ),
    ];

    for (engine, report) in reports {
        // The two report-level aggregates must agree with each other and with
        // the dimension layer, on every path.
        assert_eq!(report.quality_score(), None, "{engine}");
        let quality = report.quality.as_ref().expect("quality present");
        assert!(quality.metrics.assessed_dimensions().is_empty(), "{engine}");
        assert_eq!(quality.score(), None, "{engine}");
        assert_eq!(quality.metrics.overall_score(), None, "{engine}");
        assert!(
            matches!(quality.confidence, MetricConfidence::NotAssessed),
            "{engine}: confidence must not claim exactness about a score that \
             was never computed, got {:?}",
            quality.confidence
        );

        // The serialized form is what saved reports and agent-facing output
        // actually read, so it carries the same answer.
        let serialized = serde_json::to_value(&report).expect("report should serialize");
        assert_eq!(
            serialized["quality"]["confidence"],
            json!("NotAssessed"),
            "{engine}"
        );
    }
}

#[test]
fn test_empty_csv_quality_presence_matches_across_engines_and_serialization() {
    let csv = NamedTempFile::new().unwrap();
    let path = csv.path();

    let reports = [
        (
            "standard",
            analyze_csv_file(path, &CsvParserConfig::default())
                .expect("standard CSV analysis should succeed"),
        ),
        (
            "auto",
            Profiler::new()
                .engine(EngineType::Auto)
                .analyze_file(path)
                .expect("auto analysis should succeed"),
        ),
        (
            "incremental",
            Profiler::new()
                .engine(EngineType::Incremental)
                .analyze_file(path)
                .expect("incremental analysis should succeed"),
        ),
        (
            "columnar",
            Profiler::new()
                .engine(EngineType::Columnar)
                .analyze_file(path)
                .expect("columnar analysis should succeed"),
        ),
    ];

    for (engine, report) in reports {
        assert!(report.column_profiles.is_empty(), "{engine}");
        assert_eq!(report.execution.rows_processed, 0, "{engine}");
        assert_eq!(
            report.quality_status,
            dataprof::QualityAnalysisStatus::Computed,
            "{engine}"
        );
        assert!(report.quality_score().is_none(), "{engine}");
        let quality = report.quality.as_ref().expect("empty input was analyzed");
        assert!(quality.metrics.assessed_dimensions().is_empty(), "{engine}");

        let serialized = serde_json::to_value(&report).expect("report should serialize");
        assert!(serialized["quality"].is_object(), "{engine}");
        assert_eq!(
            serialized["quality_status"],
            json!({"state": "computed"}),
            "{engine}"
        );
        assert_eq!(
            serialized["quality"]["confidence"],
            json!("NotAssessed"),
            "{engine}"
        );
    }
}

/// Numeric inference normalizes surrounding whitespace, so the statistics
/// parser must consume the same lexical value. Previously every engine typed
/// these cells as integers but excluded both from its numeric accumulator,
/// publishing plausible zeroes with `invalid_count == 2`.
#[test]
fn whitespace_padded_numeric_stats_match_across_csv_engines() {
    let mut csv = NamedTempFile::new().unwrap();
    writeln!(csv, "name,amount").unwrap();
    writeln!(csv, "Alice, 1 ").unwrap();
    writeln!(csv, "Bob, 2 ").unwrap();
    csv.flush().unwrap();

    let reports = [
        (
            "standard",
            analyze_csv_file(csv.path(), &CsvParserConfig::default())
                .expect("standard CSV analysis should succeed"),
        ),
        (
            "auto",
            Profiler::new()
                .engine(EngineType::Auto)
                .analyze_file(csv.path())
                .expect("auto analysis should succeed"),
        ),
        (
            "incremental",
            Profiler::new()
                .engine(EngineType::Incremental)
                .analyze_file(csv.path())
                .expect("incremental analysis should succeed"),
        ),
        (
            "columnar",
            Profiler::new()
                .engine(EngineType::Columnar)
                .analyze_file(csv.path())
                .expect("columnar analysis should succeed"),
        ),
    ];

    for (engine, report) in reports {
        let amount = report
            .column_profiles
            .iter()
            .find(|column| column.name == "amount")
            .expect("amount profile");
        assert_eq!(amount.data_type, DataType::Integer, "{engine}");
        assert_eq!(amount.invalid_count, Some(0), "{engine}");
        let ColumnStats::Numeric(stats) = &amount.stats else {
            panic!("{engine} should produce numeric statistics");
        };
        assert_eq!(stats.min, 1.0, "{engine}");
        assert_eq!(stats.max, 2.0, "{engine}");
        assert_eq!(stats.mean, 1.5, "{engine}");
    }
}

/// A configured chunk is allowed to be smaller than a header or record. It is
/// a read target, not a license to profile a fragment as the whole schema.
#[test]
fn incremental_chunk_smaller_than_header_preserves_the_file() {
    let mut csv = NamedTempFile::new().unwrap();
    write!(csv, "alpha,beta\n1,2\n3,4\n").unwrap();
    csv.flush().unwrap();

    let report = Profiler::new()
        .engine(EngineType::Incremental)
        .chunk_size(ChunkSize::Fixed(5))
        .analyze_file(csv.path())
        .expect("incremental analysis should succeed");

    assert_eq!(report.execution.rows_processed, 2);
    assert_eq!(report.execution.ragged_row_count, 0);
    assert!(report.execution.source_exhausted);
    assert_eq!(
        report
            .column_profiles
            .iter()
            .map(|column| column.name.as_str())
            .collect::<Vec<_>>(),
        ["alpha", "beta"]
    );
}

/// The default 64 KiB incremental boundary may land inside a quoted multiline
/// field. Physical newlines inside that field are data, not row boundaries.
#[test]
fn multiline_csv_crossing_default_chunk_boundary_matches_columnar() {
    let mut csv = NamedTempFile::new().unwrap();
    writeln!(csv, "id,bio").unwrap();
    writeln!(csv, "0,{}", "x".repeat(65_490)).unwrap();
    writeln!(csv, "1,\"hello\n{}\"", "w".repeat(100)).unwrap();
    writeln!(csv, "2,end").unwrap();
    csv.flush().unwrap();

    let columnar = Profiler::new()
        .engine(EngineType::Columnar)
        .analyze_file(csv.path())
        .expect("columnar analysis should succeed");

    for engine in [EngineType::Auto, EngineType::Incremental] {
        let report = Profiler::new()
            .engine(engine)
            .analyze_file(csv.path())
            .expect("incremental analysis should succeed");

        assert_eq!(
            report.execution.engine.as_deref(),
            Some("incremental"),
            "{engine:?}"
        );
        assert_eq!(report.execution.rows_processed, 3, "{engine:?}");
        assert_eq!(report.execution.ragged_row_count, 0, "{engine:?}");
        assert_eq!(
            report
                .column_profiles
                .iter()
                .map(|column| (&column.name, &column.data_type, column.total_count))
                .collect::<Vec<_>>(),
            columnar
                .column_profiles
                .iter()
                .map(|column| (&column.name, &column.data_type, column.total_count))
                .collect::<Vec<_>>(),
            "{engine:?}"
        );
    }
}

/// Base numeric statistics must be exact — computed over every value, not the
/// bounded sample — and identical across engines, even when the data is
/// sorted and larger than the sample capacity (#424).
#[test]
fn test_base_numeric_stats_exact_beyond_sample_capacity() {
    let csv = create_sorted_30k_csv();
    let path = csv.path();

    let std_report = analyze_csv_file(path, &CsvParserConfig::default())
        .expect("standard CSV analysis should succeed");
    let arrow_report = Profiler::new()
        .engine(EngineType::Columnar)
        .analyze_file(path)
        .expect("Arrow CSV analysis should succeed");

    for (engine, report) in [("std", &std_report), ("arrow", &arrow_report)] {
        let id = report
            .column_profiles
            .iter()
            .find(|c| c.name == "id")
            .expect("id column");
        let ColumnStats::Numeric(n) = &id.stats else {
            panic!("id should have numeric stats");
        };
        // Exact analytical values for 0..=29999.
        assert_eq!(n.min, 0.0, "[{engine}] min must be exact");
        assert_eq!(n.max, 29_999.0, "[{engine}] max must be exact");
        assert!(
            (n.mean - 14_999.5).abs() < 1e-6,
            "[{engine}] mean must be exact, got {}",
            n.mean
        );
        // Order statistics come from a 10k sample out of 30k values, and the
        // report must say so.
        assert_eq!(
            n.is_approximate,
            Some(true),
            "[{engine}] sampled order statistics must be disclosed as approximate"
        );
        // Every value parsed: analyzed and clean is Some(0), never None —
        // and never a nonzero artifact of comparing against the sample.
        assert_eq!(id.invalid_count, Some(0), "[{engine}] invalid_count");
    }

    // The contract applies at serialized precision. Sample-derived order
    // statistics can differ because each engine retains a different sample;
    // these base statistics consume the same complete population (#547).
    let standard = serde_json::to_value(&std_report.column_profiles).unwrap();
    let arrow = serde_json::to_value(&arrow_report.column_profiles).unwrap();
    assert_eq!(
        std_report.column_profiles.len(),
        arrow_report.column_profiles.len()
    );
    for index in 0..std_report.column_profiles.len() {
        for field in ["min", "max", "mean", "std_dev", "variance"] {
            assert!(standard[index]["stats"]["Numeric"][field].is_number());
            assert!(arrow[index]["stats"]["Numeric"][field].is_number());
            assert_eq!(
                standard[index]["stats"]["Numeric"][field], arrow[index]["stats"]["Numeric"][field],
                "column {index}, serialized {field}"
            );
        }
    }
}

/// A numeric column with an unparseable non-null value must disclose it via
/// `invalid_count` — identically through every engine — so the statistics
/// denominator is auditable instead of silently shrinking (#425).
#[test]
fn test_invalid_count_surfaces_unparseable_numeric_values() {
    let mut f = NamedTempFile::new().unwrap();
    writeln!(f, "amount,label").unwrap();
    for i in 0..5 {
        writeln!(f, "{}.5,row{}", i, i).unwrap();
    }
    // Decimal-comma value: non-null, fails the numeric parse.
    writeln!(f, "\"12,50\",rowx").unwrap();
    writeln!(f, ",rownull").unwrap();
    f.flush().unwrap();

    let std_report = analyze_csv_file(f.path(), &CsvParserConfig::default()).unwrap();
    let arrow_report = Profiler::new()
        .engine(EngineType::Columnar)
        .analyze_file(f.path())
        .unwrap();

    for (engine, report) in [("std", &std_report), ("arrow", &arrow_report)] {
        let amount = report
            .column_profiles
            .iter()
            .find(|c| c.name == "amount")
            .unwrap();
        assert_eq!(amount.data_type, DataType::Float, "[{engine}]");
        assert_eq!(amount.total_count, 7, "[{engine}]");
        assert_eq!(amount.null_count, 1, "[{engine}]");
        assert_eq!(
            amount.invalid_count,
            Some(1),
            "[{engine}] the unparseable value must be counted, not silently dropped"
        );

        let label = report
            .column_profiles
            .iter()
            .find(|c| c.name == "label")
            .unwrap();
        assert_eq!(
            label.invalid_count, None,
            "[{engine}] non-numeric columns are not checked: None, not 0"
        );
    }
}

/// Small data (fewer values than the sample capacity) keeps exact order
/// statistics and must not be flagged approximate.
#[test]
fn test_small_data_not_flagged_approximate() {
    let csv = create_test_csv();
    let report = analyze_csv_file(csv.path(), &CsvParserConfig::default()).unwrap();
    let age = report
        .column_profiles
        .iter()
        .find(|c| c.name == "age")
        .unwrap();
    let ColumnStats::Numeric(n) = &age.stats else {
        panic!("age should be numeric");
    };
    assert_eq!(
        n.is_approximate, None,
        "full-coverage stats are not approximate"
    );
    assert_eq!(n.min, 20.0);
    assert_eq!(n.max, 69.0);
}

fn create_test_csv() -> NamedTempFile {
    let mut f = NamedTempFile::new().unwrap();
    writeln!(f, "name,age,salary,score").unwrap();
    for i in 0..100 {
        writeln!(
            f,
            "Person{},{},{:.2},{:.1}",
            i,
            20 + i % 50,
            30000.0 + i as f64 * 500.0,
            50.0 + (i % 50) as f64
        )
        .unwrap();
    }
    f.flush().unwrap();
    f
}

#[test]
fn test_standard_vs_arrow_csv_numeric_stats() {
    let csv = create_test_csv();
    let path = csv.path();

    // Standard CSV engine
    let std_report = analyze_csv_file(path, &CsvParserConfig::default())
        .expect("standard CSV analysis should succeed");

    // Arrow CSV engine via the unified Profiler API
    let arrow_report = Profiler::new()
        .engine(EngineType::Columnar)
        .analyze_file(path)
        .expect("Arrow CSV analysis should succeed");

    // No tolerance on the contract surface, including optional fields: an
    // absent statistic must not silently compare equal to a measured one.
    assert_eq!(
        serde_json::to_value(&std_report.column_profiles).unwrap(),
        serde_json::to_value(&arrow_report.column_profiles).unwrap(),
        "serialized column profiles must match across CSV engines"
    );

    // Quality scores are rounded metrics on the same contract surface, and are
    // compared apart from the execution provenance that surrounds them.
    assert_eq!(
        serde_json::to_value(&std_report.quality).unwrap(),
        serde_json::to_value(&arrow_report.quality).unwrap(),
        "serialized quality must match across CSV engines"
    );

    // Raw attributes retain full precision. These diagnostics allow only
    // accumulation-order differences, independently of serialized equality.
    for (standard, arrow) in std_report
        .column_profiles
        .iter()
        .zip(&arrow_report.column_profiles)
    {
        if let (ColumnStats::Numeric(a), ColumnStats::Numeric(b)) = (&standard.stats, &arrow.stats)
        {
            for (field, left, right) in [
                ("min", a.min, b.min),
                ("max", a.max, b.max),
                ("mean", a.mean, b.mean),
                (
                    "std_dev",
                    a.std_dev.expect("finite spread"),
                    b.std_dev.expect("finite spread"),
                ),
                (
                    "variance",
                    a.variance.expect("finite spread"),
                    b.variance.expect("finite spread"),
                ),
            ] {
                let tolerance = (1e-9 * left.abs().max(right.abs())).max(1e-12);
                assert!(
                    (left - right).abs() <= tolerance,
                    "{} raw {field}: {left} vs {right}",
                    standard.name
                );
            }
        }
    }
}

#[test]
fn test_precision_quality_matches_standard_and_columnar_csv() {
    let mut csv = NamedTempFile::new().unwrap();
    writeln!(csv, "amount").unwrap();
    for value in ["1.25", "2.35", "3.45", "4.5"] {
        writeln!(csv, "{value}").unwrap();
    }
    csv.flush().unwrap();

    let standard = analyze_csv_file(csv.path(), &CsvParserConfig::default()).unwrap();
    let columnar = Profiler::new()
        .engine(EngineType::Columnar)
        .analyze_file(csv.path())
        .unwrap();
    let standard_precision = standard
        .quality
        .unwrap()
        .metrics
        .precision
        .expect("standard precision");
    let columnar_precision = columnar
        .quality
        .unwrap()
        .metrics
        .precision
        .expect("columnar precision");

    assert_eq!(
        standard_precision.numeric_values_checked,
        columnar_precision.numeric_values_checked
    );
    assert_eq!(
        standard_precision.inconsistent_precision_values,
        columnar_precision.inconsistent_precision_values
    );
    assert!(
        (standard_precision.decimal_places_consistency
            - columnar_precision.decimal_places_consistency)
            .abs()
            < 0.01
    );
}

#[test]
fn test_mixed_data_column_type_consistency() {
    // A column with one non-numeric value should be String in both engines
    let mut f = NamedTempFile::new().unwrap();
    writeln!(f, "id,value,date_col").unwrap();
    writeln!(f, "1,100,2024-01-01").unwrap();
    writeln!(f, "2,200,2024-01-02").unwrap();
    writeln!(f, "3,N/A,2024-01-03").unwrap();
    writeln!(f, "4,400,2024-01-04").unwrap();
    writeln!(f, "5,500,2024-01-05").unwrap();
    f.flush().unwrap();

    let std_report = analyze_csv_file(f.path(), &CsvParserConfig::default())
        .expect("standard CSV should succeed");
    let arrow_report = Profiler::new()
        .engine(EngineType::Columnar)
        .analyze_file(f.path())
        .expect("Arrow CSV should succeed");

    for std_col in &std_report.column_profiles {
        let arrow_col = arrow_report
            .column_profiles
            .iter()
            .find(|c| c.name == std_col.name)
            .unwrap_or_else(|| panic!("Column '{}' missing from Arrow report", std_col.name));

        assert_eq!(
            std_col.data_type, arrow_col.data_type,
            "Type mismatch for column '{}': std={:?}, arrow={:?}",
            std_col.name, std_col.data_type, arrow_col.data_type
        );
    }
}

#[test]
fn test_problematic_date_columns_stay_consistent_across_engines() {
    let root = Path::new(env!("CARGO_MANIFEST_DIR")).join("examples/test_datasets");

    for (file_name, column_name) in [
        ("sales_data_problematic.csv", "order_date"),
        ("sensor_data_outliers.csv", "timestamp"),
    ] {
        let path = root.join(file_name);

        let std_report =
            analyze_csv_file(&path, &CsvParserConfig::default()).unwrap_or_else(|err| {
                panic!("standard CSV analysis failed for {}: {}", file_name, err)
            });
        let arrow_report = Profiler::new()
            .engine(EngineType::Columnar)
            .analyze_file(&path)
            .unwrap_or_else(|err| panic!("Arrow CSV analysis failed for {}: {}", file_name, err));

        let std_col = std_report
            .column_profiles
            .iter()
            .find(|c| c.name == column_name)
            .unwrap_or_else(|| {
                panic!(
                    "Column '{}' missing from standard report for {}",
                    column_name, file_name
                )
            });
        let arrow_col = arrow_report
            .column_profiles
            .iter()
            .find(|c| c.name == column_name)
            .unwrap_or_else(|| {
                panic!(
                    "Column '{}' missing from Arrow report for {}",
                    column_name, file_name
                )
            });

        assert_eq!(std_col.data_type, DataType::Date);
        assert_eq!(
            std_col.data_type, arrow_col.data_type,
            "Type mismatch for '{}' in {}: std={:?}, arrow={:?}",
            column_name, file_name, std_col.data_type, arrow_col.data_type
        );
    }
}

/// Test that batch (ArrowProfiler) produces `MetricConfidence::Exact` and that
/// streaming (IncrementalProfiler) detects nulls correctly for a small dataset.
///
/// For small datasets where sample == total rows, both engines use the uniform
/// (non-bifurcated) path. The structural Mixed-confidence test is in
/// `test_streaming_bifurcation_with_large_dataset`.
#[test]
fn test_streaming_vs_batch_quality_confidence() {
    // CSV with known null pattern: empty fields in name/value columns
    let mut f = NamedTempFile::new().unwrap();
    writeln!(f, "id,name,value").unwrap();
    for i in 0..200 {
        let name = if i % 10 == 0 { "" } else { "Alice" }; // 10% nulls in name
        let value = if i % 20 == 0 {
            ""
        } else {
            &format!("{}", i * 100)
        }; // 5% nulls in value
        writeln!(f, "{},{},{}", i, name, value).unwrap();
    }
    f.flush().unwrap();

    // Batch engine (ArrowProfiler) — full data, exact confidence
    let batch_report = Profiler::new()
        .engine(EngineType::Columnar)
        .analyze_file(f.path())
        .expect("Batch analysis should succeed");

    // Streaming engine (IncrementalProfiler)
    let streaming_report = Profiler::new()
        .engine(EngineType::Incremental)
        .analyze_file(f.path())
        .expect("Streaming analysis should succeed");

    // Both should have quality assessments
    let batch_quality = batch_report
        .quality
        .as_ref()
        .expect("Batch report should have quality");
    let streaming_quality = streaming_report
        .quality
        .as_ref()
        .expect("Streaming report should have quality");

    // Batch engine should produce Exact confidence
    assert!(
        matches!(batch_quality.confidence, MetricConfidence::Exact),
        "Batch engine should produce Exact confidence, got {:?}",
        batch_quality.confidence
    );

    // For 200 rows (< 10K reservoir), the streaming engine has sample == total.
    // The bifurcated path only triggers when sample < total, so both engines
    // use the uniform path here. The structural bifurcation test with Mixed
    // confidence is in test_streaming_bifurcation_with_large_dataset.

    // Streaming engine should detect the known null pattern in ColumnProfile.
    // The streaming null_count tracks empty strings as nulls, giving us exact
    // completeness metrics from global counters.
    let streaming_m = &streaming_quality.metrics;
    assert!(
        streaming_m.missing_values_ratio() > 0.0,
        "Streaming engine should detect missing values from empty CSV fields"
    );

    // key_uniqueness should be close (HLL ≤ 3% error for 200 rows)
    let batch_m = &batch_quality.metrics;
    assert!(
        (batch_m.key_uniqueness() - streaming_m.key_uniqueness()).abs() < 5.0,
        "key_uniqueness: batch={:.2} vs streaming={:.2}",
        batch_m.key_uniqueness(),
        streaming_m.key_uniqueness()
    );
}

/// Test bifurcation with a larger dataset where sample < total rows.
/// This forces the streaming engine to produce Mixed confidence.
#[test]
fn test_streaming_bifurcation_with_large_dataset() {
    let mut f = NamedTempFile::new().unwrap();
    writeln!(f, "id,category,amount").unwrap();
    // Write enough rows that reservoir sample (10K) < total rows
    // This triggers the bifurcated path in ReportAssembler
    for i in 0..15_000 {
        let cat = if i % 3 == 0 {
            "A"
        } else if i % 3 == 1 {
            "B"
        } else {
            "C"
        };
        let amount = if i % 100 == 0 {
            ""
        } else {
            &format!("{:.2}", i as f64 * 1.5)
        };
        writeln!(f, "{},{},{}", i, cat, amount).unwrap();
    }
    f.flush().unwrap();

    let report = Profiler::new()
        .engine(EngineType::Incremental)
        .analyze_file(f.path())
        .expect("Large streaming analysis should succeed");

    let quality = report
        .quality
        .as_ref()
        .expect("Should have quality assessment");

    // With 15K rows and 10K reservoir, sample < total → Mixed confidence
    match &quality.confidence {
        MetricConfidence::Mixed {
            exact_dimensions,
            sampled_dimensions,
            sample_size,
        } => {
            assert!(
                exact_dimensions.contains(&"completeness".to_string()),
                "completeness should be exact"
            );
            assert!(
                exact_dimensions.contains(&"key_uniqueness".to_string()),
                "key_uniqueness should be exact"
            );
            assert!(
                sampled_dimensions.contains(&"consistency".to_string()),
                "consistency should be sampled"
            );
            assert!(
                sampled_dimensions.contains(&"accuracy".to_string()),
                "accuracy should be sampled"
            );
            assert!(
                *sample_size < 15_000,
                "sample_size ({}) should be less than total rows (15000)",
                sample_size
            );
        }
        other => panic!(
            "Expected Mixed confidence for large streaming dataset, got {:?}",
            other
        ),
    }

    // Completeness should reflect the exact global counters
    // ~1% nulls in amount column (every 100th row), 0 nulls in id/category
    let m = &quality.metrics;
    assert!(
        m.missing_values_ratio() > 0.0,
        "Should detect some missing values"
    );
    assert!(
        m.missing_values_ratio() < 2.0,
        "Missing ratio should be small (~0.33%)"
    );
}

// -- Distinct-count provenance (#383) --

/// On a small dataset every column's distinct count is exact, and both engines
/// must say so: `unique_count_is_approximate == Some(false)`, never `None` or
/// `Some(true)`. An exact-looking count with no provenance is unsafe for key
/// checks, so the flag must be present and truthful.
#[test]
fn test_distinct_count_marked_exact_on_small_data_across_engines() {
    let csv = create_test_csv(); // 100 rows, well under the estimator threshold
    let path = csv.path();

    let std_report = analyze_csv_file(path, &CsvParserConfig::default()).unwrap();
    let arrow_report = Profiler::new()
        .engine(EngineType::Columnar)
        .analyze_file(path)
        .unwrap();

    for std_col in &std_report.column_profiles {
        let arrow_col = arrow_report
            .column_profiles
            .iter()
            .find(|c| c.name == std_col.name)
            .unwrap();

        assert_eq!(
            std_col.unique_count_is_approximate,
            Some(false),
            "standard engine should mark '{}' exact",
            std_col.name
        );
        assert_eq!(
            arrow_col.unique_count_is_approximate,
            Some(false),
            "columnar engine should mark '{}' exact",
            std_col.name
        );
        assert_eq!(
            std_col.unique_count, arrow_col.unique_count,
            "exact unique_count should match across engines for '{}'",
            std_col.name
        );
    }
}

/// A table-sized key column is counted exactly, and every engine agrees.
///
/// Distinct counts used to switch to a HyperLogLog estimate past 10,000
/// distinct values, so 20,000 unique ids were an approximation on every engine
/// (and on the incremental engine even a two-value column past its reservoir).
/// The exact regime now holds fingerprints and reaches a million distinct
/// values; the sketch past that is covered by the estimator's own tests.
#[test]
fn test_table_sized_cardinality_is_exact_across_engines() {
    let mut f = NamedTempFile::new().unwrap();
    writeln!(f, "id,category").unwrap();
    let rows = 20_000;
    for i in 0..rows {
        writeln!(f, "id-{i},{}", if i % 2 == 0 { "A" } else { "B" }).unwrap();
    }
    f.flush().unwrap();

    let std_report = analyze_csv_file(f.path(), &CsvParserConfig::default()).unwrap();
    let incremental_report = Profiler::new()
        .engine(EngineType::Incremental)
        .analyze_file(f.path())
        .unwrap();
    let arrow_report = Profiler::new()
        .engine(EngineType::Columnar)
        .analyze_file(f.path())
        .unwrap();

    for report in [&std_report, &incremental_report, &arrow_report] {
        for (name, distinct) in [("id", rows), ("category", 2)] {
            let column = report
                .column_profiles
                .iter()
                .find(|c| c.name == name)
                .expect("column present");
            assert_eq!(column.unique_count_is_approximate, Some(false), "{name}");
            assert_eq!(column.unique_count, Some(distinct), "{name}");
        }
    }
}

// -- Selective dimension computation --

#[test]
fn test_profiler_selective_dimensions_only_completeness() {
    let csv = create_test_csv();
    let report = Profiler::new()
        .quality_dimensions(vec![QualityDimension::Completeness])
        .analyze_file(csv.path())
        .unwrap();

    let quality = report.quality.expect("quality should be present");
    let m = &quality.metrics;

    assert!(m.completeness.is_some(), "completeness should be computed");
    assert!(m.consistency.is_none(), "consistency should be skipped");
    assert!(m.uniqueness.is_none(), "uniqueness should be skipped");
    assert!(m.accuracy.is_none(), "accuracy should be skipped");
    assert!(m.timeliness.is_none(), "timeliness should be skipped");
    assert!(m.validity.is_none(), "validity should be skipped");
    assert!(m.precision.is_none(), "precision should be skipped");

    // Score should re-normalize to completeness alone
    let score = m.overall_score().expect("completeness was assessed");
    let completeness_score = m.completeness.as_ref().unwrap().complete_records_ratio;
    assert!(
        (score - completeness_score).abs() < 0.01,
        "score {score} should equal completeness {completeness_score}"
    );
}

#[test]
fn test_profiler_selective_dimensions_subset() {
    let csv = create_test_csv();
    let report = Profiler::new()
        .quality_dimensions(vec![
            QualityDimension::Completeness,
            QualityDimension::Uniqueness,
        ])
        .analyze_file(csv.path())
        .unwrap();

    let quality = report.quality.expect("quality should be present");
    let m = &quality.metrics;

    assert!(m.completeness.is_some());
    assert!(m.consistency.is_none());
    assert!(m.uniqueness.is_some());
    assert!(m.accuracy.is_none());
    assert!(m.timeliness.is_none());
    assert!(m.validity.is_none());
    assert!(m.precision.is_none());
}

#[test]
fn test_profiler_all_dimensions_default() {
    let csv = create_test_csv();
    // No quality_dimensions() call → all dimensions computed
    let report = Profiler::new().analyze_file(csv.path()).unwrap();

    let quality = report.quality.expect("quality should be present");
    let m = &quality.metrics;

    assert!(m.completeness.is_some());
    assert!(m.consistency.is_some());
    assert!(m.uniqueness.is_some());
    assert!(m.accuracy.is_some());
    assert!(m.timeliness.is_some());
    assert!(m.validity.is_some());
    assert!(m.precision.is_some());
}

/// Regression for #417: the columnar engine used to skip duplicate-row
/// detection whenever any column contained nulls (misaligned sample
/// reservoirs), silently dropping the duplicate component of the uniqueness
/// dimension and producing a different overall quality score than the
/// incremental engine for the same bytes.
#[test]
fn test_duplicate_rows_with_nulls_match_across_engines() {
    let mut f = NamedTempFile::new().unwrap();
    writeln!(f, "id,name,value,city").unwrap();
    writeln!(f, "1,Alice,10.5,Rome").unwrap();
    writeln!(f, "2,,20.0,Milan").unwrap();
    writeln!(f, "2,,20.0,Milan").unwrap(); // duplicate row containing a null
    writeln!(f, "3,Bob,,Naples").unwrap();
    writeln!(f, "4,Alice,10.5,").unwrap();
    writeln!(f, "1,Alice,10.5,Rome").unwrap(); // duplicate of row 1
    f.flush().unwrap();

    let incremental = Profiler::new()
        .engine(EngineType::Incremental)
        .analyze_file(f.path())
        .expect("incremental analysis should succeed");
    let columnar = Profiler::new()
        .engine(EngineType::Columnar)
        .analyze_file(f.path())
        .expect("columnar analysis should succeed");

    let inc_uniq = incremental
        .quality
        .as_ref()
        .and_then(|q| q.metrics.uniqueness.as_ref())
        .expect("incremental uniqueness should be assessed");
    let col_uniq = columnar
        .quality
        .as_ref()
        .and_then(|q| q.metrics.uniqueness.as_ref())
        .expect("columnar uniqueness should be assessed");

    // Both engines must scan every row and find both duplicates, despite the
    // nulls that break per-column sample alignment.
    assert_eq!(inc_uniq.rows_checked, 6, "incremental rows_checked");
    assert_eq!(col_uniq.rows_checked, 6, "columnar rows_checked");
    assert_eq!(inc_uniq.duplicate_rows, 2, "incremental duplicate_rows");
    assert_eq!(col_uniq.duplicate_rows, 2, "columnar duplicate_rows");
    assert_eq!(
        inc_uniq.key_uniqueness, col_uniq.key_uniqueness,
        "key_uniqueness must match across engines"
    );

    // The contract that actually matters to users: identical overall score.
    let inc_score = incremental.quality_score().expect("incremental score");
    let col_score = columnar.quality_score().expect("columnar score");
    assert!(
        (inc_score - col_score).abs() < 0.01,
        "overall quality must not depend on the engine: incremental={inc_score} columnar={col_score}"
    );
}

/// Duplicate-row detection must also run on clean files (no nulls) — the
/// aligned-sample fallback already handled this, so it guards the new
/// full-stream tracker against regressing the easy case.
#[test]
fn test_duplicate_rows_without_nulls_match_across_engines() {
    let mut f = NamedTempFile::new().unwrap();
    writeln!(f, "id,name").unwrap();
    writeln!(f, "1,Alice").unwrap();
    writeln!(f, "2,Bob").unwrap();
    writeln!(f, "2,Bob").unwrap();
    writeln!(f, "3,Carol").unwrap();
    f.flush().unwrap();

    for engine in [EngineType::Incremental, EngineType::Columnar] {
        let report = Profiler::new()
            .engine(engine)
            .analyze_file(f.path())
            .expect("analysis should succeed");
        let uniq = report
            .quality
            .as_ref()
            .and_then(|q| q.metrics.uniqueness.as_ref())
            .expect("uniqueness should be assessed");
        assert_eq!(uniq.rows_checked, 4, "{engine:?} rows_checked");
        assert_eq!(uniq.duplicate_rows, 1, "{engine:?} duplicate_rows");
    }
}

/// Duplicate column names are a deterministic file property that no engine can
/// resolve; every CSV engine must reject them before profiling rather than
/// silently merging two columns into one profile (#381).
#[test]
fn test_duplicate_headers_rejected_across_csv_engines() {
    let mut f = NamedTempFile::new().unwrap();
    writeln!(f, "x,x,y").unwrap();
    writeln!(f, "1,2,a").unwrap();
    writeln!(f, "3,4,b").unwrap();
    f.flush().unwrap();

    // Standard (simple) CSV engine.
    let err = analyze_csv_file(f.path(), &CsvParserConfig::default())
        .expect_err("standard engine must reject duplicate headers");
    assert_eq!(err.category(), "duplicate_column_name");
    let msg = err.to_string();
    assert!(msg.contains("'x'"), "names the offender: {msg}");
    assert!(!msg.contains('1'), "must not echo cell values: {msg}");

    // Auto (incremental) and Columnar (Arrow) engines: a clear, categorized error,
    // never buried under "All engines failed".
    for engine in [
        EngineType::Auto,
        EngineType::Incremental,
        EngineType::Columnar,
    ] {
        let err = Profiler::new()
            .engine(engine)
            .analyze_file(f.path())
            .expect_err("engine must reject duplicate headers");
        assert_eq!(
            err.category(),
            "duplicate_column_name",
            "{engine:?} should surface the duplicate-column error directly"
        );
    }
}

/// The reason duplicates must be rejected: a rectangular source must yield one
/// profile per column, each with `total_count == rows` — never a merged column
/// whose count exceeds the row count.
#[test]
fn test_unique_headers_preserve_column_count_and_totals() {
    let mut f = NamedTempFile::new().unwrap();
    writeln!(f, "a,b,c").unwrap();
    writeln!(f, "1,2,3").unwrap();
    writeln!(f, "4,5,6").unwrap();
    f.flush().unwrap();

    for engine in [EngineType::Incremental, EngineType::Columnar] {
        let report = Profiler::new()
            .engine(engine)
            .analyze_file(f.path())
            .expect("unique headers should profile");
        assert_eq!(report.column_profiles.len(), 3, "{engine:?} column count");
        for col in &report.column_profiles {
            assert_eq!(col.total_count, 2, "{engine:?} {} total_count", col.name);
        }
    }
}

/// A ragged row must read the same through either engine (#470).
///
/// The columnar engine used to pad a short row to null and report a clean
/// parse, while rejecting an over-long one outright — so `engine="columnar"`
/// was both the only silently-clean CSV path and the only one that refused
/// data the default engine handled. Both directions now recover and count.
#[test]
fn ragged_rows_report_identically_across_engines() {
    let mut f = NamedTempFile::new().unwrap();
    writeln!(f, "name,age,city").unwrap();
    writeln!(f, "Alice,25,Rome").unwrap();
    writeln!(f, "Bob,30").unwrap(); // short: two fields
    writeln!(f, "Carol,35,LA,EXTRA").unwrap(); // over-long: four fields
    writeln!(f, "Dave,40,SF").unwrap();
    f.flush().unwrap();

    let incremental = Profiler::new()
        .engine(EngineType::Incremental)
        .analyze_file(f.path())
        .expect("incremental recovers ragged rows");
    let columnar = Profiler::new()
        .engine(EngineType::Columnar)
        .analyze_file(f.path())
        .expect("columnar recovers ragged rows");

    assert_eq!(incremental.execution.rows_processed, 4);
    assert_eq!(
        columnar.execution.rows_processed, incremental.execution.rows_processed,
        "both engines must see every row"
    );
    assert_eq!(incremental.execution.ragged_row_count, 2);
    assert_eq!(
        columnar.execution.ragged_row_count, incremental.execution.ragged_row_count,
        "the columnar engine must not report a repaired scan as a clean one"
    );
    assert_eq!(
        columnar.column_profiles.len(),
        3,
        "the surplus field must not become a fourth column"
    );
}

/// Recovery truncates an over-long row to header width, so two rows differing
/// only past the header are the same row. The fields the columnar engine has
/// to decode somewhere must not leak into duplicate tracking (#470).
#[test]
fn surplus_fields_do_not_leak_into_duplicate_detection() {
    let mut f = NamedTempFile::new().unwrap();
    writeln!(f, "name,age,city").unwrap();
    writeln!(f, "Alice,25,Rome").unwrap();
    writeln!(f, "Alice,25,Rome,X").unwrap();
    f.flush().unwrap();

    for engine in [EngineType::Incremental, EngineType::Columnar] {
        let report = Profiler::new()
            .engine(engine)
            .analyze_file(f.path())
            .expect("ragged rows profile");
        let uniqueness = report
            .quality
            .as_ref()
            .and_then(|q| q.metrics.uniqueness.as_ref())
            .unwrap_or_else(|| panic!("{engine:?} uniqueness should be assessed"));

        assert_eq!(
            uniqueness.duplicate_rows, 1,
            "{engine:?} must see the truncated row as a duplicate"
        );
    }
}

/// A column that defeated type inference is `String`, the same type a column of
/// names gets, and `invalid_count` is absent on string columns by contract. The
/// evidence that tells the two apart is `type_homogeneity`, and every engine has
/// to report the same counts for the same input or an agent's reading of a
/// column depends on which engine happened to profile it (#561).
#[test]
fn type_homogeneity_agrees_across_engines() {
    let mut f = NamedTempFile::new().unwrap();
    writeln!(f, "amount,name,when").unwrap();
    for i in 0..200 {
        // amount: 150 numbers and 50 placeholders — under the 80% threshold, so
        // inference gives up and the column is typed `string`.
        let amount = if i % 4 == 0 {
            "pending".to_string()
        } else {
            format!("{}.50", 100 + i)
        };
        writeln!(f, "{amount},person{i},2024-01-15").unwrap();
    }
    f.flush().unwrap();

    let standard = analyze_csv_file(f.path(), &CsvParserConfig::default())
        .expect("standard CSV analysis should succeed");
    let reports = [
        ("standard", standard),
        (
            "auto",
            Profiler::new()
                .engine(EngineType::Auto)
                .analyze_file(f.path())
                .expect("auto analysis should succeed"),
        ),
        (
            "incremental",
            Profiler::new()
                .engine(EngineType::Incremental)
                .analyze_file(f.path())
                .expect("incremental analysis should succeed"),
        ),
        (
            "columnar",
            Profiler::new()
                .engine(EngineType::Columnar)
                .analyze_file(f.path())
                .expect("columnar analysis should succeed"),
        ),
    ];

    for (engine, report) in reports {
        let counts = |name: &str| {
            let column = report
                .column_profiles
                .iter()
                .find(|column| column.name == name)
                .unwrap_or_else(|| panic!("[{engine}] {name} column"));
            (
                column.data_type.clone(),
                column
                    .type_homogeneity
                    .unwrap_or_else(|| panic!("[{engine}] {name} must be classified")),
            )
        };

        let (amount_type, amount) = counts("amount");
        assert_eq!(
            amount_type,
            DataType::String,
            "[{engine}] the fixture must reproduce the inference fallback"
        );
        assert_eq!(amount.numeric, 150, "[{engine}] amount numeric");
        assert_eq!(amount.text, 50, "[{engine}] amount text");
        assert_eq!(amount.dominant_share(), Some(0.75), "[{engine}] amount");

        // A genuinely textual column is the control: same `String` type, one
        // class, no mixture to report.
        let (name_type, name) = counts("name");
        assert_eq!(name_type, DataType::String, "[{engine}] name");
        assert_eq!(name.text, 200, "[{engine}] name text");
        assert_eq!(name.dominant_share(), Some(1.0), "[{engine}] name");

        // Dates are their own class, not text that happens to look structured.
        let (_, when) = counts("when");
        assert_eq!(when.date, 200, "[{engine}] when date");

        // And it survives serialization with the counts intact.
        let serialized = serde_json::to_value(&report).expect("report should serialize");
        assert_eq!(
            serialized["column_profiles"][0]["type_homogeneity"],
            json!({"numeric": 150, "date": 0, "boolean": 0, "text": 50}),
            "[{engine}] serialized amount counts"
        );
    }
}

/// A ragged row past `max_rows` is not part of the profile, so no engine may
/// fail over one. The columnar engine used to: it decoded a whole batch before
/// applying the cap, and `arrow-csv` refuses a record wider than its schema
/// (#753). Strict parsing is the deliberate exception, pinned below.
#[test]
fn a_ragged_row_past_the_row_cap_is_ignored_by_every_engine() {
    for row_past_cap in ["Bob,30", "Bob,30,Milan,extra"] {
        let mut csv = NamedTempFile::new().unwrap();
        writeln!(csv, "name,age,city").unwrap();
        writeln!(csv, "Alice,25,Rome").unwrap();
        writeln!(csv, "{row_past_cap}").unwrap();
        csv.flush().unwrap();

        for engine in [
            EngineType::Auto,
            EngineType::Incremental,
            EngineType::Columnar,
        ] {
            let report = Profiler::new()
                .engine(engine)
                .csv_flexible(true)
                .stop_when(StopCondition::MaxRows(1))
                .analyze_file(csv.path())
                .unwrap_or_else(|error| {
                    panic!("{engine:?} failed over a row past the cap ({row_past_cap:?}): {error}")
                });

            assert_eq!(report.execution.rows_processed, 1, "{engine:?}");
            assert_eq!(report.execution.ragged_row_count, 0, "{engine:?}");
            assert!(
                matches!(
                    report.execution.truncation_reason,
                    Some(TruncationReason::MaxRows(1))
                ),
                "{engine:?}: {:?}",
                report.execution.truncation_reason
            );
        }
    }
}

/// Strict parsing rejects a ragged row past the cap on every engine, because
/// each one reads past the cap before applying it: the incremental engine
/// parses a whole chunk, the columnar engine a whole pre-scan record. Whether
/// that is the right contract is #757; until it is decided, the three must not
/// drift apart, which is what this pins.
#[test]
fn strict_parsing_rejects_a_ragged_row_past_the_row_cap_on_every_engine() {
    for row_past_cap in ["Bob,30", "Bob,30,Milan,extra"] {
        let mut csv = NamedTempFile::new().unwrap();
        writeln!(csv, "name,age,city").unwrap();
        writeln!(csv, "Alice,25,Rome").unwrap();
        writeln!(csv, "{row_past_cap}").unwrap();
        csv.flush().unwrap();

        for engine in [
            EngineType::Auto,
            EngineType::Incremental,
            EngineType::Columnar,
        ] {
            let error = Profiler::new()
                .engine(engine)
                .csv_flexible(false)
                .stop_when(StopCondition::MaxRows(1))
                .analyze_file(csv.path())
                .expect_err(&format!("{engine:?} must reject {row_past_cap:?}"));

            // Every engine names both field counts, the `csv` crate's phrasing
            // rather than Arrow's opaque "incorrect number of fields".
            let message = error.to_string();
            assert!(message.contains("fields"), "{engine:?}: {message}");
            assert!(
                !message.contains("incorrect number of fields"),
                "{engine:?}: {message}"
            );
        }
    }
}

/// A cap the file never reaches is not a truncation, and neither is one it
/// reaches exactly. The columnar engine reads one record past the cap to tell
/// the two apart, so the case where there is no such record to read, twice
/// over, has to hold as well.
#[test]
fn a_cap_at_or_beyond_the_file_length_is_not_a_truncation() {
    let mut csv = NamedTempFile::new().unwrap();
    writeln!(csv, "name,age,city").unwrap();
    writeln!(csv, "Alice,25,Rome").unwrap();
    writeln!(csv, "Bob,30,Milan").unwrap();
    csv.flush().unwrap();

    for cap in [2u64, 5] {
        for engine in [
            EngineType::Auto,
            EngineType::Incremental,
            EngineType::Columnar,
        ] {
            let report = Profiler::new()
                .engine(engine)
                .stop_when(StopCondition::MaxRows(cap))
                .analyze_file(csv.path())
                .expect("a cap the file does not exceed should profile the file");

            assert_eq!(report.execution.rows_processed, 2, "{engine:?} cap {cap}");
            assert!(
                report.execution.truncation_reason.is_none(),
                "{engine:?} cap {cap}: {:?}",
                report.execution.truncation_reason
            );
        }
    }
}