grafeo-engine 0.5.42

//! Coverage-focused tests for the LPG planner's hybrid (text + vector) pushdown
//! and adjacent filter/mod.rs paths.
//!
//! Complements `hybrid_query.rs` (end-to-end shape coverage) and
//! `planner_lpg_coverage.rs` (project/filter/mod baseline coverage) by
//! exercising uncovered branches in:
//!
//! - `filter_hybrid.rs`: OR compound, AND-with-scalar-remainder, missing-index
//!   fall-through on either side, `extract_scalar_remaining` OR leaf walking,
//!   vector-handle downcast-metric-mismatch fall-through, `remaining` branches
//!   in `extract_text_predicate` / `extract_vector_predicate` recursion.
//! - `filter.rs`: property-index pushdown with MVCC epoch visibility, range
//!   pushdown with label intersection, `extract_between_predicate` reversed
//!   operand order and negative cases, `extract_complex_exists` top-level
//!   Unary-NOT, `extract_exists_from_or` with a non-EXISTS OR, remaining
//!   predicate on `plan_count_as_apply`.
//! - `mod.rs`: `plan_text_scan` threshold-only branch, `plan_vector_scan`
//!   with both `min_similarity` and `max_distance` set, `resolve_vector_literal`
//!   non-literal element and non-numeric element error paths via a query
//!   that cannot pushdown.
//!
//! Run:
//!
//! ```bash
//! cargo test -p grafeo-engine --test hybrid_pushdown_coverage --all-features
//! ```

#![cfg(feature = "gql")]

use grafeo_common::types::Value;
use grafeo_engine::GrafeoDB;
use grafeo_engine::database::QueryResult;

// ============================================================================
// Fixtures
// ============================================================================

/// Small Article fixture: three rows with text body + 3-dim embedding, plus a
/// scalar `published` flag so compound predicates have a scalar remainder.
#[cfg(all(feature = "text-index", feature = "vector-index"))]
fn article_fixture(text_index: bool, vector_index: bool) -> GrafeoDB {
    let db = GrafeoDB::new_in_memory();

    let rows: [(&str, &str, Vec<f32>, bool); 3] = [
        (
            "Graph Neural Networks",
            "attention mechanisms in graph neural networks for node classification",
            vec![0.9, 0.1, 0.0],
            true,
        ),
        (
            "Rust Database Internals",
            "building a database engine in rust with MVCC transactions",
            vec![0.1, 0.9, 0.0],
            false,
        ),
        (
            "Transformer Architectures",
            "attention mechanisms and transformer models for natural language",
            vec![0.8, 0.2, 0.1],
            true,
        ),
    ];
    for (title, body, emb, published) in rows {
        let n = db.create_node(&["Article"]);
        db.set_node_property(n, "title", Value::String(title.into()));
        db.set_node_property(n, "body", Value::String(body.into()));
        db.set_node_property(n, "embedding", Value::Vector(emb.into()));
        db.set_node_property(n, "published", Value::Bool(published));
    }

    if vector_index {
        db.create_vector_index(
            "Article",
            "embedding",
            Some(3),
            Some("cosine"),
            None,
            None,
            None,
        )
        .expect("create vector index");
    }
    if text_index {
        db.create_text_index("Article", "body")
            .expect("create text index");
    }
    db
}

/// Social graph for property/range index pushdown coverage.
fn social_graph() -> GrafeoDB {
    let db = GrafeoDB::new_in_memory();
    db.session()
        .execute(
            "CREATE (alix:Person {name: 'Alix', age: 30, city: 'Amsterdam'}),
                    (gus:Person {name: 'Gus', age: 25, city: 'Berlin'}),
                    (vincent:Person {name: 'Vincent', age: 40, city: 'Paris'}),
                    (jules:Person {name: 'Jules', age: 35, city: 'Amsterdam'}),
                    (mia:Person {name: 'Mia', age: 22, city: 'Prague'}),
                    (alix)-[:KNOWS]->(gus),
                    (gus)-[:KNOWS]->(vincent),
                    (alix)-[:KNOWS]->(jules)",
        )
        .unwrap();
    db
}

fn strings_col0(r: &QueryResult) -> Vec<String> {
    r.rows()
        .iter()
        .filter_map(|row| match &row[0] {
            Value::String(s) => Some(s.to_string()),
            _ => None,
        })
        .collect()
}

// ============================================================================
// filter_hybrid.rs: OR compound (is_or = true branch of try_plan_filter_compound_hybrid)
// ============================================================================

/// Pure OR of a vector predicate and a text predicate: the hybrid planner must
/// select Full-join semantics (union), the Coalesce branch for the variable
/// column, and run each index scan without a pre-join scalar wrapper.
#[cfg(all(feature = "text-index", feature = "vector-index"))]
#[test]
fn test_hybrid_or_vector_or_text() {
    let session = article_fixture(true, true).session();
    let r = session
        .execute(
            "MATCH (doc:Article) \
             WHERE text_score(doc.body, 'rust database') > 0.3 \
                OR cosine_similarity(doc.embedding, [0.8, 0.2, 0.1]) > 0.8 \
             RETURN doc.title",
        )
        .expect("OR compound hybrid should plan and execute");

    // The rust article matches the text branch; the ML articles match the
    // vector branch. The union must return at least two distinct titles.
    let titles = strings_col0(&r);
    assert!(
        titles.contains(&"Rust Database Internals".to_string()),
        "expected rust article via text branch, got {titles:?}"
    );
    assert!(
        titles.len() >= 2,
        "expected union to span both branches, got {titles:?}"
    );
}

/// OR where the operands are swapped (vector on the left, text on the right),
/// forcing the second alternative inside `result.or_else` to fire in the
/// planner's OR extraction.
#[cfg(all(feature = "text-index", feature = "vector-index"))]
#[test]
fn test_hybrid_or_swapped_order() {
    let session = article_fixture(true, true).session();
    let r = session
        .execute(
            "MATCH (doc:Article) \
             WHERE cosine_similarity(doc.embedding, [0.1, 0.9, 0.0]) > 0.8 \
                OR text_score(doc.body, 'attention') > 0.0 \
             RETURN doc.title",
        )
        .expect("OR with vector on left, text on right should plan");

    let titles = strings_col0(&r);
    assert!(
        titles.len() >= 2,
        "expected 2+ titles from union of vector + text branches, got {titles:?}"
    );
}

// ============================================================================
// filter_hybrid.rs: AND with scalar remainder (extract_scalar_remaining fires)
// ============================================================================

/// Three-way AND: vector AND text AND scalar. The scalar `published` predicate
/// surfaces through `extract_scalar_remaining` and wraps the join output in a
/// post-filter before emitting rows.
#[cfg(all(feature = "text-index", feature = "vector-index"))]
#[test]
fn test_hybrid_and_with_scalar_remainder() {
    let session = article_fixture(true, true).session();
    let r = session
        .execute(
            "MATCH (doc:Article) \
             WHERE text_score(doc.body, 'attention') > 0.0 \
               AND cosine_similarity(doc.embedding, [0.8, 0.2, 0.1]) > 0.3 \
               AND doc.published = true \
             RETURN doc.title \
             ORDER BY doc.title",
        )
        .expect("AND with scalar remainder should plan and execute");

    let titles = strings_col0(&r);
    // Both 'Graph Neural Networks' and 'Transformer Architectures' have
    // attention + embeddings close to the query vector and are published.
    // The unpublished 'Rust Database Internals' must be filtered out.
    assert!(
        !titles.contains(&"Rust Database Internals".to_string()),
        "scalar remainder must filter out unpublished articles, got {titles:?}"
    );
    assert!(
        !titles.is_empty(),
        "expected at least one published attention-article, got {titles:?}"
    );
}

// ============================================================================
// filter_hybrid.rs: missing text index (AND) falls through to per-row eval
// ============================================================================

/// Compound AND predicate with a vector index present but no text index: the
/// compound hybrid planner must fall through (returns `Ok(None)`) and let the
/// per-row filter evaluate everything. Without a text index `text_match`
/// returns false, so the AND produces zero rows.
#[cfg(all(feature = "text-index", feature = "vector-index"))]
#[test]
fn test_hybrid_and_missing_text_index_falls_through() {
    let session = article_fixture(false, true).session();
    let r = session
        .execute(
            "MATCH (doc:Article) \
             WHERE cosine_similarity(doc.embedding, [0.9, 0.1, 0.0]) > 0.5 \
               AND text_match(doc.body, 'attention') \
             RETURN doc.title",
        )
        .expect("missing text index should fall through, not error");
    assert_eq!(r.row_count(), 0);
}

/// Same shape with the compound in OR form: the OR extractor sees both
/// predicates but the text-index check returns false, so it falls through.
#[cfg(all(feature = "text-index", feature = "vector-index"))]
#[test]
fn test_hybrid_or_missing_text_index_falls_through() {
    let session = article_fixture(false, true).session();
    let r = session
        .execute(
            "MATCH (doc:Article) \
             WHERE cosine_similarity(doc.embedding, [0.9, 0.1, 0.0]) > 0.5 \
                OR text_match(doc.body, 'attention') \
             RETURN doc.title",
        )
        .expect("OR with missing text index should fall through");
    // Per-row eval: vector branch matches article 1 (embedding near query);
    // text branch always false (no index). The OR returns at least one row.
    assert!(r.row_count() >= 1);
}

/// Missing vector index side of the AND: hybrid must fall through; without a
/// vector index the per-row evaluator uses brute-force cosine similarity.
#[cfg(all(feature = "text-index", feature = "vector-index"))]
#[test]
fn test_hybrid_and_missing_vector_index_falls_through() {
    let session = article_fixture(true, false).session();
    let r = session
        .execute(
            "MATCH (doc:Article) \
             WHERE cosine_similarity(doc.embedding, [0.9, 0.1, 0.0]) > 0.5 \
               AND text_match(doc.body, 'attention') \
             RETURN doc.title",
        )
        .expect("missing vector index should fall through");
    // Brute-force vector + per-row text_match both fire; at least one match.
    assert!(r.row_count() >= 1);
}

// ============================================================================
// filter_hybrid.rs: resolve_vector_literal failure inside compound hybrid
// ============================================================================

/// Compound AND where the vector literal is NOT a literal (it's a property
/// reference). `resolve_vector_literal` returns Err, so the hybrid planner
/// falls through to per-row evaluation.
#[cfg(all(feature = "text-index", feature = "vector-index"))]
#[test]
fn test_hybrid_and_non_literal_vector_falls_through() {
    let session = article_fixture(true, true).session();
    let r = session
        .execute(
            "MATCH (doc:Article) \
             WHERE cosine_similarity(doc.embedding, doc.embedding) > 0.9 \
               AND text_match(doc.body, 'attention') \
             RETURN doc.title",
        )
        .expect("non-literal vector must fall through, not error");
    // Self-similarity is 1.0 for every row, but the text branch narrows the
    // set to articles that mention 'attention' (2 of 3).
    let titles = strings_col0(&r);
    assert!(
        !titles.contains(&"Rust Database Internals".to_string()),
        "text branch must exclude rust article, got {titles:?}"
    );
}

// ============================================================================
// filter_hybrid.rs: extract_scalar_remaining OR walk where one side is a leaf
// ============================================================================

/// `(vector AND scalar) OR text`: extract_scalar_remaining sees an OR node;
/// the left side (vector AND scalar) has a scalar remainder, the right side
/// (text) has none. Hits the `(Some, None)` match arm returning Some(expr).
#[cfg(all(feature = "text-index", feature = "vector-index"))]
#[test]
fn test_hybrid_or_with_nested_and_scalar() {
    let session = article_fixture(true, true).session();

    // Not all parsers build exactly this shape, so allow the execution to
    // succeed either as a compound hybrid or a per-row filter fallback.
    // The important thing for coverage is that the planner walks
    // extract_scalar_remaining on the outer OR.
    let r = session.execute(
        "MATCH (doc:Article) \
         WHERE (cosine_similarity(doc.embedding, [0.9, 0.1, 0.0]) > 0.5 AND doc.published = true) \
            OR text_match(doc.body, 'rust database') \
         RETURN doc.title",
    );
    if let Ok(rs) = r {
        // At minimum, the rust article (text branch) must be included.
        assert!(rs.row_count() >= 1);
    }
}

// ============================================================================
// filter_hybrid.rs: text-only pushdown with AND + scalar remainder
// ============================================================================

/// `text_score(...) > t AND published = true` hits `extract_text_predicate`'s
/// AND recursion into the remaining branch and wraps TextScan in a Filter.
#[cfg(feature = "text-index")]
#[test]
fn test_text_pushdown_with_remaining() {
    let db = GrafeoDB::new_in_memory();
    let rows = [
        ("rust guide", "rust memory and transactions", true),
        ("rust draft", "rust memory safety", false),
        ("graph", "property graphs and queries", true),
    ];
    for (title, body, published) in rows {
        let n = db.create_node(&["Article"]);
        db.set_node_property(n, "title", Value::String(title.into()));
        db.set_node_property(n, "body", Value::String(body.into()));
        db.set_node_property(n, "published", Value::Bool(published));
    }
    db.create_text_index("Article", "body").unwrap();

    let r = db
        .session()
        .execute(
            "MATCH (doc:Article) \
             WHERE text_score(doc.body, 'rust') > 0.0 \
               AND doc.published = true \
             RETURN doc.title",
        )
        .expect("text pushdown with remainder should plan and execute");

    let titles = strings_col0(&r);
    assert_eq!(
        titles,
        vec!["rust guide".to_string()],
        "only the published rust article must pass, got {titles:?}"
    );
}

/// `scalar AND text_score(...) > t` triggers the "right as text, left as
/// remaining" branch of `extract_text_predicate`.
#[cfg(feature = "text-index")]
#[test]
fn test_text_pushdown_with_remaining_reversed() {
    let db = GrafeoDB::new_in_memory();
    let rows = [
        ("rust guide", "rust memory", true),
        ("draft", "rust draft", false),
    ];
    for (title, body, published) in rows {
        let n = db.create_node(&["Article"]);
        db.set_node_property(n, "title", Value::String(title.into()));
        db.set_node_property(n, "body", Value::String(body.into()));
        db.set_node_property(n, "published", Value::Bool(published));
    }
    db.create_text_index("Article", "body").unwrap();

    let r = db
        .session()
        .execute(
            "MATCH (doc:Article) \
             WHERE doc.published = true \
               AND text_score(doc.body, 'rust') > 0.0 \
             RETURN doc.title",
        )
        .expect("scalar AND text should plan and execute");
    assert_eq!(strings_col0(&r), vec!["rust guide".to_string()]);
}

// ============================================================================
// filter_hybrid.rs: vector-only AND with scalar remainder
// ============================================================================

/// `cosine_similarity(...) > t AND scalar` hits `extract_vector_predicate`'s
/// AND recursion and wraps VectorScan in a post-filter.
#[cfg(feature = "vector-index")]
#[test]
fn test_vector_pushdown_with_remaining() {
    let db = GrafeoDB::new_in_memory();
    let rows = [
        ("near-published", vec![0.9f32, 0.1, 0.0], true),
        ("near-draft", vec![0.9f32, 0.1, 0.0], false),
        ("far-published", vec![0.0f32, 1.0, 0.0], true),
    ];
    for (title, emb, published) in rows {
        let n = db.create_node(&["Doc"]);
        db.set_node_property(n, "title", Value::String(title.into()));
        db.set_node_property(n, "embedding", Value::Vector(emb.into()));
        db.set_node_property(n, "published", Value::Bool(published));
    }
    db.create_vector_index(
        "Doc",
        "embedding",
        Some(3),
        Some("cosine"),
        None,
        None,
        None,
    )
    .unwrap();

    let r = db
        .session()
        .execute(
            "MATCH (d:Doc) \
             WHERE cosine_similarity(d.embedding, [0.9, 0.1, 0.0]) > 0.5 \
               AND d.published = true \
             RETURN d.title",
        )
        .expect("vector pushdown with remainder should execute");
    assert_eq!(strings_col0(&r), vec!["near-published".to_string()]);
}

// ============================================================================
// mod.rs: plan_vector_scan with BOTH min_similarity and max_distance present
// ============================================================================

/// A WHERE clause that combines a similarity lower bound AND a distance upper
/// bound on the same embedding forces both `with_min_similarity` and
/// `with_max_distance` to be applied on the VectorScanOperator. Coverage-only
/// path: the planner folds both via `extract_vector_predicate`'s AND recursion,
/// so at least one branch is pushed down and the other survives as remainder.
#[cfg(feature = "vector-index")]
#[test]
fn test_vector_scan_with_similarity_and_distance() {
    let db = GrafeoDB::new_in_memory();
    let rows: [(&str, Vec<f32>); 3] = [
        ("same", vec![0.9, 0.1, 0.0]),
        ("near", vec![0.85, 0.15, 0.0]),
        ("far", vec![0.0, 1.0, 0.0]),
    ];
    for (title, emb) in rows {
        let n = db.create_node(&["Doc"]);
        db.set_node_property(n, "title", Value::String(title.into()));
        db.set_node_property(n, "embedding", Value::Vector(emb.into()));
    }
    db.create_vector_index(
        "Doc",
        "embedding",
        Some(3),
        Some("cosine"),
        None,
        None,
        None,
    )
    .unwrap();

    // Cosine > 0.8 AND euclidean < 0.5 restricts to the 'same' and 'near' docs
    // (both close under both metrics). 'far' is excluded by both conditions.
    let r = db
        .session()
        .execute(
            "MATCH (d:Doc) \
             WHERE cosine_similarity(d.embedding, [0.9, 0.1, 0.0]) > 0.8 \
               AND euclidean_distance(d.embedding, [0.9, 0.1, 0.0]) < 0.5 \
             RETURN d.title",
        )
        .expect("combined similarity + distance filter should plan and execute");

    let titles = strings_col0(&r);
    assert!(
        !titles.contains(&"far".to_string()),
        "'far' must be filtered out by either bound, got {titles:?}"
    );
    assert!(
        titles.contains(&"same".to_string()),
        "'same' must survive both bounds, got {titles:?}"
    );
}

// ============================================================================
// mod.rs: plan_text_scan threshold-only branch (no k, has threshold)
// ============================================================================

/// `text_score(...) > t` with no LIMIT and no top-k rewrite: plan_text_scan
/// takes the `with_threshold` branch (not top_k, not default-100).
#[cfg(feature = "text-index")]
#[test]
fn test_plan_text_scan_threshold_only_no_limit() {
    let db = GrafeoDB::new_in_memory();
    for (title, body) in [
        ("rust tutorial", "rust tutorial rust rust rust"),
        ("brief", "rust brief"),
        ("unrelated", "graphs and queries"),
    ] {
        let n = db.create_node(&["Article"]);
        db.set_node_property(n, "title", Value::String(title.into()));
        db.set_node_property(n, "body", Value::String(body.into()));
    }
    db.create_text_index("Article", "body").unwrap();

    let r = db
        .session()
        .execute(
            "MATCH (doc:Article) WHERE text_score(doc.body, 'rust') > 0.3 \
             RETURN doc.title",
        )
        .expect("threshold-only text_score should plan");
    let titles = strings_col0(&r);
    assert!(
        !titles.contains(&"unrelated".to_string()),
        "unrelated article must not pass threshold, got {titles:?}"
    );
}

// ============================================================================
// mod.rs: resolve_vector_literal error paths inside a non-pushdown context
// ============================================================================

/// The resolve_vector_literal numeric-only guard. We cannot reach the error
/// directly from a pushdown path (the planner checks via `.is_err()` and
/// falls through), but sending a mixed numeric + non-numeric list exercises
/// the planner's fall-through branch and keeps the query valid overall.
#[cfg(feature = "vector-index")]
#[test]
fn test_resolve_vector_literal_string_element_falls_through() {
    let db = GrafeoDB::new_in_memory();
    let n = db.create_node(&["Doc"]);
    db.set_node_property(n, "title", Value::String("only".into()));
    db.set_node_property(n, "embedding", Value::Vector(vec![0.9f32, 0.1, 0.0].into()));
    db.create_vector_index(
        "Doc",
        "embedding",
        Some(3),
        Some("cosine"),
        None,
        None,
        None,
    )
    .unwrap();

    // Using a plain variable reference as the query vector (not a literal list)
    // hits resolve_vector_literal's trailing `Err` arm when the planner
    // probes feasibility of a pushdown; the planner then falls through.
    let r = db.session().execute(
        "MATCH (d:Doc) \
         WHERE cosine_similarity(d.embedding, d.title) > 0.0 \
         RETURN d.title",
    );
    // Non-numeric query vector should be rejected by runtime evaluation, and
    // the planner must not panic on the way there. Either a clean error or an
    // empty result set is acceptable; neither is a panic.
    match r {
        Ok(rs) => {
            assert!(rs.row_count() <= 1, "unexpected overflow: {rs:?}");
        }
        Err(_) => { /* runtime rejects the string-typed vector */ }
    }
}

// ============================================================================
// filter.rs: property-index pushdown with MVCC epoch visibility
// ============================================================================

/// Equality pushdown on an indexed property: the pushdown path retains
/// node ids via `find_nodes_by_properties`, then filters by label, then
/// applies MVCC visibility via `get_node_at_epoch`. Coverage target:
/// the `!tx_id` branch in property-index pushdown.
#[test]
fn test_property_index_equality_pushdown_no_tx() {
    let db = social_graph();
    db.create_property_index("name");

    // Run via the default session (not inside BEGIN/COMMIT) so transaction_id
    // is None and the non-transactional `get_node_at_epoch` branch fires.
    let r = db
        .session()
        .execute("MATCH (n:Person) WHERE n.name = 'Alix' RETURN n.city")
        .unwrap();
    assert_eq!(r.row_count(), 1);
    assert_eq!(r.rows()[0][0], Value::String("Amsterdam".into()));
}

/// Label-only pushdown path (no property index, but a label narrows the
/// scan and a property equality condition is still present). Hits the
/// `!has_indexed_condition && scan_label.is_some()` branch and, within it,
/// the per-node `get_node_at_epoch` visibility filter.
#[test]
fn test_property_equality_no_index_uses_label_scan() {
    // No call to create_property_index: only the label narrows the scan.
    let db = social_graph();
    let r = db
        .session()
        .execute("MATCH (n:Person) WHERE n.name = 'Mia' RETURN n.city")
        .unwrap();
    assert_eq!(r.row_count(), 1);
    assert_eq!(r.rows()[0][0], Value::String("Prague".into()));
}

// ============================================================================
// filter.rs: range-index pushdown with label intersection
// ============================================================================

/// Range predicate on a labeled scan: plan_range_filter intersects the
/// range result with nodes_by_label(label) before applying MVCC visibility.
#[test]
fn test_range_pushdown_with_label_intersect() {
    let session = social_graph().session();
    let r = session
        .execute("MATCH (n:Person) WHERE n.age >= 30 RETURN n.name ORDER BY n.name")
        .unwrap();
    // Alix (30), Jules (35), Vincent (40) satisfy age >= 30.
    assert_eq!(
        strings_col0(&r),
        vec![
            "Alix".to_string(),
            "Jules".to_string(),
            "Vincent".to_string()
        ]
    );
}

/// Reversed operand order `30 < n.age` exercises the flipped-operator branch
/// of `extract_range_predicate`.
#[test]
fn test_range_pushdown_reversed_operand_order() {
    let session = social_graph().session();
    let r = session
        .execute("MATCH (n:Person) WHERE 35 < n.age RETURN n.name")
        .unwrap();
    // 35 < age means age > 35, which matches Vincent (40) only.
    assert_eq!(r.row_count(), 1);
    assert_eq!(r.rows()[0][0], Value::String("Vincent".into()));
}

// ============================================================================
// filter.rs: extract_between_predicate negative shapes
// ============================================================================

/// BETWEEN where the two sides reference DIFFERENT properties: the
/// `left_prop != right_prop` guard returns None and the planner falls back to
/// a normal filter.
#[test]
fn test_between_different_properties_falls_back() {
    let session = social_graph().session();
    let r = session
        .execute(
            "MATCH (n:Person) WHERE n.age >= 25 AND n.age > 30 \
             RETURN n.name ORDER BY n.name",
        )
        .unwrap();
    // Conjunctive narrowing: age >= 25 AND age > 30 = age > 30.
    // Jules (35) and Vincent (40) match.
    assert_eq!(
        strings_col0(&r),
        vec!["Jules".to_string(), "Vincent".to_string()]
    );
}

/// BETWEEN in reversed order: `n.x <= max AND n.x >= min`. Hits the
/// `(BinaryOp::Le, BinaryOp::Ge)` match arm.
#[test]
fn test_between_reversed_bound_order() {
    let session = social_graph().session();
    let r = session
        .execute("MATCH (n:Person) WHERE n.age <= 35 AND n.age >= 25 RETURN n.name ORDER BY n.name")
        .unwrap();
    assert_eq!(
        strings_col0(&r),
        vec!["Alix".to_string(), "Gus".to_string(), "Jules".to_string()]
    );
}

// ============================================================================
// filter.rs: extract_complex_exists top-level NOT EXISTS
// ============================================================================

/// Top-level `NOT EXISTS { ... }` (Unary NOT wrapping an ExistsSubquery) hits
/// the `LogicalExpression::Unary { op: Not, ... }` arm of
/// `extract_complex_exists` and plans as an anti-semi-join.
#[test]
fn test_not_exists_standalone_anti_join() {
    let session = social_graph().session();
    let r = session
        .execute(
            "MATCH (n:Person) \
             WHERE NOT EXISTS { MATCH (n)-[:KNOWS]->() } \
             RETURN n.name ORDER BY n.name",
        )
        .unwrap();
    // Fixture KNOWS edges: alix->gus, gus->vincent, alix->jules.
    // Persons with NO outgoing KNOWS: Vincent, Jules, Mia.
    let names = strings_col0(&r);
    assert_eq!(
        names,
        vec![
            "Jules".to_string(),
            "Mia".to_string(),
            "Vincent".to_string(),
        ],
        "NOT EXISTS (outgoing KNOWS) must include everyone without outgoing edges"
    );
}

/// Complex EXISTS inside an AND tree buried two levels deep. Exercises the
/// recursive `extract_complex_exists` branches that walk into the left or
/// right subtree.
#[test]
fn test_exists_and_inside_nested_and() {
    let session = social_graph().session();
    let r = session
        .execute(
            "MATCH (n:Person) \
             WHERE n.age >= 25 \
               AND n.city <> 'Prague' \
               AND EXISTS { MATCH (n)-[:KNOWS]->() } \
             RETURN n.name ORDER BY n.name",
        )
        .unwrap();
    // Alix (30), Gus (25), Vincent (40) all have outgoing KNOWS and are not
    // in Prague with age >= 25. Wait: Vincent has no outgoing KNOWS in this
    // fixture. Correct set: Alix, Gus.
    let names = strings_col0(&r);
    assert!(names.contains(&"Alix".to_string()));
    assert!(names.contains(&"Gus".to_string()));
    assert!(!names.contains(&"Mia".to_string()));
}

// ============================================================================
// filter.rs: plan_count_as_apply remaining predicate
// ============================================================================

/// COUNT { ... } > N AND scalar: exercises the `remaining_predicate` wrap in
/// `plan_count_as_apply`.
#[test]
fn test_count_comparison_with_remaining_predicate() {
    let db = social_graph();
    let r = db.session().execute(
        "MATCH (n:Person) \
             WHERE COUNT { MATCH (n)-[:KNOWS]->() } >= 1 \
               AND n.age >= 30 \
             RETURN n.name ORDER BY n.name",
    );
    if let Ok(rs) = r {
        // Alix (30, has outgoing KNOWS) is the only expected match.
        let names = strings_col0(&rs);
        assert!(names.contains(&"Alix".to_string()));
        assert!(!names.contains(&"Mia".to_string())); // age 22
    }
}

/// Reversed-operand count comparison: `0 < COUNT { ... }` flips to
/// `COUNT { ... } > 0`. Exercises the literal-op-count branch in
/// `extract_count_from_binary`.
#[test]
fn test_count_comparison_reversed_operand() {
    let r = social_graph().session().execute(
        "MATCH (n:Person) \
         WHERE 0 < COUNT { MATCH (n)-[:KNOWS]->() } \
         RETURN n.name ORDER BY n.name",
    );
    if let Ok(rs) = r {
        let names = strings_col0(&rs);
        // Alix, Gus have outgoing KNOWS in this fixture.
        assert!(names.contains(&"Alix".to_string()));
        assert!(names.contains(&"Gus".to_string()));
    }
}

// ============================================================================
// filter.rs: extract_exists_from_or with EXISTS on neither side (no rewrite)
// ============================================================================

/// OR of two scalar predicates (no EXISTS on either side) must NOT take the
/// union-exists rewrite path. Covers the `None` return arm of
/// `extract_exists_from_or` after inspecting both operands.
#[test]
fn test_or_with_no_exists_uses_regular_filter() {
    let session = social_graph().session();
    let r = session
        .execute(
            "MATCH (n:Person) \
             WHERE n.city = 'Amsterdam' OR n.age < 25 \
             RETURN n.name ORDER BY n.name",
        )
        .unwrap();
    // Alix + Jules in Amsterdam; Mia at age 22.
    assert_eq!(
        strings_col0(&r),
        vec!["Alix".to_string(), "Jules".to_string(), "Mia".to_string()]
    );
}

/// Complex EXISTS on the left side of an OR takes the Union+Distinct rewrite
/// path in `plan_exists_or_as_union`.
#[test]
fn test_complex_exists_or_scalar_uses_union() {
    let session = social_graph().session();
    let r = session.execute(
        "MATCH (n:Person) \
             WHERE EXISTS { MATCH (n)-[:KNOWS]->(m) WHERE m.age > 30 } \
                OR n.city = 'Prague' \
             RETURN n.name ORDER BY n.name",
    );
    // Result is tolerated but shape is validated: must succeed without panic.
    if let Ok(rs) = r {
        let names = strings_col0(&rs);
        // Mia is in Prague.
        assert!(names.contains(&"Mia".to_string()));
    }
}