aedb 0.3.1

//! Behavior pins for the streaming PK-probe join fast path
//! (`join_window::try_streaming_pk_probe_join`): page fill past probe misses,
//! LEFT null-extension, cursor-walk coverage, OFFSET, parity with the general
//! path, DESC fallback, and cursor seek.

use super::execute_query_with_options;
use super::join_tests::{add_profiles_table, seed_profiles};
use super::tests::{execute_query, setup};
use crate::catalog::types::{Row, Value};
use crate::query::plan::{Expr, Order, Query, QueryOptions};

/// The streaming PK-probe fast path must keep pulling base rows past probe
/// misses until the page window fills, and page/cursor/truncation must match
/// what full evaluation produces.
#[test]
fn streaming_join_fills_page_past_probe_misses() {
    let (mut keyspace, mut catalog) = setup();
    add_profiles_table(&mut catalog, vec!["user_id".into()]);
    // Only even user ids have profiles: every odd base row is a probe miss.
    seed_profiles(&mut keyspace, (0..100).filter(|i| i % 2 == 0), |_| "US");

    let snapshot = keyspace.snapshot();
    let result = execute_query(
        &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")
            .order_by("u.id", Order::Asc)
            .limit(10),
    )
    .expect("inner join with misses");
    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, (0..20).step_by(2).collect::<Vec<_>>());
    assert!(result.truncated);
    assert!(result.cursor.is_some());

    // LEFT JOIN: odd ids survive with a null-extended right side, so the first
    // page is simply ids 0..10 with alternating null country.
    let left = execute_query(
        &snapshot,
        &catalog,
        "A",
        "app",
        Query::select(&["u.id", "p.country"])
            .from("users")
            .alias("u")
            .left_join("profiles", "u.id", "user_id")
            .with_last_join_alias("p")
            .order_by("u.id", Order::Asc)
            .limit(10),
    )
    .expect("left join with misses");
    let left_pairs: Vec<(i64, bool)> = left
        .rows
        .iter()
        .map(|r| {
            let id = match r.values[0] {
                Value::Integer(id) => id,
                ref other => panic!("unexpected id value: {other:?}"),
            };
            (id, matches!(r.values[1], Value::Null))
        })
        .collect();
    let expected: Vec<(i64, bool)> = (0..10).map(|id| (id, id % 2 == 1)).collect();
    assert_eq!(left_pairs, expected);
}

/// Cursor pages produced by the streaming fast path must cover the full joined
/// result exactly once, in order.
#[test]
fn streaming_join_cursor_pages_cover_full_result() {
    let (mut keyspace, mut catalog) = setup();
    add_profiles_table(&mut catalog, vec!["user_id".into()]);
    seed_profiles(&mut keyspace, (0..100).filter(|i| i % 2 == 0), |_| "US");
    let snapshot = keyspace.snapshot();

    let mut collected: Vec<i64> = Vec::new();
    let mut cursor: Option<String> = None;
    for _ in 0..20 {
        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")
                .order_by("u.id", Order::Asc)
                .limit(7),
            &QueryOptions {
                cursor: cursor.take(),
                allow_full_scan: true,
                ..QueryOptions::default()
            },
            3,
            usize::MAX,
            None,
        )
        .expect("cursor page");
        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!(cursor.is_none(), "pagination must terminate");
    assert_eq!(collected, (0..100).step_by(2).collect::<Vec<_>>());
}

/// OFFSET applies to the joined output, after probe misses are dropped.
#[test]
fn streaming_join_offset_applies_to_joined_output() {
    let (mut keyspace, mut catalog) = setup();
    add_profiles_table(&mut catalog, vec!["user_id".into()]);
    seed_profiles(&mut keyspace, (0..100).filter(|i| i % 2 == 0), |_| "US");
    let snapshot = keyspace.snapshot();

    let result = execute_query(
        &snapshot,
        &catalog,
        "A",
        "app",
        Query::select(&["u.id"])
            .from("users")
            .alias("u")
            .inner_join("profiles", "u.id", "user_id")
            .with_last_join_alias("p")
            .order_by("u.id", Order::Asc)
            .offset(10)
            .limit(5),
    )
    .expect("offset join");
    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();
    // Joined output is even ids ascending; skipping 10 lands at 20.
    assert_eq!(ids, vec![20, 22, 24, 26, 28]);
}

/// The streaming fast path must produce byte-identical pages to the general
/// join path (forced here via an always-true WHERE, which disables streaming).
#[test]
fn streaming_join_matches_general_path_output() {
    let (mut keyspace, mut catalog) = setup();
    add_profiles_table(&mut catalog, vec!["user_id".into()]);
    seed_profiles(&mut keyspace, (0..100).filter(|i| i % 3 == 0), |id| {
        if id % 2 == 0 { "US" } else { "CA" }
    });
    let snapshot = keyspace.snapshot();

    let fast = execute_query(
        &snapshot,
        &catalog,
        "A",
        "app",
        Query::select(&["u.id", "u.name", "p.country"])
            .from("users")
            .alias("u")
            .inner_join("profiles", "u.id", "user_id")
            .with_last_join_alias("p")
            .order_by("u.id", Order::Asc)
            .limit(12),
    )
    .expect("streaming path");
    let general = execute_query(
        &snapshot,
        &catalog,
        "A",
        "app",
        Query::select(&["u.id", "u.name", "p.country"])
            .from("users")
            .alias("u")
            .inner_join("profiles", "u.id", "user_id")
            .with_last_join_alias("p")
            .where_(Expr::Gte("u.age".into(), Value::Integer(i64::MIN)))
            .order_by("u.id", Order::Asc)
            .limit(12),
    )
    .expect("general path");
    assert_eq!(fast.rows, general.rows);
    assert_eq!(fast.truncated, general.truncated);
}

/// Descending PK order is ineligible for streaming and must still return the
/// correctly sorted page via the general path.
#[test]
fn streaming_join_desc_order_falls_back_to_sorted_page() {
    let (mut keyspace, mut catalog) = setup();
    add_profiles_table(&mut catalog, vec!["user_id".into()]);
    seed_profiles(&mut keyspace, (0..100).filter(|i| i % 2 == 0), |_| "US");
    let snapshot = keyspace.snapshot();

    let result = execute_query(
        &snapshot,
        &catalog,
        "A",
        "app",
        Query::select(&["u.id"])
            .from("users")
            .alias("u")
            .inner_join("profiles", "u.id", "user_id")
            .with_last_join_alias("p")
            .order_by("u.id", Order::Desc)
            .limit(5),
    )
    .expect("desc join");
    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![98, 96, 94, 92, 90]);
}

/// Cursor continuation on the streaming fast path must seek directly past the
/// previous page instead of re-examining it from the start of the base table.
#[test]
fn streaming_join_cursor_page_seeks_past_prior_pages() {
    let (mut keyspace, mut catalog) = setup();
    add_profiles_table(&mut catalog, vec!["user_id".into()]);
    seed_profiles(&mut keyspace, (0..100).filter(|i| i % 2 == 0), |_| "US");
    let snapshot = keyspace.snapshot();

    let page = |cursor: Option<String>| {
        execute_query_with_options(
            &snapshot,
            &catalog,
            "A",
            "app",
            Query::select(&["u.id"])
                .from("users")
                .alias("u")
                .inner_join("profiles", "u.id", "user_id")
                .with_last_join_alias("p")
                .order_by("u.id", Order::Asc)
                .limit(7),
            &QueryOptions {
                cursor,
                allow_full_scan: true,
                ..QueryOptions::default()
            },
            3,
            usize::MAX,
            None,
        )
        .expect("cursor page")
    };
    let first = page(None);
    let second = page(first.cursor.clone());
    let second_ids: Vec<i64> = second
        .rows
        .iter()
        .map(|r| match r.values[0] {
            Value::Integer(id) => id,
            ref other => panic!("unexpected id value: {other:?}"),
        })
        .collect();
    assert_eq!(second_ids, vec![14, 16, 18, 20, 22, 24, 26]);
    // A skip-scan restart would re-examine the first page's seven joined rows
    // before producing this page; the seek must not.
    assert!(
        second.rows_examined <= 8,
        "cursor page re-examined prior pages: rows_examined = {}",
        second.rows_examined
    );
}

/// Negative integer primary keys exercise the encoded-key sign-flip: the
/// cursor seek and the streamed ordering must both agree with signed numeric
/// order across the negative/positive boundary.
#[test]
fn streaming_join_cursor_seek_orders_negative_pks_correctly() {
    let (mut keyspace, mut catalog) = setup();
    add_profiles_table(&mut catalog, vec!["user_id".into()]);
    // Extend users with negative ids; profiles exist for every id in [-5, 5).
    for id in -5..0 {
        keyspace.upsert_row(
            "A",
            "app",
            "users",
            vec![Value::Integer(id)],
            Row {
                values: vec![
                    Value::Integer(id),
                    Value::Text(format!("n{}", -id).into()),
                    Value::Integer(30),
                    Value::Null,
                ],
            },
            1,
        );
    }
    seed_profiles(&mut keyspace, -5..5, |_| "US");
    let snapshot = keyspace.snapshot();

    let page = |cursor: Option<String>| {
        execute_query_with_options(
            &snapshot,
            &catalog,
            "A",
            "app",
            Query::select(&["u.id"])
                .from("users")
                .alias("u")
                .inner_join("profiles", "u.id", "user_id")
                .with_last_join_alias("p")
                .order_by("u.id", Order::Asc)
                .limit(4),
            &QueryOptions {
                cursor,
                allow_full_scan: true,
                ..QueryOptions::default()
            },
            3,
            usize::MAX,
            None,
        )
        .expect("cursor page")
    };
    let ids = |result: &super::QueryResult| -> Vec<i64> {
        result
            .rows
            .iter()
            .map(|r| match r.values[0] {
                Value::Integer(id) => id,
                ref other => panic!("unexpected id value: {other:?}"),
            })
            .collect()
    };
    let first = page(None);
    assert_eq!(ids(&first), vec![-5, -4, -3, -2]);
    // The second page's cursor sort key is a negative integer; the seek must
    // resume exactly at -1 and cross into the positives in signed order.
    let second = page(first.cursor.clone());
    assert_eq!(ids(&second), vec![-1, 0, 1, 2]);
    assert!(second.rows_examined <= 5);
    let third = page(second.cursor.clone());
    assert_eq!(ids(&third), vec![3, 4]);
    assert!(third.cursor.is_none());
}

/// DESC-ordered streaming joins walk the base table in reverse, and cursor
/// pages seek backwards without re-examining prior pages.
#[test]
fn streaming_join_desc_pages_and_seeks() {
    let (mut keyspace, mut catalog) = setup();
    add_profiles_table(&mut catalog, vec!["user_id".into()]);
    seed_profiles(&mut keyspace, (0..100).filter(|i| i % 2 == 0), |_| "US");
    let snapshot = keyspace.snapshot();

    let page = |cursor: Option<String>| {
        execute_query_with_options(
            &snapshot,
            &catalog,
            "A",
            "app",
            Query::select(&["u.id"])
                .from("users")
                .alias("u")
                .inner_join("profiles", "u.id", "user_id")
                .with_last_join_alias("p")
                .order_by("u.id", Order::Desc)
                .limit(7),
            &QueryOptions {
                cursor,
                allow_full_scan: true,
                ..QueryOptions::default()
            },
            3,
            usize::MAX,
            None,
        )
        .expect("desc cursor page")
    };
    let ids = |result: &super::QueryResult| -> Vec<i64> {
        result
            .rows
            .iter()
            .map(|r| match r.values[0] {
                Value::Integer(id) => id,
                ref other => panic!("unexpected id value: {other:?}"),
            })
            .collect()
    };
    let first = page(None);
    assert_eq!(ids(&first), vec![98, 96, 94, 92, 90, 88, 86]);
    let second = page(first.cursor.clone());
    assert_eq!(ids(&second), vec![84, 82, 80, 78, 76, 74, 72]);
    assert!(
        second.rows_examined <= 8,
        "desc cursor page must seek: rows_examined = {}",
        second.rows_examined
    );

    let mut collected: Vec<i64> = Vec::new();
    let mut cursor: Option<String> = None;
    for _ in 0..20 {
        let result = page(cursor.take());
        collected.extend(ids(&result));
        cursor = result.cursor;
        if cursor.is_none() {
            break;
        }
    }
    assert_eq!(
        collected,
        (0..100).rev().filter(|i| i % 2 == 0).collect::<Vec<_>>()
    );
}

/// A join table whose rows were evicted to the cold tier must still join:
/// probes resolve segment copies per key, on both the streaming fast path
/// (hot base) and the general PK-probe path.
#[test]
fn join_probes_resolve_cold_tier_join_table() {
    use crate::storage::kv_segment::KvSegmentStore;
    use std::sync::Arc;

    let dir = tempfile::tempdir().expect("temp dir");
    let store = Arc::new(KvSegmentStore::open(dir.path()).expect("open segment store"));
    let mut keyspace = crate::storage::keyspace::Keyspace::default();
    keyspace.attach_kv_segment_store(Arc::clone(&store));
    let mut catalog = crate::catalog::Catalog::default();
    catalog.create_project("A").expect("project");
    catalog
        .create_table(
            "A",
            "app",
            "users",
            vec![
                crate::catalog::schema::ColumnDef {
                    name: "id".into(),
                    col_type: crate::catalog::types::ColumnType::Integer,
                    nullable: false,
                },
                crate::catalog::schema::ColumnDef {
                    name: "name".into(),
                    col_type: crate::catalog::types::ColumnType::Text,
                    nullable: false,
                },
            ],
            vec!["id".into()],
        )
        .expect("users table");
    add_profiles_table(&mut catalog, vec!["user_id".into()]);

    // Seed and evict the join table first, then seed the base table hot, so
    // the streaming fast path stays eligible while every probe goes cold.
    seed_profiles(&mut keyspace, 0..20, |_| "US");
    keyspace
        .flush_table_rows_to_segments_to_memory_target(0)
        .expect("evict profiles");
    for i in 0..20i64 {
        keyspace.upsert_row(
            "A",
            "app",
            "users",
            vec![Value::Integer(i)],
            Row {
                values: vec![Value::Integer(i), Value::Text(format!("u{i}").into())],
            },
            (i + 100) as u64,
        );
    }
    let snapshot = keyspace.snapshot();
    assert!(
        !snapshot
            .table("A", "app", "profiles")
            .expect("profiles")
            .row_segments
            .is_empty(),
        "profiles must be cold"
    );

    let result = execute_query(
        &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")
            .order_by("u.id", Order::Asc)
            .limit(5),
    )
    .expect("join over cold profiles");
    let pairs: Vec<(i64, &str)> = result
        .rows
        .iter()
        .map(|r| match (&r.values[0], &r.values[1]) {
            (Value::Integer(id), Value::Text(c)) => (*id, c.as_str()),
            other => panic!("unexpected row: {other:?}"),
        })
        .collect();
    assert_eq!(
        pairs,
        vec![(0, "US"), (1, "US"), (2, "US"), (3, "US"), (4, "US")]
    );
}