test(document): split index tests into sub-modules and add partial/fanout coverage

Split sql_index_naming.rs into tests/sql_index/ (naming, planner, backfill,
partial, helpers) to respect the 500-line hard limit and group tests by
concern. The file is now a thin entry point that includes each sub-module
via #[path = ...] so they compile in one test crate.

New tests cover partial-index predicate filtering (rows outside the WHERE
scope are not indexed, not UNIQUE-checked), the conjunct-aware planner
rewrite (equality nested in AND still picks up the index), and cluster
backfill fan-out (every node must execute BackfillIndex, not merely
replicate Raft state).

Harness gains document_index_backfill_count() on NodeHandle (reads the
new SystemMetrics atomic) and wires system_metrics into the CoreLoop in
lifecycle.rs so per-node counters are observable from tests.

Author: farhan-syah

nodedb/tests/common/cluster_harness/node/inspect.rs

@@ -37,6 +37,23 @@ impl TestClusterNode {
             .unwrap_or(0)
     }
 
+    /// Count of `DocumentOp::BackfillIndex` handler invocations on
+    /// this node's Data Plane since startup. A CREATE INDEX against a
+    /// cluster must fan out backfill to every node — this counter
+    /// exposes whether the local core actually executed the primitive
+    /// (a positive value) versus merely replicating Raft state from
+    /// the coordinator (counter stays 0).
+    pub fn document_index_backfill_count(&self) -> u64 {
+        self.shared
+            .system_metrics
+            .as_ref()
+            .map(|m| {
+                m.document_index_backfills
+                    .load(std::sync::atomic::Ordering::Relaxed)
+            })
+            .unwrap_or(0)
+    }
+
     /// Number of active collections visible on this node (read through
     /// the local `SystemCatalog` redb — populated by the
     /// `MetadataCommitApplier` on every node via

nodedb/tests/common/cluster_harness/node/lifecycle.rs

@@ -153,12 +153,16 @@ impl TestClusterNode {
         let data_side = data_sides.into_iter().next().expect("data side");
         let event_producer = event_producers.into_iter().next().expect("event producer");
         let core_dir = data_dir_path.clone();
+        let core_metrics = shared.system_metrics.clone();
         let (core_stop_tx, core_stop_rx) = std::sync::mpsc::channel::<()>();
         let core_handle = tokio::task::spawn_blocking(move || {
             let mut core =
                 CoreLoop::open(0, data_side.request_rx, data_side.response_tx, &core_dir)
                     .expect("core open");
             core.set_event_producer(event_producer);
+            if let Some(m) = core_metrics {
+                core.set_metrics(m);
+            }
             // Continue ticking only while the channel is Empty.
             // `Ok(())` means we got an explicit stop signal;
             // `Disconnected` means the sender was dropped (e.g. the

nodedb/tests/sql_ddl_cluster.rs

@@ -221,6 +221,67 @@ async fn ddl_create_drop_trigger_replicates() {
     cluster.shutdown().await;
 }
 
+// ── Index backfill fan-out ───────────────────────────────────────
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 6)]
+async fn create_index_backfill_runs_on_every_node() {
+    let cluster = TestCluster::spawn_three().await.expect("cluster");
+    cluster
+        .exec_ddl_on_any_leader("CREATE COLLECTION idx_fanout_coll")
+        .await
+        .expect("create coll");
+    wait_for(
+        "collection visible on all nodes",
+        Duration::from_secs(10),
+        Duration::from_millis(50),
+        || {
+            cluster
+                .nodes
+                .iter()
+                .all(|n| n.cached_collection_count() >= 1)
+        },
+    )
+    .await;
+
+    cluster
+        .exec_ddl_on_any_leader("INSERT INTO idx_fanout_coll (id, email) VALUES ('u1', 'a@b.com')")
+        .await
+        .expect("insert row");
+
+    cluster
+        .exec_ddl_on_any_leader("CREATE INDEX idx_fanout_email ON idx_fanout_coll(email)")
+        .await
+        .expect("create index");
+
+    // Every node's Data Plane must have executed BackfillIndex at
+    // least once — a distributed CREATE INDEX that runs the backfill
+    // only on the coordinator silently under-populates the index on
+    // rows hosted by other vShards. The counter distinguishes local
+    // handler invocation from Raft-replicated catalog state. Poll
+    // briefly to let any async applier post-effects settle before
+    // asserting final counts.
+    let deadline = std::time::Instant::now() + Duration::from_secs(5);
+    while std::time::Instant::now() < deadline
+        && !cluster
+            .nodes
+            .iter()
+            .all(|n| n.document_index_backfill_count() >= 1)
+    {
+        tokio::time::sleep(Duration::from_millis(50)).await;
+    }
+    let counts: Vec<u64> = cluster
+        .nodes
+        .iter()
+        .map(|n| n.document_index_backfill_count())
+        .collect();
+    assert!(
+        counts.iter().all(|&c| c >= 1),
+        "backfill must run on every node; got {:?}",
+        counts
+    );
+    cluster.shutdown().await;
+}
+
 // ── Schedule ─────────────────────────────────────────────────────
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 6)]

nodedb/tests/sql_index/backfill.rs

@@ -0,0 +1,96 @@
+//! Two-phase Building→Ready backfill and UNIQUE enforcement.
+
+use super::common::pgwire_harness::TestServer;
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn create_unique_index_rejects_duplicate_insert() {
+    let server = TestServer::start().await;
+
+    server
+        .exec("CREATE COLLECTION idx_unique_enforce")
+        .await
+        .unwrap();
+    server
+        .exec("CREATE UNIQUE INDEX ON idx_unique_enforce(email)")
+        .await
+        .unwrap();
+
+    // First insert must succeed.
+    server
+        .exec("INSERT INTO idx_unique_enforce { id: 'a', email: 'x@y.z' }")
+        .await
+        .unwrap();
+
+    // Second insert with the same indexed value must be rejected. Today the
+    // UNIQUE keyword is parsed (`is_unique`) but never persisted anywhere,
+    // so duplicates succeed silently — a correctness bug that is part of
+    // the same design flaw as the reporter's point-lookup issue: CREATE
+    // INDEX DDL modifiers are parsed but not dispatched to the config or
+    // enforcement layer.
+    server
+        .expect_error(
+            "INSERT INTO idx_unique_enforce { id: 'b', email: 'x@y.z' }",
+            "unique",
+        )
+        .await;
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn create_index_backfills_existing_rows() {
+    let server = TestServer::start().await;
+
+    server.exec("CREATE COLLECTION bf_col").await.unwrap();
+    server
+        .exec("INSERT INTO bf_col { id: 'a', email: 'one@x.com' }")
+        .await
+        .unwrap();
+    server
+        .exec("INSERT INTO bf_col { id: 'b', email: 'two@x.com' }")
+        .await
+        .unwrap();
+
+    // CREATE INDEX runs AFTER the rows exist. The two-phase
+    // Building→Ready backfill pipeline must populate the index from the
+    // pre-existing documents before flipping to Ready; otherwise a
+    // subsequent lookup against the index would miss the rows (same
+    // silent-miss class as the original reporter's bug).
+    server.exec("CREATE INDEX ON bf_col(email)").await.unwrap();
+
+    let rows = server
+        .query_text("SELECT id FROM bf_col WHERE email = 'one@x.com'")
+        .await
+        .expect("indexed SELECT must succeed");
+    assert_eq!(
+        rows.len(),
+        1,
+        "indexed SELECT must return exactly one row, got: {rows:?}"
+    );
+    assert!(
+        rows[0].contains("\"a\""),
+        "indexed SELECT row must reference doc id 'a', got: {}",
+        rows[0]
+    );
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn create_unique_index_rejects_existing_duplicates() {
+    let server = TestServer::start().await;
+
+    server.exec("CREATE COLLECTION bf_unique").await.unwrap();
+    server
+        .exec("INSERT INTO bf_unique { id: 'a', code: 'ABC' }")
+        .await
+        .unwrap();
+    server
+        .exec("INSERT INTO bf_unique { id: 'b', code: 'ABC' }")
+        .await
+        .unwrap();
+
+    // CREATE UNIQUE INDEX on a collection that already contains
+    // duplicates must fail at the backfill phase — detecting the
+    // violation before the Ready flip so the catalog never advertises
+    // an index that doesn't actually enforce uniqueness.
+    server
+        .expect_error("CREATE UNIQUE INDEX ON bf_unique(code)", "unique")
+        .await;
+}

nodedb/tests/sql_index/helpers.rs

@@ -0,0 +1,32 @@
+//! Shared helpers for the `sql_index_naming` test binary.
+
+#![allow(dead_code)] // Not every sub-file uses every helper.
+
+use super::common::pgwire_harness::TestServer;
+
+/// Return all rows of EXPLAIN <sql> concatenated, lowercased.
+///
+/// Used to assert that the chosen physical plan references an index lookup
+/// rather than a full scan when a WHERE predicate lands on an indexed field.
+pub async fn explain_lower(server: &TestServer, sql: &str) -> String {
+    let rows = server
+        .query_text(&format!("EXPLAIN {sql}"))
+        .await
+        .expect("EXPLAIN must succeed");
+    rows.join("\n").to_lowercase()
+}
+
+/// Return all rows of EXPLAIN <sql> concatenated, preserving case.
+///
+/// The pgwire EXPLAIN handler emits the `PhysicalPlan` via `{:?}` debug
+/// format, so structural fields like `filters: [..]`, `projection: [..]`,
+/// `limit: N`, etc. appear verbatim. Plan-shape assertions use this
+/// form; `explain_lower` is only for case-insensitive plan-variant
+/// name checks.
+pub async fn explain_raw(server: &TestServer, sql: &str) -> String {
+    let rows = server
+        .query_text(&format!("EXPLAIN {sql}"))
+        .await
+        .expect("EXPLAIN must succeed");
+    rows.join("\n")
+}

nodedb/tests/sql_index/naming.rs

@@ -0,0 +1,101 @@
+//! CREATE INDEX naming, SHOW INDEXES, and parse-time shape tests.
+//!
+//! The "silent no-op" regression mode (DDL parses, audit records
+//! ownership, but the secondary index is never wired into the
+//! collection config) is guarded here at the registration surface.
+//! Deeper behavioral checks (planner, backfill, partial) live in
+//! sibling files.
+
+use super::common::pgwire_harness::TestServer;
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn create_index_named() {
+    let server = TestServer::start().await;
+
+    server.exec("CREATE COLLECTION idx_named").await.unwrap();
+    server
+        .exec("INSERT INTO idx_named { id: 'a', role: 'admin' }")
+        .await
+        .unwrap();
+
+    // Named index — standard SQL form.
+    server
+        .exec("CREATE INDEX my_idx ON idx_named(role)")
+        .await
+        .unwrap();
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn create_index_unnamed_auto_name() {
+    let server = TestServer::start().await;
+
+    server.exec("CREATE COLLECTION idx_unnamed").await.unwrap();
+    server
+        .exec("INSERT INTO idx_unnamed { id: 'a', email: 'a@b.com' }")
+        .await
+        .unwrap();
+
+    // No name — should auto-generate name and succeed.
+    server
+        .exec("CREATE INDEX ON idx_unnamed(email)")
+        .await
+        .unwrap();
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn create_index_fields_keyword() {
+    let server = TestServer::start().await;
+
+    server.exec("CREATE COLLECTION idx_fields").await.unwrap();
+    server
+        .exec("INSERT INTO idx_fields { id: 'a', tag: 'rust' }")
+        .await
+        .unwrap();
+
+    // FIELDS keyword form — should succeed.
+    server
+        .exec("CREATE INDEX ON idx_fields FIELDS tag")
+        .await
+        .unwrap();
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn create_unique_index_unnamed() {
+    let server = TestServer::start().await;
+
+    server.exec("CREATE COLLECTION idx_unique").await.unwrap();
+    server
+        .exec("INSERT INTO idx_unique { id: 'a', code: 'ABC' }")
+        .await
+        .unwrap();
+
+    // Unnamed UNIQUE index.
+    server
+        .exec("CREATE UNIQUE INDEX ON idx_unique(code)")
+        .await
+        .unwrap();
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn show_indexes_lists_created_index() {
+    let server = TestServer::start().await;
+
+    server.exec("CREATE COLLECTION idx_show").await.unwrap();
+    server
+        .exec("CREATE INDEX idx_show_role ON idx_show(role)")
+        .await
+        .unwrap();
+
+    // Positive lock-in: SHOW INDEXES must list a freshly created index. This
+    // is the user-visible confirmation that creation succeeded; without it,
+    // operators have no feedback channel. (First-column read of `query_text`
+    // returns the index name.)
+    let names = server
+        .query_text("SHOW INDEXES")
+        .await
+        .expect("SHOW INDEXES must succeed");
+    assert!(
+        names.iter().any(|n| n == "idx_show_role"),
+        "SHOW INDEXES must list created index, got: {names:?}"
+    );
+}