aedb 0.3.1

use super::execute_query_with_options;
use super::tests::setup;
use crate::catalog::schema::ColumnDef;
use crate::catalog::types::{ColumnType, Row, Value};
use crate::query::error::QueryError;
use crate::query::plan::{Aggregate, Order, Query, QueryOptions, col, lit};

#[test]
fn bounded_scan_is_enforced_when_full_scan_not_allowed() {
    let (keyspace, catalog) = setup();
    let snapshot = keyspace.snapshot();
    let err = execute_query_with_options(
        &snapshot,
        &catalog,
        "A",
        "app",
        Query::select(&["*"]).from("users"),
        &QueryOptions::default(),
        1,
        10_000,
        None,
    )
    .expect_err("should reject full scan");
    assert!(matches!(err, QueryError::InvalidQuery { .. }));
}

#[test]
fn default_execute_query_rejects_unbounded_full_scan() {
    let (keyspace, catalog) = setup();
    let snapshot = keyspace.snapshot();
    let err = super::execute_query(
        &snapshot,
        &catalog,
        "A",
        "app",
        Query::select(&["*"]).from("users"),
    )
    .expect_err("default execute_query should reject unbounded full scan");
    assert!(matches!(err, QueryError::InvalidQuery { .. }));
}

#[test]
fn non_join_page_size_is_capped_by_max_scan_rows() {
    let (keyspace, catalog) = setup();
    let snapshot = keyspace.snapshot();
    let result = execute_query_with_options(
        &snapshot,
        &catalog,
        "A",
        "app",
        Query::select(&["*"])
            .from("users")
            .order_by("id", Order::Asc)
            .limit(50),
        &QueryOptions::default(),
        9,
        10,
        None,
    )
    .expect("bounded page");
    assert_eq!(result.rows.len(), 10);
    assert!(result.cursor.is_some());
    assert!(result.rows_examined <= 100);
}

#[test]
fn aggregate_limit_does_not_bypass_scan_bound() {
    let (keyspace, catalog) = setup();
    let snapshot = keyspace.snapshot();
    let err = execute_query_with_options(
        &snapshot,
        &catalog,
        "A",
        "app",
        Query::select(&["*"])
            .from("users")
            .aggregate(Aggregate::Count)
            .limit(1),
        &QueryOptions::default(),
        7,
        10,
        None,
    )
    .expect_err("aggregate should still honor scan bound");
    assert!(matches!(err, QueryError::ScanBoundExceeded { .. }));
}

#[test]
fn cursor_does_not_bypass_scan_bound() {
    let (keyspace, catalog) = setup();
    let snapshot = keyspace.snapshot();
    let first = execute_query_with_options(
        &snapshot,
        &catalog,
        "A",
        "app",
        Query::select(&["*"])
            .from("users")
            .order_by("id", Order::Asc)
            .limit(5),
        &QueryOptions::default(),
        9,
        100,
        None,
    )
    .expect("first page");
    let err = execute_query_with_options(
        &snapshot,
        &catalog,
        "A",
        "app",
        Query::select(&["*"])
            .from("users")
            .order_by("id", Order::Asc)
            .limit(5),
        &QueryOptions {
            cursor: first.cursor,
            ..QueryOptions::default()
        },
        9,
        10,
        None,
    )
    .expect_err("cursor path should still honor scan bound");
    assert!(matches!(err, QueryError::ScanBoundExceeded { .. }));
}

#[test]
fn join_scan_bound_uses_cardinality_aware_estimate_when_right_side_is_primary_key() {
    let (mut keyspace, mut catalog) = setup();
    catalog
        .create_table(
            "A",
            "app",
            "profiles",
            vec![
                ColumnDef {
                    name: "user_id".into(),
                    col_type: ColumnType::Integer,
                    nullable: false,
                },
                ColumnDef {
                    name: "country".into(),
                    col_type: ColumnType::Text,
                    nullable: false,
                },
            ],
            vec!["user_id".into()],
        )
        .expect("profiles table");
    for i in 0..50 {
        keyspace.upsert_row(
            "A",
            "app",
            "profiles",
            vec![Value::Integer(i)],
            Row::from_values(vec![Value::Integer(i), Value::Text("US".into())]),
            1,
        );
    }
    let snapshot = keyspace.snapshot();
    let result = execute_query_with_options(
        &snapshot,
        &catalog,
        "A",
        "app",
        Query::select(&["u.id", "p.country"])
            .from("users")
            .alias("u")
            .inner_join("profiles", "u.id", "user_id")
            .with_last_join_alias("p")
            .limit(10),
        &QueryOptions::default(),
        1,
        1_000,
        None,
    )
    .expect("primary-key join should stay within scan bound");
    assert_eq!(result.rows.len(), 10);
    assert!(result.rows_examined <= 50);
}

/// A pure-window scan (LIMIT, no cursor/order/filter) may stop reading at the
/// page window, but the page itself must be identical to full evaluation:
/// primary-key order with OFFSET applied before LIMIT.
#[test]
fn pure_window_scan_returns_exact_page() {
    let (keyspace, catalog) = setup();
    let snapshot = keyspace.snapshot();
    let result = super::tests::execute_query(
        &snapshot,
        &catalog,
        "A",
        "app",
        Query::select(&["id"]).from("users").offset(5).limit(10),
    )
    .expect("windowed scan");
    let ids: Vec<i64> = result
        .rows
        .iter()
        .map(|r| match r.values[0] {
            Value::Integer(id) => id,
            ref other => panic!("unexpected id value: {other:?}"),
        })
        .collect();
    assert_eq!(ids, (5..15).collect::<Vec<_>>());
    assert!(result.truncated);
}

/// A residually-filtered scan must stay unbounded: the filter drops source
/// rows, so stopping at the page window would under-fill the page. `email`
/// has no index, forcing the residual full-scan path.
#[test]
fn residual_filtered_scan_is_not_window_bounded() {
    let (keyspace, catalog) = setup();
    let snapshot = keyspace.snapshot();
    let result = super::tests::execute_query(
        &snapshot,
        &catalog,
        "A",
        "app",
        Query::select(&["id"])
            .from("users")
            .where_(col("email").like(lit("%@example.com")))
            .limit(5),
    )
    .expect("residual scan");
    let ids: Vec<i64> = result
        .rows
        .iter()
        .map(|r| match r.values[0] {
            Value::Integer(id) => id,
            ref other => panic!("unexpected id value: {other:?}"),
        })
        .collect();
    // Odd ids carry @example.com addresses; a wrongly bounded source would
    // only surface the first few.
    assert_eq!(ids, vec![1, 3, 5, 7, 9]);
    assert!(result.truncated);
}

/// A residually-filtered index lookup hydrates rows lazily but must still
/// produce the exact page full evaluation would: candidates stream in index
/// (name-lexicographic) order and the residual LIKE filter plus LIMIT pick
/// the first matches.
#[test]
fn residual_index_scan_returns_exact_page_via_lazy_hydration() {
    let (keyspace, catalog) = setup();
    let snapshot = keyspace.snapshot();
    let result = super::tests::execute_query(
        &snapshot,
        &catalog,
        "A",
        "app",
        Query::select(&["id"])
            .from("users")
            .where_(col("name").like(lit("u%7%")))
            .limit(5),
    )
    .expect("residual index scan");
    let ids: Vec<i64> = result
        .rows
        .iter()
        .map(|r| match r.values[0] {
            Value::Integer(id) => id,
            ref other => panic!("unexpected id value: {other:?}"),
        })
        .collect();
    // by_name index order is lexicographic: u17 < u27 < u37 < u47 < u57 are
    // the first names containing a '7'.
    assert_eq!(ids, vec![17, 27, 37, 47, 57]);
    assert!(result.truncated);
}

/// A row-fetch failure on a candidate the page actually needs must surface as
/// a query error, not truncate the result. A spilled row with no value store
/// attached makes the fetch fail deterministically.
#[test]
fn lazy_hydration_surfaces_fetch_errors_for_needed_rows() {
    let (mut keyspace, catalog) = setup();
    let table = keyspace
        .table_by_namespace_key_mut(&crate::catalog::namespace_key("A", "app"), "users")
        .expect("users table");
    table.rows.insert(
        crate::storage::encoded_key::EncodedKey::from_values(&[Value::Integer(37)]),
        crate::storage::keyspace::StoredRow::spilled_versioned(
            1,
            crate::storage::value_store::PersistentValueRef {
                offset: 0,
                len: 8,
                blake3_hash: [0; 32],
            },
        ),
    );
    let snapshot = keyspace.snapshot();
    // u37 is the third residual match; a LIMIT 5 page needs it.
    let err = super::tests::execute_query(
        &snapshot,
        &catalog,
        "A",
        "app",
        Query::select(&["id"])
            .from("users")
            .where_(col("name").like(lit("u%7%")))
            .limit(5),
    )
    .expect_err("fetch of poisoned row must fail the query");
    assert!(
        format!("{err:?}").contains("value store"),
        "unexpected error: {err:?}"
    );
}

/// A page fully satisfied before the failing candidate succeeds: lazy
/// hydration never touches rows past the pipeline's stop point (the same
/// contract as the window-bounded scan paths).
#[test]
fn lazy_hydration_skips_unneeded_rows_entirely() {
    let (mut keyspace, catalog) = setup();
    let table = keyspace
        .table_by_namespace_key_mut(&crate::catalog::namespace_key("A", "app"), "users")
        .expect("users table");
    table.rows.insert(
        crate::storage::encoded_key::EncodedKey::from_values(&[Value::Integer(37)]),
        crate::storage::keyspace::StoredRow::spilled_versioned(
            1,
            crate::storage::value_store::PersistentValueRef {
                offset: 0,
                len: 8,
                blake3_hash: [0; 32],
            },
        ),
    );
    let snapshot = keyspace.snapshot();
    // LIMIT 1 needs u17 plus the u27 lookahead — u37 is never fetched.
    let result = super::tests::execute_query(
        &snapshot,
        &catalog,
        "A",
        "app",
        Query::select(&["id"])
            .from("users")
            .where_(col("name").like(lit("u%7%")))
            .limit(1),
    )
    .expect("page satisfied before poisoned row");
    let ids: Vec<i64> = result
        .rows
        .iter()
        .map(|r| match r.values[0] {
            Value::Integer(id) => id,
            ref other => panic!("unexpected id value: {other:?}"),
        })
        .collect();
    assert_eq!(ids, vec![17]);
    assert!(result.truncated);
}

/// A residual match past the first candidate chunk (64 postings) must still be
/// found: the chunked stream advances its window until the index range is
/// exhausted. `u99` is the only name matching `u%99%` and sits near the end of
/// the 100-entry name-ordered candidate range.
#[test]
fn chunked_residual_stream_finds_matches_past_first_chunk() {
    let (keyspace, catalog) = setup();
    let snapshot = keyspace.snapshot();
    let result = super::tests::execute_query(
        &snapshot,
        &catalog,
        "A",
        "app",
        Query::select(&["id"])
            .from("users")
            .where_(col("name").like(lit("u%99%")))
            .limit(5),
    )
    .expect("chunked residual scan");
    let ids: Vec<i64> = result
        .rows
        .iter()
        .map(|r| match r.values[0] {
            Value::Integer(id) => id,
            ref other => panic!("unexpected id value: {other:?}"),
        })
        .collect();
    assert_eq!(ids, vec![99]);
    assert!(!result.truncated);
    assert!(result.cursor.is_none());
}

/// A residual pattern with no matches drains the whole candidate range and
/// returns an empty, non-truncated result.
#[test]
fn chunked_residual_stream_handles_zero_matches() {
    let (keyspace, catalog) = setup();
    let snapshot = keyspace.snapshot();
    let result = super::tests::execute_query(
        &snapshot,
        &catalog,
        "A",
        "app",
        Query::select(&["id"])
            .from("users")
            .where_(col("name").like(lit("u%zzz%")))
            .limit(5),
    )
    .expect("no-match residual scan");
    assert!(result.rows.is_empty());
    assert!(!result.truncated);
}

/// Read-set-capturing queries take the eager candidate path (touched pks must
/// be recorded up front) but must return the same page as the streamed path.
#[test]
fn capturing_residual_query_matches_streamed_page() {
    let (keyspace, catalog) = setup();
    let snapshot = keyspace.snapshot();
    let query = || {
        Query::select(&["id"])
            .from("users")
            .where_(col("name").like(lit("u%7%")))
            .limit(5)
    };
    let streamed = super::tests::execute_query(&snapshot, &catalog, "A", "app", query())
        .expect("streamed page");
    let mut collector = crate::query::executor::ReadSetCollector::new();
    let captured = crate::query::executor::execute_query_with_options_capturing(
        crate::query::executor::CapturingQueryExecutionRequest {
            snapshot: &snapshot,
            catalog: &catalog,
            project_id: "A",
            scope_id: "app",
            query: query(),
            options: &QueryOptions {
                allow_full_scan: true,
                ..QueryOptions::default()
            },
            snapshot_seq: 0,
            max_scan_rows: usize::MAX,
            read_set: Some(&mut collector),
        },
    )
    .expect("capturing page");
    assert_eq!(streamed.rows, captured.rows);
    assert_eq!(streamed.truncated, captured.truncated);
    // The streamed capture is coarse (whole-table range): it never enumerates
    // the full candidate set, so invalidation must come from a range record.
    let read_set = collector.into_inner();
    assert!(
        !read_set.ranges.is_empty(),
        "capturing residual query must record an invalidation range"
    );
}

/// COUNT(*) with no predicate must not depend on row contents: the synthetic
/// count source must produce the identical result (value, rows_examined) to
/// the content-scanning path, forced here via an always-true predicate.
#[test]
fn count_star_matches_scanning_aggregate() {
    let (keyspace, catalog) = setup();
    let snapshot = keyspace.snapshot();
    let fast = super::tests::execute_query(
        &snapshot,
        &catalog,
        "A",
        "app",
        Query::select(&["count_star"])
            .from("users")
            .aggregate(Aggregate::Count),
    )
    .expect("count fast path");
    let scanned = super::tests::execute_query(
        &snapshot,
        &catalog,
        "A",
        "app",
        Query::select(&["count_star"])
            .from("users")
            .where_(col("age").gte(lit(i64::MIN)))
            .aggregate(Aggregate::Count),
    )
    .expect("count via scan");
    assert_eq!(fast.rows, scanned.rows);
    assert_eq!(fast.rows_examined, scanned.rows_examined);
    assert_eq!(fast.rows.len(), 1);
}

/// The synthetic count source must preserve the scan-bound guard: a COUNT(*)
/// (full evaluation, no LIMIT) over a table larger than max_scan_rows is
/// rejected exactly as the scanning path rejects it.
#[test]
fn count_star_preserves_scan_bound_guard() {
    let (keyspace, catalog) = setup();
    let snapshot = keyspace.snapshot();
    let err = execute_query_with_options(
        &snapshot,
        &catalog,
        "A",
        "app",
        Query::select(&["count_star"])
            .from("users")
            .aggregate(Aggregate::Count),
        &QueryOptions::default(),
        0,
        50,
        None,
    )
    .expect_err("count over max_scan_rows must be rejected");
    match err {
        QueryError::ScanBoundExceeded {
            estimated_rows,
            max_scan_rows,
        } => {
            assert_eq!(estimated_rows, 100);
            assert_eq!(max_scan_rows, 50);
        }
        other => panic!("unexpected error: {other:?}"),
    }
}

/// IN over the primary key resolves to point probes: ascending-PK output
/// order, collapsed duplicates, and skipped missing keys — identical to the
/// scan-and-filter result — while examining only the probed rows.
#[test]
fn pk_in_probes_match_scan_semantics() {
    let (keyspace, catalog) = setup();
    let snapshot = keyspace.snapshot();
    let result = super::tests::execute_query(
        &snapshot,
        &catalog,
        "A",
        "app",
        Query::select(&["id"])
            .from("users")
            .where_(col("id").in_(vec![
                lit(7_i64),
                lit(3_i64),
                lit(3_i64),
                lit(99_i64),
                lit(1000_i64),
            ]))
            .limit(10),
    )
    .expect("pk in probe");
    let ids: Vec<i64> = result
        .rows
        .iter()
        .map(|r| match r.values[0] {
            Value::Integer(id) => id,
            ref other => panic!("unexpected id value: {other:?}"),
        })
        .collect();
    assert_eq!(ids, vec![3, 7, 99]);
    assert!(
        result.rows_examined <= 4,
        "IN over the pk must probe, not scan: rows_examined = {}",
        result.rows_examined
    );
}

/// A mixed-type IN list must keep the coercing scan semantics (U64 5 matches
/// the Integer pk 5 via compare_values), which type-tagged probes cannot
/// honor — so the probe gate has to fall back to the scan.
#[test]
fn pk_in_mixed_types_fall_back_to_coercing_scan() {
    let (keyspace, catalog) = setup();
    let snapshot = keyspace.snapshot();
    let result = super::tests::execute_query(
        &snapshot,
        &catalog,
        "A",
        "app",
        Query::select(&["id"])
            .from("users")
            .where_(col("id").in_(vec![Value::U64(5), Value::Integer(7)]))
            .limit(10),
    )
    .expect("mixed-type in");
    let ids: Vec<i64> = result
        .rows
        .iter()
        .map(|r| match r.values[0] {
            Value::Integer(id) => id,
            ref other => panic!("unexpected id value: {other:?}"),
        })
        .collect();
    // U64(5) must still match the Integer pk 5 through comparison coercion.
    assert_eq!(ids, vec![5, 7]);
}

/// ORDER BY pk DESC LIMIT walks the PK map in reverse and stops at the page
/// window instead of scanning and sorting the whole table.
#[test]
fn pk_ordered_desc_scan_returns_bounded_latest_page() {
    let (keyspace, catalog) = setup();
    let snapshot = keyspace.snapshot();
    let result = super::tests::execute_query(
        &snapshot,
        &catalog,
        "A",
        "app",
        Query::select(&["id"])
            .from("users")
            .order_by("id", Order::Desc)
            .limit(5),
    )
    .expect("desc pk scan");
    let ids: Vec<i64> = result
        .rows
        .iter()
        .map(|r| match r.values[0] {
            Value::Integer(id) => id,
            ref other => panic!("unexpected id value: {other:?}"),
        })
        .collect();
    assert_eq!(ids, vec![99, 98, 97, 96, 95]);
    assert!(result.truncated);
    assert!(result.cursor.is_some());
    assert!(
        result.rows_examined <= 6,
        "desc pk scan must stop at the window: rows_examined = {}",
        result.rows_examined
    );
}

/// The PK-ordered walk applies OFFSET in-walk without materializing skipped
/// rows, matching the general path's page exactly.
#[test]
fn pk_ordered_scan_applies_offset_in_walk() {
    let (keyspace, catalog) = setup();
    let snapshot = keyspace.snapshot();
    let result = super::tests::execute_query(
        &snapshot,
        &catalog,
        "A",
        "app",
        Query::select(&["id"])
            .from("users")
            .order_by("id", Order::Asc)
            .offset(10)
            .limit(5),
    )
    .expect("offset pk scan");
    let ids: Vec<i64> = result
        .rows
        .iter()
        .map(|r| match r.values[0] {
            Value::Integer(id) => id,
            ref other => panic!("unexpected id value: {other:?}"),
        })
        .collect();
    assert_eq!(ids, (10..15).collect::<Vec<_>>());
}

/// Descending cursor pages seek and cover the whole table exactly once, in
/// reverse PK order, without re-examining prior pages.
#[test]
fn pk_ordered_desc_cursor_walk_covers_all_rows() {
    let (keyspace, catalog) = setup();
    let snapshot = keyspace.snapshot();
    let mut collected: Vec<i64> = Vec::new();
    let mut cursor: Option<String> = None;
    for page in 0..30 {
        let result = execute_query_with_options(
            &snapshot,
            &catalog,
            "A",
            "app",
            Query::select(&["id"])
                .from("users")
                .order_by("id", Order::Desc)
                .limit(7),
            &QueryOptions {
                cursor: cursor.take(),
                allow_full_scan: true,
                ..QueryOptions::default()
            },
            3,
            usize::MAX,
            None,
        )
        .expect("desc cursor page");
        if page > 0 {
            assert!(
                result.rows_examined <= 8,
                "cursor page must seek, not rescan: rows_examined = {}",
                result.rows_examined
            );
        }
        collected.extend(result.rows.iter().map(|r| match r.values[0] {
            Value::Integer(id) => id,
            ref other => panic!("unexpected id value: {other:?}"),
        }));
        cursor = result.cursor;
        if cursor.is_none() {
            break;
        }
    }
    assert_eq!(collected, (0..100).rev().collect::<Vec<_>>());
}

/// The bounded PK-ordered walk must return the identical page to full
/// evaluation, forced here through an always-true indexed predicate (which
/// routes the query through the general candidate + sort machinery).
#[test]
fn pk_ordered_desc_matches_general_path() {
    let (keyspace, catalog) = setup();
    let snapshot = keyspace.snapshot();
    let fast = super::tests::execute_query(
        &snapshot,
        &catalog,
        "A",
        "app",
        Query::select(&["id", "name"])
            .from("users")
            .order_by("id", Order::Desc)
            .limit(12),
    )
    .expect("bounded walk");
    let general = super::tests::execute_query(
        &snapshot,
        &catalog,
        "A",
        "app",
        Query::select(&["id", "name"])
            .from("users")
            .where_(col("age").gte(lit(i64::MIN)))
            .order_by("id", Order::Desc)
            .limit(12),
    )
    .expect("general path");
    assert_eq!(fast.rows, general.rows);
    assert_eq!(fast.truncated, general.truncated);
}

/// COUNT(*) over a table carrying tombstones takes the key-only tier merge
/// and must not count tombstone-only keys.
#[test]
fn count_star_ignores_tombstone_only_keys() {
    let (mut keyspace, catalog) = setup();
    let table = keyspace
        .table_by_namespace_key_mut(&crate::catalog::namespace_key("A", "app"), "users")
        .expect("users table");
    table.row_tombstones.insert(
        crate::storage::encoded_key::EncodedKey::from_values(&[Value::Integer(1000)]),
        1,
    );
    let snapshot = keyspace.snapshot();
    let result = super::tests::execute_query(
        &snapshot,
        &catalog,
        "A",
        "app",
        Query::select(&["count_star"])
            .from("users")
            .aggregate(Aggregate::Count),
    )
    .expect("count with tombstones");
    assert_eq!(result.rows.len(), 1);
    assert_eq!(result.rows_examined, 100);
}