tecton-compute 0.1.1

//! Edge-case and boundary integration tests for tecton-compute.

use std::io::Write;
use std::path::PathBuf;
use tecton_compute::{execute_query, ComputeError, SELECT_STAR_DEFAULT_LIMIT};

fn write_csv(path: &std::path::Path, header: &str, rows: impl IntoIterator<Item = String>) {
    let mut f = std::fs::File::create(path).expect("create csv");
    writeln!(f, "{header}").unwrap();
    for row in rows {
        writeln!(f, "{row}").unwrap();
    }
}

#[tokio::test]
async fn invalid_sql_returns_clear_error() {
    let dir = tempfile::tempdir().unwrap();
    let csv = dir.path().join("ok.csv");
    write_csv(&csv, "id,name", ["1,alpha".into(), "2,beta".into()]);

    let err = execute_query(csv.to_str().unwrap(), "SELECT id FORM data")
        .await
        .expect_err("invalid SQL must fail");

    let msg = err.to_string().to_ascii_lowercase();
    assert!(
        msg.contains("datafusion")
            || msg.contains("syntax")
            || msg.contains("error")
            || msg.contains("form"),
        "unexpected error message: {err}"
    );
}

#[tokio::test]
async fn missing_file_is_handled_gracefully() {
    let missing = PathBuf::from("definitely_missing_tecton_dataset_12345.csv");
    let err = execute_query(missing.to_str().unwrap(), "SELECT 1")
        .await
        .expect_err("missing file must fail");

    match err {
        ComputeError::FileNotFound(path) => {
            assert!(path.contains("definitely_missing_tecton_dataset_12345.csv"));
        }
        other => panic!("expected FileNotFound, got: {other}"),
    }
}

#[tokio::test]
async fn complex_join_and_aggregations() {
    let dir = tempfile::tempdir().unwrap();
    let customers = dir.path().join("customers.csv");
    let orders = dir.path().join("orders.csv");

    write_csv(
        &customers,
        "customer_id,gender,region",
        [
            "1,Female,EU".into(),
            "2,Male,US".into(),
            "3,Female,US".into(),
            "4,Non-binary,EU".into(),
        ],
    );
    write_csv(
        &orders,
        "order_id,customer_id,amount",
        [
            "10,1,100".into(),
            "11,1,40".into(),
            "12,2,20".into(),
            "13,3,80".into(),
            "14,3,10".into(),
            "15,4,5".into(),
        ],
    );

    // Multi-table JOIN via DataFusion SessionContext (execute_query is single-table).
    use datafusion::prelude::*;
    let ctx = SessionContext::new();
    ctx.register_csv(
        "customers",
        customers.to_str().unwrap(),
        CsvReadOptions::new(),
    )
    .await
    .unwrap();
    ctx.register_csv("orders", orders.to_str().unwrap(), CsvReadOptions::new())
        .await
        .unwrap();

    let sql = r#"
        SELECT
            c.gender,
            c.region,
            COUNT(*) AS order_count,
            SUM(o.amount) AS total_amount,
            AVG(o.amount) AS avg_amount
        FROM customers c
        INNER JOIN orders o ON c.customer_id = o.customer_id
        WHERE o.amount >= 5
        GROUP BY c.gender, c.region
        HAVING COUNT(*) >= 1
        ORDER BY total_amount DESC
    "#;

    let batches = ctx.sql(sql).await.unwrap().collect().await.unwrap();
    assert!(!batches.is_empty(), "JOIN aggregation should return batches");
    let rows: usize = batches.iter().map(|b| b.num_rows()).sum();
    assert!(rows >= 3, "expected aggregated groups, got {rows}");
}

#[tokio::test]
async fn streaming_boundary_returns_exactly_10k_rows() {
    let dir = tempfile::tempdir().unwrap();
    let csv = dir.path().join("ten_k.csv");
    {
        let mut f = std::fs::File::create(&csv).unwrap();
        writeln!(f, "id,label").unwrap();
        for i in 0..12_000 {
            writeln!(f, "{i},row_{i}").unwrap();
        }
    }

    // Explicit LIMIT at the streaming hard-cap (10_000).
    let result = execute_query(
        csv.to_str().unwrap(),
        "SELECT id, label FROM data ORDER BY id LIMIT 10000",
    )
    .await
    .expect("boundary query");

    assert_eq!(
        result.data.len(),
        10_000,
        "expected exactly 10k streamed rows"
    );
    assert!(result.execution_time_ms < 120_000);
}

#[tokio::test]
async fn select_star_safety_still_caps_below_stream_limit() {
    let dir = tempfile::tempdir().unwrap();
    let csv = dir.path().join("many.csv");
    {
        let mut f = std::fs::File::create(&csv).unwrap();
        writeln!(f, "id").unwrap();
        for i in 0..5_000 {
            writeln!(f, "{i}").unwrap();
        }
    }

    let result = execute_query(csv.to_str().unwrap(), "SELECT * FROM data")
        .await
        .unwrap();

    assert!(result.data.len() <= SELECT_STAR_DEFAULT_LIMIT);
    assert!(
        result
            .optimization_applied
            .as_deref()
            .unwrap_or("")
            .contains("LIMIT"),
        "expected LIMIT optimization note"
    );
}

#[tokio::test]
async fn mutating_sql_is_rejected() {
    let dir = tempfile::tempdir().unwrap();
    let csv = dir.path().join("ok.csv");
    write_csv(&csv, "id", ["1".into()]);

    let err = execute_query(csv.to_str().unwrap(), "DELETE FROM data")
        .await
        .expect_err("mutations must be rejected");
    assert!(err.to_string().to_ascii_lowercase().contains("not allowed"));
}