kglite 0.16.0

//! DirGraph regression tests extracted from mod.rs.

use super::*;

#[cfg(test)]
mod multi_label_tests {
    use super::*;
    use crate::datatypes::Value;
    use crate::graph::schema::NodeData;
    use crate::graph::storage::GraphWrite;

    fn add_node(graph: &mut DirGraph, id: &str, node_type: &str) -> NodeIndex {
        let nd = NodeData::new(
            Value::String(id.to_string()),
            Value::String(id.to_string()),
            node_type.to_string(),
            HashMap::new(),
            &mut graph.interner,
        );
        let idx = GraphWrite::add_node(&mut graph.graph, nd);
        graph
            .type_indices
            .entry_or_default(node_type.to_string())
            .push(idx);
        idx
    }

    #[test]
    fn add_node_label_idempotent_and_no_op_on_primary() {
        let mut g = DirGraph::new();
        let idx = add_node(&mut g, "n1", "Person");
        let reviewer = g.interner.get_or_intern("Reviewer");
        let person = g.interner.get_or_intern("Person");

        assert!(g.add_node_label(idx, reviewer));
        assert!(g.has_secondary_labels);
        assert_eq!(g.secondary_label_index[&reviewer], vec![idx]);

        // Idempotent — second add is a no-op.
        assert!(!g.add_node_label(idx, reviewer));
        assert_eq!(g.secondary_label_index[&reviewer], vec![idx]);

        // Primary type is a no-op too.
        assert!(!g.add_node_label(idx, person));

        let labels = g.node_labels(idx);
        assert_eq!(labels.len(), 2);
        assert_eq!(labels[0], person);
        assert_eq!(labels[1], reviewer);
    }

    #[test]
    fn remove_node_label_errors_on_primary() {
        let mut g = DirGraph::new();
        let idx = add_node(&mut g, "n1", "Person");
        let person = g.interner.get_or_intern("Person");

        let err = g.remove_node_label(idx, person).unwrap_err();
        assert!(err.contains("primary label"));
    }

    #[test]
    fn remove_node_label_clears_index_when_last_node_drops_it() {
        let mut g = DirGraph::new();
        let a = add_node(&mut g, "a", "Person");
        let b = add_node(&mut g, "b", "Person");
        let reviewer = g.interner.get_or_intern("Reviewer");

        g.add_node_label(a, reviewer);
        g.add_node_label(b, reviewer);
        assert_eq!(g.secondary_label_index[&reviewer].len(), 2);

        assert!(g.remove_node_label(a, reviewer).unwrap());
        assert_eq!(g.secondary_label_index[&reviewer], vec![b]);
        assert!(g.has_secondary_labels);

        assert!(g.remove_node_label(b, reviewer).unwrap());
        assert!(!g.secondary_label_index.contains_key(&reviewer));
        // No labels left anywhere, fast-skip resets.
        assert!(!g.has_secondary_labels);
    }

    #[test]
    fn rebuild_does_not_clobber_secondary_index() {
        // After 0.10.5's perf fix, NodeData no longer carries
        // extra_labels — `secondary_label_index` is the canonical
        // store. `rebuild_type_indices` rebuilds only type_indices
        // and leaves the secondary index intact (it's repopulated by
        // the load path via the disk sidecar / .kgl section).
        let mut g = DirGraph::new();
        let idx = add_node(&mut g, "n1", "Person");
        let reviewer = g.interner.get_or_intern("Reviewer");
        g.add_node_label(idx, reviewer);

        let before = g.secondary_label_index.clone();
        let before_flag = g.has_secondary_labels;

        g.rebuild_type_indices();

        // Secondary index is untouched.
        assert_eq!(g.secondary_label_index, before);
        assert_eq!(g.has_secondary_labels, before_flag);
        // Primary type_indices is rebuilt correctly.
        assert_eq!(
            g.type_indices.get("Person").map(|s| s.iter().collect()),
            Some(vec![idx])
        );
    }

    #[test]
    fn dir_graph_node_labels_returns_primary_plus_extras() {
        // The canonical path for "all labels of node X" is
        // `DirGraph::node_labels` (which scans `secondary_label_index`).
        // Backend trait `node_labels_of` returns only the primary
        // type and is no longer the authoritative source.
        let mut g = DirGraph::new();
        let idx = add_node(&mut g, "n1", "Person");
        let reviewer = g.interner.get_or_intern("Reviewer");
        let person = g.interner.get_or_intern("Person");
        g.add_node_label(idx, reviewer);

        let labels = g.node_labels(idx);
        assert_eq!(labels, vec![person, reviewer]);
    }

    #[test]
    fn nodes_with_label_single_label_fast_path() {
        // With no secondary labels anywhere, nodes_with_label must
        // return exactly type_indices[label] — the byte-identical
        // result every primary-only call site produced pre-multi-label.
        let mut g = DirGraph::new();
        let a = add_node(&mut g, "a", "Person");
        let b = add_node(&mut g, "b", "Person");
        add_node(&mut g, "w", "Widget");

        assert!(!g.has_secondary_labels);
        assert_eq!(g.nodes_with_label("Person"), vec![a, b]);
        assert_eq!(g.nodes_with_label("Widget").len(), 1);
        assert!(g.nodes_with_label("Absent").is_empty());
    }

    #[test]
    fn nodes_with_label_unions_primary_and_secondary() {
        let mut g = DirGraph::new();
        let a = add_node(&mut g, "a", "Person"); // primary Person, + VIP
        let b = add_node(&mut g, "b", "Person"); // primary Person only
        let w = add_node(&mut g, "w", "Widget"); // primary Widget, + VIP
        let vip = g.interner.get_or_intern("VIP");
        g.add_node_label(a, vip);
        g.add_node_label(w, vip);

        // Primary lookups still include their primary-typed nodes.
        let persons = g.nodes_with_label("Person");
        assert_eq!(persons, vec![a, b]);

        // :VIP is a secondary-only label — union pulls from both buckets.
        let mut vips = g.nodes_with_label("VIP");
        vips.sort();
        let mut expected = vec![a, w];
        expected.sort();
        assert_eq!(vips, expected);
    }

    #[test]
    fn node_has_label_primary_secondary_and_absent() {
        let mut g = DirGraph::new();
        let a = add_node(&mut g, "a", "Person");
        let person = g.interner.get_or_intern("Person");
        let vip = g.interner.get_or_intern("VIP");
        let ghost = g.interner.get_or_intern("Ghost");
        g.add_node_label(a, vip);

        assert!(g.node_has_label(a, person)); // primary
        assert!(g.node_has_label(a, vip)); // secondary
        assert!(!g.node_has_label(a, ghost)); // absent
    }

    #[test]
    fn detach_delete_evicts_secondary_label_index() {
        use std::collections::HashSet;
        let mut g = DirGraph::new();
        let a = add_node(&mut g, "a", "Person");
        let b = add_node(&mut g, "b", "Person");
        let vip = g.interner.get_or_intern("VIP");
        g.add_node_label(a, vip);
        g.add_node_label(b, vip);
        assert_eq!(g.secondary_label_index[&vip].len(), 2);

        let to_del: HashSet<NodeIndex> = [a].into_iter().collect();
        crate::graph::mutation::maintain::detach_delete_nodes(&mut g, &to_del);

        // `a` evicted from the secondary index; `b` survives. Without the
        // eviction the StableDiGraph would keep `a` live in the bucket and
        // `nodes_with_label` / counts would over-report.
        assert_eq!(g.secondary_label_index.get(&vip).map(|v| v.len()), Some(1));
        assert!(g.has_secondary_labels);
        assert_eq!(g.nodes_with_label("VIP"), vec![b]);
    }
}

#[cfg(test)]
mod bulk_index_freshness_tests {
    use super::*;
    use crate::datatypes::values::{DataFrame, Value};
    use crate::graph::mutation::maintain::add_nodes;

    fn people(rows: Vec<(&str, &str)>) -> DataFrame {
        DataFrame::from_cypher_rows(
            vec!["id".to_string(), "city".to_string()],
            rows.into_iter()
                .map(|(id, city)| {
                    vec![
                        Value::String(id.to_string()),
                        Value::String(city.to_string()),
                    ]
                })
                .collect(),
        )
        .expect("dataframe")
    }

    #[test]
    fn add_nodes_keeps_property_index_fresh() {
        let mut g = DirGraph::new();
        add_nodes(
            &mut g,
            people(vec![("p1", "Oslo")]),
            "Person".to_string(),
            "id".to_string(),
            None,
            None,
        )
        .expect("first load");

        assert_eq!(g.create_index("Person", "city"), 1);

        add_nodes(
            &mut g,
            people(vec![("p2", "Oslo"), ("p3", "Bergen")]),
            "Person".to_string(),
            "id".to_string(),
            None,
            None,
        )
        .expect("second load");

        let oslo = g
            .lookup_by_index("Person", "city", &Value::String("Oslo".to_string()))
            .unwrap_or_default();
        assert_eq!(oslo.len(), 2, "bulk load left the property index stale");
    }

    #[test]
    fn add_nodes_keeps_range_and_composite_indexes_fresh() {
        let mut g = DirGraph::new();
        add_nodes(
            &mut g,
            people(vec![("p1", "Oslo")]),
            "Person".to_string(),
            "id".to_string(),
            None,
            None,
        )
        .expect("first load");

        g.create_range_index("Person", "city");
        g.create_composite_index("Person", &["city"]);

        add_nodes(
            &mut g,
            people(vec![("p2", "Oslo")]),
            "Person".to_string(),
            "id".to_string(),
            None,
            None,
        )
        .expect("second load");

        let oslo = Value::String("Oslo".to_string());
        let ranged = g
            .lookup_range(
                "Person",
                "city",
                std::ops::Bound::Included(&oslo),
                std::ops::Bound::Included(&oslo),
            )
            .unwrap_or_default();
        assert_eq!(ranged.len(), 2, "bulk load left the range index stale");

        let composite = g
            .lookup_by_composite_index("Person", &["city".to_string()], &[oslo])
            .unwrap_or_default();
        assert_eq!(
            composite.len(),
            2,
            "bulk load left the composite index stale"
        );
    }
}

#[cfg(test)]
mod constraint_snapshot_tests {
    use super::*;

    /// `populate_index_keys` snapshots the declared UNIQUE constraints out of a
    /// `HashMap`, whose iteration order is reseeded per process. Left unsorted,
    /// two saves of the same graph produce different bytes — and because the
    /// order only varies *between* processes, no single-process test catches it.
    /// Asserting the snapshot is sorted pins the invariant directly.
    #[test]
    fn populate_index_keys_snapshots_unique_constraints_sorted() {
        let mut graph = DirGraph::new();
        // Declared out of order, and across two node types, so an unsorted
        // snapshot has plenty of room to disagree with a sorted one.
        for (node_type, properties) in [
            ("Person", vec!["email"]),
            ("Order", vec!["ref"]),
            ("Person", vec!["city", "street"]),
            ("Person", vec!["ssn"]),
            ("Order", vec!["customer", "seq"]),
        ] {
            graph
                .create_unique_constraint(node_type, &properties)
                .expect("empty graph cannot violate a constraint");
        }

        graph.populate_index_keys();

        // Spelled out rather than compared against `sorted(snapshot)`: a
        // self-referential assertion can pass by luck when the HashMap happens
        // to hand back an already-ordered set.
        let expected: Vec<(String, Vec<String>)> = [
            ("Order", vec!["customer", "seq"]),
            ("Order", vec!["ref"]),
            ("Person", vec!["city", "street"]),
            ("Person", vec!["email"]),
            ("Person", vec!["ssn"]),
        ]
        .into_iter()
        .map(|(t, props)| {
            (
                t.to_string(),
                props.into_iter().map(str::to_string).collect(),
            )
        })
        .collect();
        assert_eq!(
            graph.unique_constraint_keys, expected,
            "unique_constraint_keys must be persisted in a deterministic order"
        );
    }
}

/// Index freshness after the two *property-overwrite* paths.
///
/// Sibling of `bulk_index_freshness_tests` above, which covers the bulk
/// *append*. The failure mode here is strictly worse: appending behind a stale
/// index hides rows, whereas overwriting a value behind one makes
/// `MATCH (n:T {prop: <old value>})` return a node that no longer holds the old
/// value. A wrong answer, not a missing one.
#[cfg(test)]
mod overwrite_index_freshness_tests {
    use super::*;
    use crate::datatypes::values::{DataFrame, Value};
    use crate::graph::mutation::maintain::{add_nodes, update_node_properties};
    use crate::graph::storage::GraphWrite;

    fn people(rows: Vec<(&str, &str)>) -> DataFrame {
        DataFrame::from_cypher_rows(
            vec!["id".to_string(), "city".to_string()],
            rows.into_iter()
                .map(|(id, city)| {
                    vec![
                        Value::String(id.to_string()),
                        Value::String(city.to_string()),
                    ]
                })
                .collect(),
        )
        .expect("dataframe")
    }

    /// `update_node_properties` writes through the batch path, which skips the
    /// per-write index maintenance the Cypher SET path performs. Before the fix
    /// the equality index kept the pre-update value, so a lookup for the *old*
    /// value still returned the node.
    #[test]
    fn update_node_properties_keeps_the_property_index_fresh() {
        let mut g = DirGraph::new();
        add_nodes(
            &mut g,
            people(vec![("p1", "Oslo")]),
            "Person".to_string(),
            "id".to_string(),
            None,
            None,
        )
        .expect("load");

        assert_eq!(g.create_index("Person", "city"), 1);
        let node = g
            .type_indices
            .get("Person")
            .and_then(|nodes| nodes.iter().next())
            .expect("the loaded Person node");

        update_node_properties(
            &mut g,
            &[(Some(node), Value::String("Bergen".to_string()))],
            "city",
        )
        .expect("update");

        let stale = g
            .lookup_by_index("Person", "city", &Value::String("Oslo".to_string()))
            .unwrap_or_default();
        assert!(
            stale.is_empty(),
            "the index still resolves the overwritten value 'Oslo' to {stale:?} — \
             MATCH (n:Person {{city: 'Oslo'}}) would return a node whose city is 'Bergen'"
        );

        let fresh = g
            .lookup_by_index("Person", "city", &Value::String("Bergen".to_string()))
            .unwrap_or_default();
        assert_eq!(fresh, vec![node], "the new value is not indexed");
    }

    /// The range and composite structures share the same refresh, so they must
    /// forget the overwritten value too.
    #[test]
    fn update_node_properties_keeps_range_and_composite_indexes_fresh() {
        let mut g = DirGraph::new();
        add_nodes(
            &mut g,
            people(vec![("p1", "Oslo")]),
            "Person".to_string(),
            "id".to_string(),
            None,
            None,
        )
        .expect("load");

        g.create_range_index("Person", "city");
        g.create_composite_index("Person", &["city"]);
        let node = g
            .type_indices
            .get("Person")
            .and_then(|nodes| nodes.iter().next())
            .expect("the loaded Person node");

        update_node_properties(
            &mut g,
            &[(Some(node), Value::String("Bergen".to_string()))],
            "city",
        )
        .expect("update");

        let oslo = Value::String("Oslo".to_string());
        let ranged = g
            .lookup_range(
                "Person",
                "city",
                std::ops::Bound::Included(&oslo),
                std::ops::Bound::Included(&oslo),
            )
            .unwrap_or_default();
        assert!(
            ranged.is_empty(),
            "the range index still resolves the overwritten value: {ranged:?}"
        );

        let composite = g
            .lookup_by_composite_index("Person", &["city".to_string()], &[oslo])
            .unwrap_or_default();
        assert!(
            composite.is_empty(),
            "the composite index still resolves the overwritten value: {composite:?}"
        );
    }

    /// Bulk-update validation is intentionally reused when batch actions are
    /// assembled. Duplicate live rows must still count as duplicate updates,
    /// while missing/absent rows retain the existing report and error shape.
    #[test]
    fn update_node_properties_reuses_validation_without_changing_report_semantics() {
        let mut g = DirGraph::new();
        add_nodes(
            &mut g,
            people(vec![("p1", "Oslo"), ("p2", "Trondheim")]),
            "Person".to_string(),
            "id".to_string(),
            None,
            None,
        )
        .expect("load");

        let loaded: Vec<NodeIndex> = g
            .type_indices
            .get("Person")
            .expect("the loaded Person nodes")
            .iter()
            .collect();
        let [node, dead] = loaded.as_slice() else {
            panic!("expected exactly two loaded Person nodes");
        };
        let (node, dead) = (*node, *dead);
        GraphWrite::remove_node(&mut g.graph, dead).expect("remove the second node");
        let missing = NodeIndex::new(node.index() + 10_000);
        let report = update_node_properties(
            &mut g,
            &[
                (Some(node), Value::Int64(7)),
                (Some(node), Value::Int64(7)),
                (Some(dead), Value::Int64(7)),
                (Some(missing), Value::Int64(7)),
                (None, Value::Int64(7)),
            ],
            "city",
        )
        .expect("valid rows still update when other rows are absent");

        assert_eq!(
            report.nodes_updated, 2,
            "duplicate live rows remain updates"
        );
        assert_eq!(
            report.nodes_skipped, 6,
            "dead, missing, and absent rows retain validation + assembly skip accounting"
        );
        assert_eq!(report.errors.len(), 5);
        for invalid in [dead, missing] {
            assert!(report
                .errors
                .iter()
                .any(|error| error == &format!("Node index {:?} not found in graph", invalid)));
            assert!(report
                .errors
                .iter()
                .any(|error| error == &format!("Node index {:?} is out of bounds", invalid)));
        }
        assert!(report
            .errors
            .iter()
            .any(|error| error.contains("Type mismatch")));
        assert_eq!(
            g.node_view(node)
                .and_then(|data| data.get_property("city"))
                .map(|value| value.into_owned()),
            Some(Value::Int64(7))
        );
    }
}

/// The range index must fork by pointer, not by copy — the same structural
/// property `property_indices` gets from [`super::index_layer::LayeredIndex`].
///
/// Held-view first write at 100k measured 0.889 ms with a range index against
/// 0.048 ms with an equality index (P4, 2026-08-13): the whole gap was the
/// plain `BTreeMap` being deep-cloned on the copy-on-write fork.
#[cfg(test)]
mod range_index_fork_tests {
    use super::*;
    use crate::datatypes::Value;
    use crate::graph::session::{execute_mut, ExecuteOptions};

    fn run(graph: &mut DirGraph, query: &str) {
        let params = HashMap::new();
        let opts = ExecuteOptions::eager(&params);
        execute_mut(graph, query, &opts).unwrap_or_else(|e| panic!("`{query}` failed: {e}"));
    }

    fn indexed_graph() -> DirGraph {
        let mut graph = DirGraph::new();
        run(
            &mut graph,
            "UNWIND range(0, 999) AS i CREATE (:Item {id: i, qty: i % 97})",
        );
        graph.create_range_index("Item", "qty");
        graph
    }

    fn bucket_ptr(graph: &DirGraph, value: i64) -> *const petgraph::graph::NodeIndex {
        graph
            .range_indices
            .get(&("Item".to_string(), "qty".to_string()))
            .expect("range index present")
            .get(&Value::Int64(value))
            .expect("bucket present")
            .as_ptr()
    }

    /// A fork shares the range index's buckets outright.
    ///
    /// `as_ptr` on the members `Vec` is the observation: a deep-cloned
    /// `BTreeMap` reallocates every bucket, a shared immutable level hands both
    /// sides the same allocation.
    #[test]
    fn a_fork_shares_the_range_index_buckets() {
        let graph = indexed_graph();
        let fork = graph.clone();

        for value in [0i64, 13, 96] {
            assert_eq!(
                bucket_ptr(&graph, value),
                bucket_ptr(&fork, value),
                "bucket {value} was copied by the fork instead of shared"
            );
        }
    }

    /// ...and the writer's edits stay invisible to the reader that forced the
    /// fork, which is what makes the sharing safe.
    #[test]
    fn a_write_after_the_fork_leaves_the_readers_buckets_alone() {
        let mut writer = indexed_graph();
        let reader = writer.clone();

        run(&mut writer, "MATCH (n:Item {id: 5}) SET n.qty = 500");

        let key = ("Item".to_string(), "qty".to_string());
        let reader_bucket = reader.range_indices[&key]
            .get(&Value::Int64(5))
            .expect("the reader keeps its pre-write bucket");
        assert!(
            reader_bucket.len() > 1,
            "the reader's bucket must still hold every pre-write member"
        );
        assert!(
            reader.range_indices[&key].get(&Value::Int64(500)).is_none(),
            "the reader must not see the writer's new bucket"
        );
        assert!(
            writer.range_indices[&key]
                .get(&Value::Int64(500))
                .is_some_and(|members| members.len() == 1),
            "the writer's new bucket must exist on its side"
        );

        // Ordered iteration is what a range index is for: the merged view must
        // still come out sorted, tombstones and overlays included.
        let values: Vec<i64> = writer.range_indices[&key]
            .iter()
            .filter_map(|(value, _)| match value {
                Value::Int64(n) => Some(*n),
                _ => None,
            })
            .collect();
        let mut sorted = values.clone();
        sorted.sort_unstable();
        assert_eq!(values, sorted, "the merged iteration must stay ordered");
        assert_eq!(values.last(), Some(&500));
    }
}

/// `ensure_column_store_for_push` used to rebuild the whole store whenever the
/// registered `TypeSchema` had grown past the store's own — every existing row
/// re-pushed into a fresh store, per newly-seen key. With `ColumnStore::push_row`
/// appending its own columns that rebuild is not merely an optimisation
/// opportunity, it is wrong work: it is O(rows x cols) on a path whose contract
/// is one row.
#[cfg(test)]
mod ensure_column_store_for_push_tests {
    use super::*;
    use crate::datatypes::Value;
    use crate::graph::storage::column_store::{
        column_store_row_pushes, reset_column_store_row_pushes,
    };

    fn push(graph: &mut DirGraph, node_type: &str, pairs: &[(&str, Value)]) -> u32 {
        let interned: Vec<(InternedKey, Value)> = pairs
            .iter()
            .map(|(k, v)| (graph.interner.get_or_intern(k), v.clone()))
            .collect();
        let keys: Vec<InternedKey> = interned.iter().map(|(k, _)| *k).collect();
        graph.ensure_type_schema_keys(node_type, &keys);
        let store = graph.ensure_column_store_for_push(node_type);
        store.push_row(&interned)
    }

    #[test]
    fn a_widening_key_set_never_rebuilds_the_store() {
        let mut g = DirGraph::new();
        for i in 0..50i64 {
            push(&mut g, "Item", &[("p0", Value::Int64(i))]);
        }

        reset_column_store_row_pushes();
        // Three statements, each introducing a property the type has never
        // carried. Before the append path this cost 51 + 52 + 53 row pushes.
        push(
            &mut g,
            "Item",
            &[("p0", Value::Int64(50)), ("p1", Value::Int64(1))],
        );
        push(
            &mut g,
            "Item",
            &[("p0", Value::Int64(51)), ("p2", Value::Int64(2))],
        );
        push(
            &mut g,
            "Item",
            &[("p0", Value::Int64(52)), ("p3", Value::Int64(3))],
        );
        assert_eq!(
            column_store_row_pushes(),
            3,
            "growing a type's schema rebuilt its ColumnStore row by row"
        );

        // ... and nothing was lost or shifted by the growth.
        let store = g.column_store("Item").expect("store");
        let p0 = InternedKey::from_str("p0");
        assert_eq!(store.row_count(), 53);
        for i in 0..53u32 {
            assert_eq!(store.get(i, p0), Some(Value::Int64(i as i64)));
        }
        assert_eq!(
            store.get(50, InternedKey::from_str("p1")),
            Some(Value::Int64(1))
        );
        assert_eq!(
            store.get(51, InternedKey::from_str("p2")),
            Some(Value::Int64(2))
        );
        assert_eq!(
            store.get(52, InternedKey::from_str("p3")),
            Some(Value::Int64(3))
        );
        // Rows that predate a column read as absent.
        assert_eq!(store.get(0, InternedKey::from_str("p1")), None);
    }

    #[test]
    fn a_rebuild_would_have_resurrected_tombstoned_rows() {
        // The rebuild loop re-pushed `0..row_count` and never carried the
        // tombstone bitmap across, so a row deleted before a schema growth came
        // back as a live row. `materialize_for_append`, the other migrate-style
        // copy in the store, does re-tombstone — the two disagreed.
        let mut g = DirGraph::new();
        for i in 0..8i64 {
            push(&mut g, "Item", &[("p0", Value::Int64(i))]);
        }
        Arc::make_mut(g.column_store_mut("Item").expect("store")).tombstone(3);
        assert_eq!(g.column_store("Item").expect("store").live_count(), 7);

        push(
            &mut g,
            "Item",
            &[("p0", Value::Int64(8)), ("fresh", Value::Int64(1))],
        );

        let store = g.column_store("Item").expect("store");
        assert_eq!(
            store.live_count(),
            8,
            "a schema growth resurrected a tombstoned row"
        );
        assert_eq!(store.get(3, InternedKey::from_str("p0")), None);
    }
}

/// Auto-vacuum's trigger, and the kind of garbage it could not see.
///
/// `node_bound - node_count` counts *free petgraph slots*, which a later
/// create takes back. A columnar row does not work that way: a delete leaves
/// its row behind and a create appends a new one, so replacement churn grows
/// the store without ever moving the node-slot reading off zero. Measured on
/// the disk backend, which is already always-columnar: 1,500 delete/create
/// pairs over a 2,000-node type left 3,500 rows for 2,000 live nodes at
/// `fragmentation_ratio` 0.000. Under the always-columnar flip that becomes
/// every graph's steady state.
#[cfg(test)]
mod auto_vacuum_trigger_tests {
    use super::*;
    use crate::datatypes::{DataFrame, Value};

    fn columnar_items(n: i64) -> DirGraph {
        let mut g = DirGraph::new();
        let rows: Vec<Vec<Value>> = (1..=n)
            .map(|i| {
                vec![
                    Value::Int64(i),
                    Value::String(format!("t{i}")),
                    Value::Int64(i * 10),
                ]
            })
            .collect();
        let df = DataFrame::from_cypher_rows(
            vec!["id".to_string(), "title".to_string(), "c0".to_string()],
            rows,
        )
        .unwrap();
        crate::graph::mutation::maintain::add_nodes(
            &mut g,
            df,
            "Item".to_string(),
            "id".to_string(),
            Some("title".to_string()),
            None,
        )
        .unwrap();
        g.enable_columnar();
        g
    }

    /// Rows in the store that no live node points at, with the petgraph slot
    /// count reading clean — the churn residue, reproduced directly.
    fn orphan_rows(g: &mut DirGraph, count: usize) {
        let key = g.interner.get_or_intern("c0");
        let store = g.ensure_column_store_for_push("Item");
        for i in 0..count {
            store.push_id(&Value::Int64(1_000_000 + i as i64));
            store.push_title(&Value::String(format!("dead{i}")));
            store.push_row(&[(key, Value::Int64(-1))]);
        }
    }

    #[test]
    fn columnar_garbage_triggers_a_vacuum_that_reclaims_it() {
        let mut g = columnar_items(200);
        g.auto_vacuum_threshold = Some(0.3);
        orphan_rows(&mut g, 150);

        let (total, live) = g.columnar_row_census();
        assert_eq!((total, live), (350, 200), "fixture drift");
        assert_eq!(
            g.graph.node_bound() - g.graph.node_count(),
            0,
            "precondition: the node-slot reading must be clean, or this test is \
             measuring the old trigger"
        );

        assert!(
            g.check_auto_vacuum(),
            "43% of the type's rows are garbage and auto-vacuum did not fire: \
             the trigger is reading free petgraph slots, which replacement \
             churn returns to zero"
        );
        let (total, live) = g.columnar_row_census();
        assert_eq!(
            (total, live),
            (200, 200),
            "the vacuum fired but reclaimed no columnar rows"
        );
        // The live data is untouched by the reclamation.
        assert_eq!(g.graph.node_count(), 200);
    }

    #[test]
    fn a_clean_store_does_not_trigger_a_vacuum() {
        // Non-vacuity's other half: the trigger must still say no.
        let mut g = columnar_items(200);
        g.auto_vacuum_threshold = Some(0.3);
        assert!(!g.check_auto_vacuum());

        // ... and garbage under the small-graph floor is not worth a rebuild.
        orphan_rows(&mut g, 40);
        assert!(!g.check_auto_vacuum());

        // ... nor is garbage below the ratio threshold.
        let mut g = columnar_items(2000);
        g.auto_vacuum_threshold = Some(0.3);
        orphan_rows(&mut g, 300);
        assert!(!g.check_auto_vacuum());
        assert_eq!(g.columnar_row_census(), (2300, 2000));
    }
}