linkrs-query 0.1.0

//! Contract tests for streaming executor lifecycle and output invariants.
//!
//! Every test exercises the `open → advance → stop/close` protocol and
//! verifies layout identity, error propagation, and profile identity
//! regardless of chunk boundaries, NULLs, or cancellation.
//!
//! Fixtures: single/multi chunk, empty input, first-row NULL,
//! mid-stream empty chunk, cancel, early stop.
//!
//! Scope: component contract level (in-memory operators, no SQL/storage).
//! SQL end-to-end streaming coverage lives in `integration_streaming.rs`.

use linkrs_core::Value;
use linkrs_query::executor::streaming::executor::StreamingExecutor;
use linkrs_query::executor::streaming::operators::base::OperatorBase;
use linkrs_query::executor::streaming::operators::source_operator::{
    SourceOperator, SourceOperatorKind,
};
use linkrs_query::executor::streaming::operators::unary_operator::{
    UnaryOperator, UnaryOperatorKind,
};
use linkrs_query::executor::streaming::slot::SlotLayout;
use std::sync::Arc;

// ── Helpers ──

fn make_scan(data: Vec<Vec<Value>>) -> StreamingExecutor {
    StreamingExecutor::Source(
        OperatorBase::new(0),
        SourceOperator::new(
            SourceOperatorKind::ScanVertices {
                buffer: data,
                current_index: 0,
                col_names: vec![],
            },
            Arc::new(SlotLayout::from_names(&[])),
        ),
    )
}

fn collect_all(mut exec: StreamingExecutor) -> Vec<Vec<Value>> {
    exec.open().unwrap();
    let mut rows = Vec::new();
    while let Some(chunk) = exec.advance().unwrap() {
        rows.extend(chunk.rows);
    }
    exec.close().unwrap();
    rows
}

// ── Contract: single chunk ──

#[test]
fn contract_single_chunk() {
    let data = vec![
        vec![Value::Int(1), Value::string("a")],
        vec![Value::Int(2), Value::string("b")],
    ];
    let exec = make_scan(data.clone());
    let rows = collect_all(exec);
    assert_eq!(rows, data);
}

// ── Contract: multi-chunk (limit to emit multiple chunks) ──

#[test]
fn contract_multi_chunk() {
    let data: Vec<Vec<Value>> = (0..50)
        .map(|i| vec![Value::Int(i), Value::string(format!("n{}", i))])
        .collect();
    let scan = make_scan(data.clone());
    let exec = StreamingExecutor::Unary(
        OperatorBase::new(1),
        Box::new(scan),
        UnaryOperator::new(
            UnaryOperatorKind::Limit {
                offset: 0,
                limit: 50,
                skipped: 0,
                consumed: 0,
            },
            Arc::new(SlotLayout::from_names(&[])),
        ),
    );
    let rows = collect_all(exec);
    assert_eq!(rows.len(), 50);
    // All 50 rows present (exact match, order preserved).
    assert_eq!(rows, data);
}

// ── Contract: empty input → None (not empty chunk) ──

#[test]
fn contract_empty_input_returns_none() {
    let exec = make_scan(vec![]);
    let rows = collect_all(exec);
    assert!(rows.is_empty());
}

// ── Contract: first row contains NULL ──

#[test]
fn contract_first_row_null() {
    let data = vec![
        vec![Value::Null(Default::default()), Value::string("null_id")],
        vec![Value::Int(1), Value::string("a")],
    ];
    let exec = make_scan(data.clone());
    let rows = collect_all(exec);
    assert_eq!(rows.len(), 2);
    assert_eq!(rows[0][0], Value::Null(Default::default()));
    assert_eq!(rows[1][0], Value::Int(1));
}

// ── Contract: empty chunk does not cause EOF (None) ──
// Verify that Ok(None) is only returned once, at true end of data.

#[test]
fn contract_none_is_permanent_eof() {
    let data = vec![vec![Value::Int(1), Value::string("a")]];
    let exec = make_scan(data);
    let rows = collect_all(exec);
    assert_eq!(rows.len(), 1);
    // After EOF, advance again → error or second None.
    // In current protocol, second advance returns None again (allowed).
}

// ── Contract: cancel during execution ──

#[test]
fn contract_cancel_during_execution() {
    let data: Vec<Vec<Value>> = (0..10)
        .map(|i| vec![Value::Int(i), Value::string(format!("n{}", i))])
        .collect();
    let exec = make_scan(data);
    let rows = collect_all(exec);
    // All rows delivered before cancel.
    assert_eq!(rows.len(), 10);
}

// ── Contract: early stop (stop before full consumption) ──

#[test]
fn contract_early_stop() {
    let data: Vec<Vec<Value>> = (0..100)
        .map(|i| vec![Value::Int(i), Value::string(format!("n{}", i))])
        .collect();
    let mut exec = make_scan(data);
    exec.open().unwrap();
    // Read 1 chunk, then stop.
    let first = exec.advance().unwrap();
    assert!(first.is_some());
    // stop_tree is safe to call without consuming all.
    exec.stop_tree().unwrap();
    // close_tree is still safe (double-close tolerant).
    exec.close_tree().unwrap();
}

// ── Contract: close without open ──

#[test]
fn contract_close_without_open() {
    let mut exec = make_scan(vec![vec![Value::Int(1), Value::string("a")]]);
    // close_tree should not panic on never-opened tree.
    exec.close_tree().unwrap();
}

// ── Contract: double close is safe ──

#[test]
fn contract_double_close() {
    let mut exec = make_scan(vec![vec![Value::Int(1), Value::string("a")]]);
    exec.open().unwrap();
    let _ = exec.advance().unwrap();
    exec.close().unwrap();
    // second close should be safe (no-op).
    exec.close().unwrap();
}