chore(kv): elide redundant index rewrites on identical in-place updates (#279)

nodedb/src/engine/kv/index/composite.rs

@@ -100,6 +100,14 @@ impl KvCompositeIndex {
         }
     }
 
+    /// Whether the index contains the composite (field_values, primary_key) pair.
+    pub(crate) fn contains(&self, field_values: &[&[u8]], primary_key: &[u8]) -> bool {
+        let key = Self::build_key(field_values);
+        self.tree
+            .get(&key)
+            .is_some_and(|keys| keys.contains(primary_key))
+    }
+
     /// Exact-match lookup on all fields.
     pub fn lookup_eq(&self, field_values: &[&[u8]]) -> Vec<&[u8]> {
         let key = Self::build_key(field_values);

nodedb/src/engine/kv/index/field.rs

@@ -81,6 +81,13 @@ impl KvFieldIndex {
         }
     }
 
+    /// Whether the index contains the (value, primary_key) pair.
+    pub(crate) fn contains(&self, field_value: &[u8], primary_key: &[u8]) -> bool {
+        self.tree
+            .get(field_value)
+            .is_some_and(|keys| keys.contains(primary_key))
+    }
+
     /// Exact-match lookup: find all primary keys where field == value.
     pub fn lookup_eq(&self, field_value: &[u8]) -> Vec<&[u8]> {
         self.tree

nodedb/src/engine/kv/index/set.rs

@@ -23,6 +23,18 @@ pub struct KvIndexSet {
     total_index_writes: u64,
 }
 
+fn composite_vals<'a>(ci: &KvCompositeIndex, values: &'a [(&str, &[u8])]) -> Vec<&'a [u8]> {
+    ci.fields()
+        .iter()
+        .filter_map(|f| {
+            values
+                .iter()
+                .find(|(name, _)| *name == f.as_str())
+                .map(|(_, v)| *v)
+        })
+        .collect()
+}
+
 impl KvIndexSet {
     pub fn new() -> Self {
         Self {
@@ -155,62 +167,89 @@ impl KvIndexSet {
 
         let mut writes = 0;
 
-        // Remove old single-field index entries (if this is an update).
+        // Single-field indexes. Elide a no-op in-place update: when the old
+        // indexed value equals the new one *and* the pair is already present,
+        // remove+insert is pure churn. A row written before a backfill=false
+        // registration is absent from the index; the next PUT with identical
+        // bytes must insert it.
         if let Some(old_values) = old_field_values {
             for idx in &mut self.indexes {
-                for &(field, value) in old_values {
-                    if field == idx.field() {
-                        idx.remove(value, primary_key);
+                let f = idx.field();
+                let new_val = field_values
+                    .iter()
+                    .find(|(field, _)| *field == f)
+                    .map(|(_, v)| *v);
+                let old_val = old_values
+                    .iter()
+                    .find(|(field, _)| *field == f)
+                    .map(|(_, v)| *v);
+                if new_val == old_val {
+                    if let Some(v) = new_val {
+                        if idx.contains(v, primary_key) {
+                            continue; // true no-op: already indexed
+                        }
+                        // old == new but absent (backfill=false gap): insert only —
+                        // there's no stale entry to remove, so don't count/perform one.
+                        idx.insert(v.to_vec(), primary_key.to_vec());
                         writes += 1;
                     }
+                    continue;
                 }
-            }
-        }
-
-        // Insert new single-field index entries.
-        for idx in &mut self.indexes {
-            for &(field, value) in field_values {
-                if field == idx.field() {
-                    idx.insert(value.to_vec(), primary_key.to_vec());
+                if let Some(v) = old_val {
+                    idx.remove(v, primary_key);
+                    writes += 1;
+                }
+                if let Some(v) = new_val {
+                    idx.insert(v.to_vec(), primary_key.to_vec());
                     writes += 1;
                 }
             }
+        } else {
+            // Insert-only (new row): no old values to compare.
+            for idx in &mut self.indexes {
+                for &(field, value) in field_values {
+                    if field == idx.field() {
+                        idx.insert(value.to_vec(), primary_key.to_vec());
+                        writes += 1;
+                    }
+                }
+            }
         }
 
         // Maintain composite indexes.
         for ci in &mut self.composite_indexes {
-            // Remove old composite entry.
-            if let Some(old_values) = old_field_values {
-                let old_vals: Vec<&[u8]> = ci
-                    .fields()
-                    .iter()
-                    .filter_map(|f| {
-                        old_values
-                            .iter()
-                            .find(|(name, _)| *name == f.as_str())
-                            .map(|(_, v)| *v)
-                    })
-                    .collect();
-                if old_vals.len() == ci.fields().len() {
-                    ci.remove(&old_vals, primary_key);
-                    writes += 1;
+            // Composite entry, eliding the same no-op update case.
+            match old_field_values {
+                Some(old_values) => {
+                    let old_vals = composite_vals(ci, old_values);
+                    let new_vals = composite_vals(ci, field_values);
+                    if old_vals.len() == ci.fields().len() && new_vals.len() == ci.fields().len() {
+                        if old_vals == new_vals {
+                            if !ci.contains(&new_vals, primary_key) {
+                                ci.insert(&new_vals, primary_key.to_vec());
+                                writes += 1;
+                            }
+                        } else {
+                            ci.remove(&old_vals, primary_key);
+                            writes += 1;
+                            ci.insert(&new_vals, primary_key.to_vec());
+                            writes += 1;
+                        }
+                    } else if old_vals.len() == ci.fields().len() {
+                        ci.remove(&old_vals, primary_key);
+                        writes += 1;
+                    } else if new_vals.len() == ci.fields().len() {
+                        ci.insert(&new_vals, primary_key.to_vec());
+                        writes += 1;
+                    }
+                }
+                None => {
+                    let new_vals = composite_vals(ci, field_values);
+                    if new_vals.len() == ci.fields().len() {
+                        ci.insert(&new_vals, primary_key.to_vec());
+                        writes += 1;
+                    }
                 }
-            }
-
-            // Insert new composite entry.
-            let new_vals: Vec<&[u8]> = ci
-                .fields()
-                .iter()
-                .filter_map(|f| {
-                    field_values
-                        .iter()
-                        .find(|(name, _)| *name == f.as_str())
-                        .map(|(_, v)| *v)
-                })
-                .collect();
-            if new_vals.len() == ci.fields().len() {
-                ci.insert(&new_vals, primary_key.to_vec());
-                writes += 1;
             }
         }
 
@@ -334,6 +373,24 @@ mod tests {
         assert_eq!(set.lookup_eq("status", b"active").len(), 1);
     }
 
+    #[test]
+    fn identical_update_elides_index_writes() {
+        // An in-place update that leaves the indexed value unchanged must not
+        // rewrite the index when the pair is already present.
+        let mut set = KvIndexSet::new();
+        set.add_index("status", 0);
+        set.on_put(b"k1", &[("status", b"active")], None);
+        let writes = set.on_put(
+            b"k1",
+            &[("status", b"active")],
+            Some(&[("status", b"active")]),
+        );
+        assert_eq!(
+            writes, 0,
+            "identical update must elide index writes (got {writes})"
+        );
+    }
+
     #[test]
     fn index_set_on_put_update_replaces_old() {
         let mut set = KvIndexSet::new();
@@ -420,6 +477,46 @@ mod tests {
         assert!(ci.lookup_eq(&[b"x", b"y"]).is_empty());
     }
 
+    #[test]
+    fn backfill_absent_pair_still_gets_inserted() {
+        // old == new alone is not enough to elide — the pair must actually be
+        // present. A row written before a backfill=false registration was never
+        // indexed, so the next identical PUT has to insert it.
+        let mut set = KvIndexSet::new();
+        set.add_index("status", 0);
+        // No prior on_put — simulates backfill=false: this row exists but
+        // was never run through the indexer.
+        let writes = set.on_put(
+            b"k1",
+            &[("status", b"active")],
+            Some(&[("status", b"active")]), // old == new, but never indexed
+        );
+        assert_eq!(
+            writes, 1,
+            "backfill=false: absent pair must insert, got {writes}"
+        );
+        assert_eq!(set.lookup_eq("status", b"active"), vec![b"k1".as_slice()]);
+    }
+
+    #[test]
+    fn composite_backfill_absent_pair_still_gets_inserted() {
+        let mut set = KvIndexSet::new();
+        set.add_composite_index(vec!["region".into(), "status".into()], vec![0, 1]);
+        let writes = set.on_put(
+            b"k1",
+            &[("region", b"us-east"), ("status", b"active")],
+            Some(&[("region", b"us-east"), ("status", b"active")]),
+        );
+        assert_eq!(
+            writes, 1,
+            "composite backfill=false: absent pair must insert"
+        );
+        let ci = set
+            .get_composite_index(&["region".into(), "status".into()])
+            .expect("composite index was registered");
+        assert_eq!(ci.lookup_eq(&[b"us-east", b"active"]).len(), 1);
+    }
+
     /// The export accessors must see exactly the indexes that were registered —
     /// a checkpoint that iterated a partial view would publish rows whose index
     /// registrations are missing.