feat(document): partial index support with cluster-wide backfill fan-out

Add WHERE-predicate support to document secondary indexes throughout the
full stack, fixing the silent-miss class (issue #71) where a partial UNIQUE
index applied uniqueness constraints to rows outside its predicate scope.

Predicate text is carried on RegisteredIndex from DDL through the wire
format to the Data Plane, where it is parsed once into IndexPredicate and
evaluated per row on every write, UNIQUE check, and backfill pass. Only
rows for which the predicate is explicitly true are indexed or constrained
(Postgres partial-index semantics — NULL and non-boolean results exclude
the row).

Cluster CREATE INDEX now fans the BackfillIndex primitive to every Active
peer via RaftRpc::ExecuteRequest after the coordinator's local dispatch
succeeds. Without the fan-out, non-coordinator nodes that hosted rows the
coordinator never visited would produce incomplete indexes on Ready flip.

The planner rewrite also gains conjunct-aware equality extraction so a
WHERE clause of the form `a = 1 AND b > 2` correctly drives an index
lookup on `a` while carrying `b > 2` as a residual post-filter, and
partial indexes are rejected when the query's conjuncts don't entail the
index predicate (conservative structural check). DISTINCT is no longer
an automatic fallback to full-scan on the indexed-fetch path.

Author: farhan-syah

nodedb-sql/src/engine_rules/mod.rs

@@ -124,36 +124,71 @@ pub(crate) fn try_document_index_lookup(
     params: &ScanParams,
     engine: EngineType,
 ) -> Option<SqlPlan> {
-    // Sort / distinct / window functions are not yet supported on the
-    // indexed-fetch path — fall back to a full scan so existing behavior
-    // stays correct. Extending the handler later is additive and doesn't
-    // invalidate the rewrite.
-    if !params.sort_keys.is_empty() || params.distinct || !params.window_functions.is_empty() {
+    // Sort / window functions still force a full scan because the
+    // IndexedFetch handler does not yet order rows or evaluate window
+    // aggregates. DISTINCT is safe on the index path: the handler
+    // emits each matched doc exactly once and any further dedup is a
+    // cheap Control-Plane pass on the scan-shaped response.
+    if !params.sort_keys.is_empty() || !params.window_functions.is_empty() {
         return None;
     }
 
     // Iterate filters to find the first equality candidate that lines up
     // with a Ready index. Keep the remaining filters as post-filters.
-    // Two predicate shapes appear in practice: the resolver-emitted
-    // `FilterExpr::Comparison` (compact) and the generic
-    // `FilterExpr::Expr(SqlExpr::BinaryOp { Column, Eq, Literal })`
-    // wrapper the default path produces. Both unambiguously express
-    // equality on a column — handle both.
+    // Predicates appear in three shapes:
+    //   - `FilterExpr::Comparison { field, Eq, value }` — resolver path.
+    //   - `FilterExpr::Expr(BinaryOp { Column, Eq, Literal })` — generic.
+    //   - `FilterExpr::Expr(BinaryOp { left AND right })` — the
+    //     top-level AND the `convert_where_to_filters` path produces
+    //     for compound WHERE clauses like `a = 1 AND b > 2`. Descend
+    //     into AND conjuncts so an equality nested in a conjunction
+    //     still picks up the index, and the non-equality siblings
+    //     survive as a residual conjunction carried on the
+    //     IndexedFetch node (not as a wrapping Filter plan node —
+    //     that wrapping is the Control-Plane post-filter anti-pattern
+    //     the handler's `filters` payload exists to eliminate).
+    // Pre-collect the query's top-level conjuncts once. Used for
+    // partial-index entailment: every conjunct of the index predicate
+    // must appear as a conjunct of the query WHERE for the rewrite to
+    // be sound, otherwise the index would silently omit rows the
+    // query expects to see.
+    let query_conjuncts: Vec<SqlExpr> = params
+        .filters
+        .iter()
+        .flat_map(|f| match &f.expr {
+            FilterExpr::Expr(e) => {
+                let mut v = Vec::new();
+                flatten_and(e, &mut v);
+                v
+            }
+            _ => Vec::new(),
+        })
+        .collect();
+
     for (i, f) in params.filters.iter().enumerate() {
-        let Some((field, value)) = extract_equality(&f.expr) else {
+        let Some((field, value, residual)) = extract_equality_with_residual(&f.expr) else {
             continue;
         };
         let canonical = canonical_index_field(&field);
-        let Some(idx) = params
-            .indexes
-            .iter()
-            .find(|i| i.state == IndexState::Ready && i.field == canonical)
-        else {
+        let Some(idx) = params.indexes.iter().find(|i| {
+            i.state == IndexState::Ready
+                && i.field == canonical
+                && partial_index_entailed(i.predicate.as_deref(), &query_conjuncts)
+        }) else {
             continue;
         };
 
         let mut remaining = params.filters.clone();
-        remaining.remove(i);
+        match residual {
+            Some(expr) => {
+                remaining[i] = Filter {
+                    expr: FilterExpr::Expr(expr),
+                };
+            }
+            None => {
+                remaining.remove(i);
+            }
+        }
 
         let lookup_value = if idx.case_insensitive {
             lowercase_string_value(&value)
@@ -180,32 +215,131 @@ pub(crate) fn try_document_index_lookup(
     None
 }
 
-/// Pull `(column_name, equality_value)` out of a filter expression if it
-/// is a column-equals-literal predicate in either of the planner's two
-/// encodings.
-fn extract_equality(expr: &FilterExpr) -> Option<(String, SqlValue)> {
+/// Pull `(column_name, equality_value, residual_expr)` out of a filter
+/// expression if it contains a column-equals-literal predicate that can
+/// drive an index lookup. Returns `None` if no usable equality is
+/// present. The returned residual is the rest of the conjunction with
+/// the equality removed — `None` when the filter was a bare equality.
+fn extract_equality_with_residual(
+    expr: &FilterExpr,
+) -> Option<(String, SqlValue, Option<SqlExpr>)> {
     match expr {
         FilterExpr::Comparison {
             field,
             op: CompareOp::Eq,
             value,
-        } => Some((field.clone(), value.clone())),
-        FilterExpr::Expr(SqlExpr::BinaryOp { left, op, right }) => {
-            let (col, lit) = match (left.as_ref(), op, right.as_ref()) {
-                (SqlExpr::Column { name, .. }, BinaryOp::Eq, SqlExpr::Literal(v)) => {
-                    (name.clone(), v.clone())
-                }
-                (SqlExpr::Literal(v), BinaryOp::Eq, SqlExpr::Column { name, .. }) => {
-                    (name.clone(), v.clone())
-                }
-                _ => return None,
-            };
-            Some((col, lit))
+        } => Some((field.clone(), value.clone(), None)),
+        FilterExpr::Expr(sql_expr) => {
+            let (col, lit, residual) = split_equality_from_expr(sql_expr)?;
+            Some((col, lit, residual))
         }
         _ => None,
     }
 }
 
+/// Return `(column, literal, residual_expr)` for an equality found
+/// anywhere in a right-leaning AND conjunction tree, or `None` if no
+/// bare column-equals-literal predicate exists. The residual preserves
+/// every sibling conjunct in their original order; `None` means the
+/// expression was a bare equality with nothing left behind.
+fn split_equality_from_expr(expr: &SqlExpr) -> Option<(String, SqlValue, Option<SqlExpr>)> {
+    // Gather the conjuncts of a top-level AND chain left-to-right.
+    let mut conjuncts: Vec<SqlExpr> = Vec::new();
+    flatten_and(expr, &mut conjuncts);
+
+    // Find the first conjunct that is a bare column-equals-literal.
+    let eq_idx = conjuncts.iter().position(is_column_eq_literal)?;
+    let eq = conjuncts.remove(eq_idx);
+    let (col, lit) = match eq {
+        SqlExpr::BinaryOp { left, op, right } => match (*left, op, *right) {
+            (SqlExpr::Column { name, .. }, BinaryOp::Eq, SqlExpr::Literal(v)) => (name, v),
+            (SqlExpr::Literal(v), BinaryOp::Eq, SqlExpr::Column { name, .. }) => (name, v),
+            _ => return None,
+        },
+        _ => return None,
+    };
+
+    let residual = rebuild_and(conjuncts);
+    Some((col, lit, residual))
+}
+
+/// Append every leaf of a right-leaning `AND` tree to `out`. Non-AND
+/// expressions are a single leaf.
+fn flatten_and(expr: &SqlExpr, out: &mut Vec<SqlExpr>) {
+    match expr {
+        SqlExpr::BinaryOp {
+            left,
+            op: BinaryOp::And,
+            right,
+        } => {
+            flatten_and(left, out);
+            flatten_and(right, out);
+        }
+        other => out.push(other.clone()),
+    }
+}
+
+/// Reassemble a list of conjuncts into a right-leaning AND tree.
+/// Empty input returns `None`; single input returns itself.
+fn rebuild_and(mut conjuncts: Vec<SqlExpr>) -> Option<SqlExpr> {
+    let last = conjuncts.pop()?;
+    Some(
+        conjuncts
+            .into_iter()
+            .rfold(last, |acc, next| SqlExpr::BinaryOp {
+                left: Box::new(next),
+                op: BinaryOp::And,
+                right: Box::new(acc),
+            }),
+    )
+}
+
+/// Decide whether the query's WHERE conjuncts entail the partial-index
+/// predicate. `None` predicate means a full index — trivially entailed.
+/// Initial version uses conjunct-level structural equality: every
+/// conjunct of the index predicate must appear (by `PartialEq`) as a
+/// conjunct of the query. This is conservative and deliberately so —
+/// a false positive here would silently omit rows from query results,
+/// which is the exact bug #73 reports.
+fn partial_index_entailed(predicate: Option<&str>, query_conjuncts: &[SqlExpr]) -> bool {
+    let Some(text) = predicate else {
+        return true;
+    };
+    let Ok(parsed) = crate::parse_expr_string(text) else {
+        // A catalog predicate we can't parse is not entailed — refuse
+        // to use the index rather than assume anything about its
+        // contents. The DDL path validates at CREATE INDEX time, so
+        // reaching this branch indicates drift.
+        return false;
+    };
+    let mut index_conjuncts: Vec<SqlExpr> = Vec::new();
+    flatten_and(&parsed, &mut index_conjuncts);
+    // Structural equality via the stable `Debug` representation:
+    // `SqlExpr` doesn't derive `PartialEq` (several nested variants
+    // carry types that can't derive it cheaply), but the Debug form is
+    // canonical for equivalent trees produced by the same parser. This
+    // is conservative — it matches only identical AST shapes and not,
+    // e.g., `a = 1` vs `1 = a`. Callers should write index predicates
+    // in the same normal form they use in query WHERE clauses, which
+    // is the convention everywhere else in the codebase.
+    index_conjuncts.iter().all(|ic| {
+        let ic_dbg = format!("{ic:?}");
+        query_conjuncts.iter().any(|qc| format!("{qc:?}") == ic_dbg)
+    })
+}
+
+fn is_column_eq_literal(expr: &SqlExpr) -> bool {
+    matches!(
+        expr,
+        SqlExpr::BinaryOp { left, op: BinaryOp::Eq, right }
+            if matches!(
+                (left.as_ref(), right.as_ref()),
+                (SqlExpr::Column { .. }, SqlExpr::Literal(_))
+                | (SqlExpr::Literal(_), SqlExpr::Column { .. }),
+            )
+    )
+}
+
 fn canonical_index_field(field: &str) -> String {
     if field.starts_with("$.") || field.starts_with('$') {
         field.to_string()

nodedb-sql/src/types.rs

@@ -530,6 +530,12 @@ pub struct IndexSpec {
     pub case_insensitive: bool,
     /// Build state. Only `Ready` indexes drive query rewrites.
     pub state: IndexState,
+    /// Partial-index predicate as raw SQL text (`WHERE <expr>` body
+    /// without the keyword), or `None` for full indexes. The planner
+    /// uses this to reject rewrites whose WHERE clause doesn't entail
+    /// the predicate — matching against such a partial index would
+    /// omit rows the index didn't cover.
+    pub predicate: Option<String>,
 }
 
 /// Planner-facing index state. Mirrors the catalog variant but lives here

nodedb/src/bridge/physical_plan/document.rs

@@ -256,6 +256,13 @@ pub struct RegisteredIndex {
     pub unique: bool,
     pub case_insensitive: bool,
     pub state: RegisteredIndexState,
+    /// Partial-index predicate as raw SQL text (`WHERE <expr>` body,
+    /// without the `WHERE` keyword). `None` for full indexes.
+    /// The write path parses and evaluates this per document; rows
+    /// where the predicate is false are not inserted into the index
+    /// and do not participate in UNIQUE enforcement.
+    #[serde(default)]
+    pub predicate: Option<String>,
 }
 
 /// Document engine physical operations (schemaless + strict + DML).
@@ -402,6 +409,11 @@ pub enum DocumentOp {
         is_array: bool,
         unique: bool,
         case_insensitive: bool,
+        /// Partial-index predicate (raw SQL text of the `WHERE` body)
+        /// or `None` for full indexes. Rows where the predicate is
+        /// false are skipped — not indexed, not UNIQUE-checked.
+        #[serde(default)]
+        predicate: Option<String>,
     },
 
     /// Truncate: delete ALL documents in a collection.

nodedb/src/bridge/physical_plan/wire.rs

@@ -99,6 +99,7 @@ mod tests {
                 unique: false,
                 case_insensitive: false,
                 state: crate::bridge::physical_plan::RegisteredIndexState::Ready,
+                predicate: None,
             }],
             crdt_enabled: false,
             storage_mode: crate::bridge::physical_plan::StorageMode::Schemaless,

nodedb/src/control/metrics/system.rs

@@ -67,6 +67,12 @@ pub struct SystemMetrics {
     pub document_inserts: AtomicU64,
     pub document_reads: AtomicU64,
     pub document_collections: AtomicU64,
+    /// Count of `DocumentOp::BackfillIndex` handler invocations on this
+    /// node's Data Plane. Per-node: every node increments its own
+    /// counter exactly when its local core runs the backfill primitive.
+    /// Tests assert per-node fan-out for distributed CREATE INDEX by
+    /// reading this counter on every node after DDL completion.
+    pub document_index_backfills: AtomicU64,
 
     pub columnar_segments: AtomicU64,
     pub columnar_compaction_queue: AtomicU64,
@@ -329,6 +335,11 @@ impl SystemMetrics {
         self.document_reads.fetch_add(1, Ordering::Relaxed);
     }
 
+    pub fn record_document_index_backfill(&self) {
+        self.document_index_backfills
+            .fetch_add(1, Ordering::Relaxed);
+    }
+
     pub fn update_document_collections(&self, count: u64) {
         self.document_collections.store(count, Ordering::Relaxed);
     }

nodedb/src/control/planner/catalog_adapter.rs

@@ -240,6 +240,7 @@ impl SqlCatalog for OriginCatalog {
                         nodedb_sql::types::IndexState::Ready
                     }
                 },
+                predicate: i.predicate.clone(),
             })
             .collect();
 

nodedb/src/control/server/native/dispatch/plan_builder/document.rs

@@ -327,6 +327,7 @@ pub(crate) fn build_register(fields: &TextFields, collection: &str) -> crate::Re
             unique: false,
             case_insensitive: false,
             state: crate::bridge::physical_plan::RegisteredIndexState::Ready,
+            predicate: None,
         })
         .collect();
 

nodedb/src/control/server/pgwire/ddl/collection/create/register.rs

@@ -75,6 +75,7 @@ fn derive_auto_indexes<'a>(
             unique: false,
             case_insensitive: false,
             state: crate::bridge::physical_plan::RegisteredIndexState::Ready,
+            predicate: None,
         })
         .collect()
 }
@@ -102,6 +103,7 @@ fn extend_with_catalog_indexes(
             unique: idx.unique,
             case_insensitive: idx.case_insensitive,
             state,
+            predicate: idx.predicate.clone(),
         };
         if let Some(existing) = out.iter_mut().find(|e| e.path == spec.path) {
             *existing = spec;

nodedb/src/control/server/pgwire/ddl/collection/index.rs

@@ -227,10 +227,11 @@ pub async fn create_index(
     let backfill_plan = crate::bridge::envelope::PhysicalPlan::Document(
         crate::bridge::physical_plan::DocumentOp::BackfillIndex {
             collection: collection.clone(),
-            path: extraction_path,
+            path: extraction_path.clone(),
             is_array,
             unique: is_unique,
             case_insensitive,
+            predicate: where_condition.clone(),
         },
     );
     let backfill_resp = crate::control::server::dispatch_utils::dispatch_to_data_plane(
@@ -257,6 +258,25 @@ pub async fn create_index(
         return Err(sqlstate_error(code, &detail));
     }
 
+    // Phase 2b: fan the same backfill op to every other cluster node.
+    // `execute_backfill_index` is vShard-local per core, so without
+    // this step non-coordinator nodes never populate the index for
+    // the rows they host — the #71 silent-miss bug. Single-node and
+    // peerless clusters short-circuit inside the helper.
+    super::index_fanout::backfill_on_peers(
+        state,
+        super::index_fanout::PeerBackfill {
+            tenant_id,
+            collection: &collection,
+            path: &extraction_path,
+            is_array,
+            unique: is_unique,
+            case_insensitive,
+            predicate: where_condition.as_deref(),
+        },
+    )
+    .await?;
+
     // Phase 3: flip to Ready. Re-read the collection so any concurrent
     // mutation (e.g. another DDL on the same collection — blocked by
     // descriptor drain in cluster mode, serialized by pgwire session in