nodedb-query 0.6.1

// SPDX-License-Identifier: Apache-2.0

//! Binary filter evaluation on raw MessagePack documents.
//!
//! `ScanFilter::matches_binary(doc: &[u8])` evaluates a filter predicate
//! directly on msgpack bytes without decoding to `serde_json::Value`.
//! Uses `Value::eq_coerced`/`partial_cmp_coerced` for type coercion — single
//! source of truth shared with the JSON filter path.

use std::cmp::Ordering;

use crate::expr::EvalError;
use crate::msgpack_scan::field::extract_field;
use crate::msgpack_scan::index::FieldIndex;
use crate::msgpack_scan::reader::{
    array_header, map_header, read_null, read_str, read_value, skip_value,
};
use crate::scan_filter::like::sql_like_match;
use crate::scan_filter::types::column_compare;
use crate::scan_filter::{FilterOp, ScanFilter};

impl ScanFilter {
    /// Evaluate an AND-group of filters, short-circuiting on the first
    /// `false` or the first evaluation error — same semantics as
    /// `group.iter().all(|f| f.matches_binary(doc))` had before filter
    /// evaluation could fail. A `pub` associated
    /// function (rather than a private free function) so the many call
    /// sites across the `nodedb` crate that would otherwise write
    /// `filters.iter().all(|f| f.matches_binary(doc))` have a single
    /// drop-in replacement instead of each hand-rolling the same
    /// short-circuit loop.
    pub fn all_match_binary(group: &[ScanFilter], doc: &[u8]) -> Result<bool, EvalError> {
        for f in group {
            if !f.matches_binary(doc)? {
                return Ok(false);
            }
        }
        Ok(true)
    }

    /// Indexed counterpart of [`ScanFilter::all_match_binary`].
    pub fn all_match_binary_indexed(
        group: &[ScanFilter],
        doc: &[u8],
        idx: &FieldIndex,
    ) -> Result<bool, EvalError> {
        for f in group {
            if !f.matches_binary_indexed(doc, idx)? {
                return Ok(false);
            }
        }
        Ok(true)
    }
}

impl ScanFilter {
    /// Evaluate this filter against a raw MessagePack document.
    ///
    /// Zero deserialization — extracts only the needed field bytes.
    ///
    /// Returns `Err(EvalError::DivisionByZero)` when this is (or contains,
    /// via an `OR` group) a `FilterOp::Expr` predicate that divides or
    /// takes a modulus by zero — see
    /// `scan_filter::ScanFilter::matches_value` for the WHERE-clause
    /// behavior-flip rationale, which applies identically here.
    pub fn matches_binary(&self, doc: &[u8]) -> Result<bool, EvalError> {
        match self.op {
            FilterOp::MatchAll | FilterOp::Exists | FilterOp::NotExists => return Ok(true),
            FilterOp::Or => {
                for clause in &self.clauses {
                    if Self::all_match_binary(clause, doc)? {
                        return Ok(true);
                    }
                }
                return Ok(false);
            }
            FilterOp::Expr => {
                return match (self.expr.as_ref(), nodedb_types::value_from_msgpack(doc)) {
                    (Some(expr), Ok(value)) => Ok(crate::value_ops::is_truthy(&expr.eval(&value)?)),
                    _ => Ok(false),
                };
            }
            FilterOp::Eq
            | FilterOp::Ne
            | FilterOp::Gt
            | FilterOp::Gte
            | FilterOp::Lt
            | FilterOp::Lte
            | FilterOp::Contains
            | FilterOp::Like
            | FilterOp::NotLike
            | FilterOp::Ilike
            | FilterOp::NotIlike
            | FilterOp::In
            | FilterOp::NotIn
            | FilterOp::IsNull
            | FilterOp::IsNotNull
            | FilterOp::ArrayContains
            | FilterOp::ArrayContainsAll
            | FilterOp::ArrayOverlap
            | FilterOp::GtColumn
            | FilterOp::GteColumn
            | FilterOp::LtColumn
            | FilterOp::LteColumn
            | FilterOp::EqColumn
            | FilterOp::NeColumn => {}
        }

        let (start, end) = match extract_field(doc, 0, &self.field) {
            Some(r) => r,
            None => {
                // Qualified-name fallback: "amount" might be stored as "orders.amount".
                let suffix = format!(".{}", self.field);
                match find_field_by_suffix(doc, &suffix) {
                    Some(r) => r,
                    None => return Ok(self.op == FilterOp::IsNull),
                }
            }
        };

        Ok(eval_op(self, doc, start, end))
    }

    /// Evaluate using a pre-built `FieldIndex` for O(1) field lookup.
    ///
    /// Use when evaluating multiple predicates on the same document.
    pub fn matches_binary_indexed(&self, doc: &[u8], idx: &FieldIndex) -> Result<bool, EvalError> {
        match self.op {
            FilterOp::MatchAll | FilterOp::Exists | FilterOp::NotExists => return Ok(true),
            FilterOp::Or => {
                for clause in &self.clauses {
                    if Self::all_match_binary_indexed(clause, doc, idx)? {
                        return Ok(true);
                    }
                }
                return Ok(false);
            }
            FilterOp::Expr => {
                return match (self.expr.as_ref(), nodedb_types::value_from_msgpack(doc)) {
                    (Some(expr), Ok(value)) => Ok(crate::value_ops::is_truthy(&expr.eval(&value)?)),
                    _ => Ok(false),
                };
            }
            FilterOp::Eq
            | FilterOp::Ne
            | FilterOp::Gt
            | FilterOp::Gte
            | FilterOp::Lt
            | FilterOp::Lte
            | FilterOp::Contains
            | FilterOp::Like
            | FilterOp::NotLike
            | FilterOp::Ilike
            | FilterOp::NotIlike
            | FilterOp::In
            | FilterOp::NotIn
            | FilterOp::IsNull
            | FilterOp::IsNotNull
            | FilterOp::ArrayContains
            | FilterOp::ArrayContainsAll
            | FilterOp::ArrayOverlap
            | FilterOp::GtColumn
            | FilterOp::GteColumn
            | FilterOp::LtColumn
            | FilterOp::LteColumn
            | FilterOp::EqColumn
            | FilterOp::NeColumn => {}
        }

        // Same lookup order as `matches_binary`: the exact key, then a
        // qualified key carrying the bare name. Absent, and so NULL, only
        // when neither exists.
        let (start, end) = match idx
            .get(&self.field)
            .or_else(|| find_field_by_suffix(doc, &format!(".{}", self.field)))
        {
            Some(r) => r,
            None => return Ok(self.op == FilterOp::IsNull),
        };

        Ok(eval_op(self, doc, start, end))
    }
}

/// Shared filter op evaluation — used by both `matches_binary` and `matches_binary_indexed`.
fn eval_op(filter: &ScanFilter, doc: &[u8], start: usize, _end: usize) -> bool {
    match filter.op {
        FilterOp::IsNull => read_null(doc, start),
        FilterOp::IsNotNull => !read_null(doc, start),
        FilterOp::Eq => eq_value(doc, start, &filter.value),
        FilterOp::Ne => !eq_value(doc, start, &filter.value),
        FilterOp::Gt => cmp_value(doc, start, &filter.value) == Some(Ordering::Greater),
        FilterOp::Gte => matches!(
            cmp_value(doc, start, &filter.value),
            Some(Ordering::Greater | Ordering::Equal)
        ),
        FilterOp::Lt => cmp_value(doc, start, &filter.value) == Some(Ordering::Less),
        FilterOp::Lte => matches!(
            cmp_value(doc, start, &filter.value),
            Some(Ordering::Less | Ordering::Equal)
        ),
        FilterOp::Contains => {
            if let (Some(s), Some(pattern)) = (read_str(doc, start), filter.value.as_str()) {
                s.contains(pattern)
            } else {
                false
            }
        }
        FilterOp::Like => str_match(doc, start, &filter.value, false, false),
        FilterOp::NotLike => str_match(doc, start, &filter.value, false, true),
        FilterOp::Ilike => str_match(doc, start, &filter.value, true, false),
        FilterOp::NotIlike => str_match(doc, start, &filter.value, true, true),
        FilterOp::In => {
            if let Some(mut iter) = filter.value.as_array_iter() {
                iter.any(|v| eq_value(doc, start, v))
            } else {
                false
            }
        }
        FilterOp::NotIn => {
            if let Some(mut iter) = filter.value.as_array_iter() {
                !iter.any(|v| eq_value(doc, start, v))
            } else {
                true
            }
        }
        FilterOp::ArrayContains => array_any(doc, start, |elem_start| {
            eq_value(doc, elem_start, &filter.value)
        }),
        FilterOp::ArrayContainsAll => {
            if let Some(mut needles) = filter.value.as_array_iter() {
                needles.all(|needle| {
                    array_any(doc, start, |elem_start| eq_value(doc, elem_start, needle))
                })
            } else {
                false
            }
        }
        FilterOp::ArrayOverlap => {
            if let Some(mut needles) = filter.value.as_array_iter() {
                needles.any(|needle| {
                    array_any(doc, start, |elem_start| eq_value(doc, elem_start, needle))
                })
            } else {
                false
            }
        }
        // Column-vs-column: extract right-side field from the same doc.
        FilterOp::GtColumn
        | FilterOp::GteColumn
        | FilterOp::LtColumn
        | FilterOp::LteColumn
        | FilterOp::EqColumn
        | FilterOp::NeColumn => {
            let other_col = match &filter.value {
                nodedb_types::Value::String(s) => s.as_str(),
                _ => return false,
            };
            // Try exact field name, then qualified suffix match.
            let other_range = extract_field(doc, 0, other_col).or_else(|| {
                let suffix = format!(".{other_col}");
                find_field_by_suffix(doc, &suffix)
            });
            let Some((other_start, _)) = other_range else {
                return false;
            };
            // Read both sides as Value and compare.
            let left = read_value(doc, start).unwrap_or(nodedb_types::Value::Null);
            let right = read_value(doc, other_start).unwrap_or(nodedb_types::Value::Null);
            column_compare(filter.op, &left, &right)
        }
        // Handled before field extraction by both `matches_binary` paths.
        FilterOp::MatchAll | FilterOp::Exists | FilterOp::NotExists => true,
        FilterOp::Or | FilterOp::Expr => false,
    }
}

/// Find a field in a msgpack map by suffix match (e.g., ".amount" matches "orders.amount").
fn find_field_by_suffix(doc: &[u8], suffix: &str) -> Option<(usize, usize)> {
    let (count, mut pos) = map_header(doc, 0)?;
    for _ in 0..count {
        let key = read_str(doc, pos);
        let key_end = skip_value(doc, pos)?;
        let val_start = key_end;
        let val_end = skip_value(doc, val_start)?;
        if let Some(k) = key
            && k.ends_with(suffix)
        {
            return Some((val_start, val_end));
        }
        pos = val_end;
    }
    None
}

// ── Helpers ────────────────────────────────────────────────────────────

/// Coerced equality: read msgpack value at offset → compare with `Value`.
/// Uses `Value::eq_coerced` — single source of truth for type coercion.
#[inline]
fn eq_value(buf: &[u8], offset: usize, filter_val: &nodedb_types::Value) -> bool {
    if read_null(buf, offset) {
        return filter_val.is_null();
    }
    match read_value(buf, offset) {
        Some(field_val) => filter_val.eq_coerced(&field_val),
        None => false,
    }
}

/// Coerced ordering: read msgpack value at offset → compare with `Value`.
/// Uses `Value::partial_cmp_coerced` — single source of truth for ordering.
///
/// Returns ordering of field_val relative to filter_val (field <=> filter),
/// or `None` when the cell cannot be read or the pair has no defined order,
/// so a range predicate over it matches nothing.
#[inline]
fn cmp_value(buf: &[u8], offset: usize, filter_val: &nodedb_types::Value) -> Option<Ordering> {
    read_value(buf, offset)?.partial_cmp_coerced(filter_val)
}

/// LIKE/ILIKE/NOT LIKE/NOT ILIKE helper.
#[inline]
fn str_match(
    buf: &[u8],
    offset: usize,
    pattern_val: &nodedb_types::Value,
    icase: bool,
    negate: bool,
) -> bool {
    let result = if let (Some(s), Some(pattern)) = (read_str(buf, offset), pattern_val.as_str()) {
        sql_like_match(s, pattern, icase)
    } else {
        false
    };
    if negate { !result } else { result }
}

/// Iterate msgpack array elements, return true if any satisfies predicate.
fn array_any(buf: &[u8], start: usize, mut pred: impl FnMut(usize) -> bool) -> bool {
    let Some((count, mut pos)) = array_header(buf, start) else {
        return false;
    };
    for _ in 0..count {
        if pred(pos) {
            return true;
        }
        let Some(next) = skip_value(buf, pos) else {
            return false;
        };
        pos = next;
    }
    false
}

#[cfg(test)]
mod tests {
    use super::*;
    use serde_json::json;

    fn encode(v: &serde_json::Value) -> Vec<u8> {
        nodedb_types::json_msgpack::json_to_msgpack(v).expect("encode")
    }

    fn filter(field: &str, op: &str, value: nodedb_types::Value) -> ScanFilter {
        ScanFilter {
            field: field.into(),
            op: FilterOp::parse(op).expect(op),
            value,
            clauses: vec![],
            expr: None,
        }
    }

    #[test]
    fn eq_integer() {
        let doc = encode(&json!({"age": 25}));
        assert!(
            filter("age", "eq", nodedb_types::Value::Integer(25))
                .matches_binary(&doc)
                .unwrap()
        );
        assert!(
            !filter("age", "eq", nodedb_types::Value::Integer(30))
                .matches_binary(&doc)
                .unwrap()
        );
    }

    /// An instant cell compares by epoch microseconds against a typed
    /// instant literal and against an ISO-8601 string literal.
    #[test]
    fn instant_cell_compares_by_micros() {
        use nodedb_types::{InstantKind, NdbDateTime, Value};
        const MICROS: i64 = 1_583_402_400_000_000;
        let mut doc = Vec::new();
        crate::msgpack_scan::write_map_header(&mut doc, 1);
        crate::msgpack_scan::write_kv_instant(&mut doc, "ts", InstantKind::Naive, MICROS);

        let same = Value::NaiveDateTime(NdbDateTime::from_micros(MICROS));
        let later = Value::NaiveDateTime(NdbDateTime::from_micros(MICROS + 1));
        assert!(
            filter("ts", "eq", same.clone())
                .matches_binary(&doc)
                .unwrap()
        );
        assert!(
            !filter("ts", "eq", later.clone())
                .matches_binary(&doc)
                .unwrap()
        );
        assert!(filter("ts", "lt", later).matches_binary(&doc).unwrap());
        assert!(filter("ts", "gte", same).matches_binary(&doc).unwrap());
        assert!(
            filter("ts", "eq", Value::String("2020-03-05T10:00:00Z".into()))
                .matches_binary(&doc)
                .unwrap()
        );
        assert!(
            filter("ts", "gt", Value::String("2020-03-05 09:00:00".into()))
                .matches_binary(&doc)
                .unwrap()
        );
        let idx = FieldIndex::build(&doc, 0).unwrap_or_else(FieldIndex::empty);
        assert!(
            filter(
                "ts",
                "eq",
                Value::NaiveDateTime(NdbDateTime::from_micros(MICROS))
            )
            .matches_binary_indexed(&doc, &idx)
            .unwrap()
        );
    }

    #[test]
    fn eq_coerces_string_to_integer() {
        let doc = encode(&json!({"age": 25}));
        assert!(
            filter("age", "eq", nodedb_types::Value::String("25".into()))
                .matches_binary(&doc)
                .unwrap()
        );
    }

    #[test]
    fn gt_coerces_string_to_integer() {
        let doc = encode(&json!({"score": "90"}));
        assert!(
            filter("score", "gt", nodedb_types::Value::Integer(80))
                .matches_binary(&doc)
                .unwrap()
        );
    }

    #[test]
    fn eq_string() {
        let doc = encode(&json!({"name": "alice"}));
        assert!(
            filter("name", "eq", nodedb_types::Value::String("alice".into()))
                .matches_binary(&doc)
                .unwrap()
        );
    }

    #[test]
    fn eq_coercion_int_vs_string() {
        let doc = encode(&json!({"age": 25}));
        assert!(
            filter("age", "eq", nodedb_types::Value::String("25".into()))
                .matches_binary(&doc)
                .unwrap()
        );
    }

    #[test]
    fn eq_coercion_string_vs_int() {
        let doc = encode(&json!({"score": "90"}));
        assert!(
            filter("score", "eq", nodedb_types::Value::Integer(90))
                .matches_binary(&doc)
                .unwrap()
        );
    }

    #[test]
    fn ne() {
        let doc = encode(&json!({"x": 1}));
        assert!(
            filter("x", "ne", nodedb_types::Value::Integer(2))
                .matches_binary(&doc)
                .unwrap()
        );
        assert!(
            !filter("x", "ne", nodedb_types::Value::Integer(1))
                .matches_binary(&doc)
                .unwrap()
        );
    }

    #[test]
    fn gt_lt() {
        let doc = encode(&json!({"v": 10}));
        assert!(
            filter("v", "gt", nodedb_types::Value::Integer(5))
                .matches_binary(&doc)
                .unwrap()
        );
        assert!(
            !filter("v", "gt", nodedb_types::Value::Integer(15))
                .matches_binary(&doc)
                .unwrap()
        );
        assert!(
            filter("v", "lt", nodedb_types::Value::Integer(15))
                .matches_binary(&doc)
                .unwrap()
        );
        assert!(
            !filter("v", "lt", nodedb_types::Value::Integer(5))
                .matches_binary(&doc)
                .unwrap()
        );
    }

    #[test]
    fn gte_lte() {
        let doc = encode(&json!({"v": 10}));
        assert!(
            filter("v", "gte", nodedb_types::Value::Integer(10))
                .matches_binary(&doc)
                .unwrap()
        );
        assert!(
            filter("v", "gte", nodedb_types::Value::Integer(5))
                .matches_binary(&doc)
                .unwrap()
        );
        assert!(
            !filter("v", "gte", nodedb_types::Value::Integer(15))
                .matches_binary(&doc)
                .unwrap()
        );
        assert!(
            filter("v", "lte", nodedb_types::Value::Integer(10))
                .matches_binary(&doc)
                .unwrap()
        );
    }

    #[test]
    fn is_null_not_null() {
        let doc = encode(&json!({"a": null, "b": 1}));
        assert!(
            filter("a", "is_null", nodedb_types::Value::Null)
                .matches_binary(&doc)
                .unwrap()
        );
        assert!(
            !filter("b", "is_null", nodedb_types::Value::Null)
                .matches_binary(&doc)
                .unwrap()
        );
        assert!(
            filter("b", "is_not_null", nodedb_types::Value::Null)
                .matches_binary(&doc)
                .unwrap()
        );
    }

    #[test]
    fn missing_field_is_null() {
        let doc = encode(&json!({"x": 1}));
        assert!(
            filter("missing", "is_null", nodedb_types::Value::Null)
                .matches_binary(&doc)
                .unwrap()
        );
    }

    #[test]
    fn contains_str() {
        let doc = encode(&json!({"msg": "hello world"}));
        assert!(
            filter(
                "msg",
                "contains",
                nodedb_types::Value::String("world".into())
            )
            .matches_binary(&doc)
            .unwrap()
        );
    }

    #[test]
    fn like_ilike() {
        let doc = encode(&json!({"name": "Alice"}));
        assert!(
            filter("name", "like", nodedb_types::Value::String("Ali%".into()))
                .matches_binary(&doc)
                .unwrap()
        );
        assert!(
            !filter("name", "like", nodedb_types::Value::String("ali%".into()))
                .matches_binary(&doc)
                .unwrap()
        );
        assert!(
            filter("name", "ilike", nodedb_types::Value::String("ali%".into()))
                .matches_binary(&doc)
                .unwrap()
        );
        assert!(
            filter(
                "name",
                "not_like",
                nodedb_types::Value::String("Bob%".into())
            )
            .matches_binary(&doc)
            .unwrap()
        );
    }

    #[test]
    fn in_not_in() {
        let doc = encode(&json!({"status": "active"}));
        let vals = nodedb_types::Value::Array(vec![
            nodedb_types::Value::String("active".into()),
            nodedb_types::Value::String("pending".into()),
        ]);
        assert!(
            ScanFilter {
                field: "status".into(),
                op: FilterOp::In,
                value: vals.clone(),
                clauses: vec![],
                expr: None
            }
            .matches_binary(&doc)
            .unwrap()
        );

        let doc2 = encode(&json!({"status": "deleted"}));
        assert!(
            ScanFilter {
                field: "status".into(),
                op: FilterOp::NotIn,
                value: vals,
                clauses: vec![],
                expr: None
            }
            .matches_binary(&doc2)
            .unwrap()
        );
    }

    #[test]
    fn array_contains() {
        let doc = encode(&json!({"tags": ["rust", "db", "fast"]}));
        assert!(
            filter(
                "tags",
                "array_contains",
                nodedb_types::Value::String("rust".into())
            )
            .matches_binary(&doc)
            .unwrap()
        );
        assert!(
            !filter(
                "tags",
                "array_contains",
                nodedb_types::Value::String("slow".into())
            )
            .matches_binary(&doc)
            .unwrap()
        );
    }

    #[test]
    fn array_contains_all() {
        let doc = encode(&json!({"tags": ["a", "b", "c"]}));
        let needles = nodedb_types::Value::Array(vec![
            nodedb_types::Value::String("a".into()),
            nodedb_types::Value::String("c".into()),
        ]);
        assert!(
            ScanFilter {
                field: "tags".into(),
                op: FilterOp::ArrayContainsAll,
                value: needles,
                clauses: vec![],
                expr: None
            }
            .matches_binary(&doc)
            .unwrap()
        );
    }

    #[test]
    fn array_overlap() {
        let doc = encode(&json!({"tags": ["x", "y"]}));
        let needles = nodedb_types::Value::Array(vec![
            nodedb_types::Value::String("y".into()),
            nodedb_types::Value::String("z".into()),
        ]);
        assert!(
            ScanFilter {
                field: "tags".into(),
                op: FilterOp::ArrayOverlap,
                value: needles,
                clauses: vec![],
                expr: None
            }
            .matches_binary(&doc)
            .unwrap()
        );
    }

    #[test]
    fn or_clauses() {
        let doc = encode(&json!({"x": 5}));
        let f = ScanFilter {
            field: String::new(),
            op: FilterOp::Or,
            value: nodedb_types::Value::Null,
            clauses: vec![
                vec![filter("x", "eq", nodedb_types::Value::Integer(10))],
                vec![filter("x", "eq", nodedb_types::Value::Integer(5))],
            ],
            expr: None,
        };
        assert!(f.matches_binary(&doc).unwrap());
    }

    #[test]
    fn match_all() {
        let doc = encode(&json!({"any": "thing"}));
        assert!(
            filter("", "match_all", nodedb_types::Value::Null)
                .matches_binary(&doc)
                .unwrap()
        );
    }

    #[test]
    fn float_comparison() {
        let doc = encode(&json!({"temp": 36.6}));
        assert!(
            filter("temp", "gt", nodedb_types::Value::Float(30.0))
                .matches_binary(&doc)
                .unwrap()
        );
        assert!(
            filter("temp", "lt", nodedb_types::Value::Float(40.0))
                .matches_binary(&doc)
                .unwrap()
        );
    }

    #[test]
    fn bool_eq() {
        let doc = encode(&json!({"active": true}));
        assert!(
            filter("active", "eq", nodedb_types::Value::Bool(true))
                .matches_binary(&doc)
                .unwrap()
        );
        assert!(
            !filter("active", "eq", nodedb_types::Value::Bool(false))
                .matches_binary(&doc)
                .unwrap()
        );
    }

    #[test]
    fn gt_coercion_string_field() {
        let doc = encode(&json!({"score": "90"}));
        assert!(
            filter("score", "gt", nodedb_types::Value::Integer(80))
                .matches_binary(&doc)
                .unwrap()
        );
    }

    #[test]
    fn timestamp_upper_bound_matches() {
        let doc = encode(&json!({"t": "2026-07-02T13:00:00.000000Z"}));
        assert!(
            filter(
                "t",
                "lte",
                nodedb_types::Value::String("2026-07-02 14:00:00".into())
            )
            .matches_binary(&doc)
            .unwrap()
        );
        assert!(
            filter(
                "t",
                "lt",
                nodedb_types::Value::String("2026-07-02 14:00:00".into())
            )
            .matches_binary(&doc)
            .unwrap()
        );
        assert!(
            filter(
                "t",
                "gte",
                nodedb_types::Value::String("2026-07-02 12:00:00".into())
            )
            .matches_binary(&doc)
            .unwrap()
        );
        assert!(
            filter(
                "t",
                "eq",
                nodedb_types::Value::String("2026-07-02 13:00:00".into())
            )
            .matches_binary(&doc)
            .unwrap()
        );
    }

    // ── Indexed variant tests ──────────────────────────────────────────

    #[test]
    fn indexed_matches_same_as_sequential() {
        let doc = encode(&json!({"a": 1, "b": "hello", "c": true, "d": null}));
        let idx = FieldIndex::build(&doc, 0).unwrap();

        let filters = vec![
            filter("a", "eq", nodedb_types::Value::Integer(1)),
            filter("b", "contains", nodedb_types::Value::String("ell".into())),
            filter("c", "eq", nodedb_types::Value::Bool(true)),
            filter("d", "is_null", nodedb_types::Value::Null),
            filter("missing", "is_null", nodedb_types::Value::Null),
        ];

        for f in &filters {
            assert_eq!(
                f.matches_binary(&doc).unwrap(),
                f.matches_binary_indexed(&doc, &idx).unwrap(),
                "mismatch for field={} op={:?}",
                f.field,
                f.op
            );
        }
    }
}