Remove redundant rows from subscription updates

Author: joshua-spacetime

crates/core/src/subscription/delta.rs

@@ -1,7 +1,9 @@
 use anyhow::Result;
+use hashbrown::HashMap;
 use spacetimedb_execution::{Datastore, DeltaStore};
 use spacetimedb_lib::metrics::ExecutionMetrics;
 use spacetimedb_subscription::SubscriptionPlan;
+use spacetimedb_vm::relation::RelValue;
 
 use crate::host::module_host::UpdatesRelValue;
 
@@ -21,19 +23,37 @@ pub fn eval_delta<'a, Tx: Datastore + DeltaStore>(
     plan: &SubscriptionPlan,
 ) -> Result<Option<UpdatesRelValue<'a>>> {
     metrics.delta_queries_evaluated += 1;
-    let mut inserts = vec![];
-    let mut deletes = vec![];
+
+    let mut insert_counts = HashMap::new();
+    let mut delete_counts = HashMap::new();
 
     plan.for_each_insert(tx, metrics, &mut |row| {
-        inserts.push(row.into());
+        *insert_counts.entry(row).or_default() += 1;
         Ok(())
     })?;
 
     plan.for_each_delete(tx, metrics, &mut |row| {
-        deletes.push(row.into());
+        match insert_counts.get_mut(&row) {
+            None | Some(0) => {
+                *delete_counts.entry(row).or_default() += 1;
+            }
+            Some(n) => {
+                *n -= 1;
+            }
+        }
         Ok(())
     })?;
 
+    let mut inserts = vec![];
+    let mut deletes = vec![];
+
+    for (row, n) in insert_counts.into_iter().filter(|(_, n)| *n > 0) {
+        inserts.extend(std::iter::repeat_n(row, n).map(RelValue::from));
+    }
+    for (row, n) in delete_counts.into_iter().filter(|(_, n)| *n > 0) {
+        deletes.extend(std::iter::repeat_n(row, n).map(RelValue::from));
+    }
+
     // Return `None` for empty updates
     if inserts.is_empty() && deletes.is_empty() {
         return Ok(None);

crates/core/src/subscription/module_subscription_actor.rs

@@ -1692,7 +1692,7 @@ mod tests {
 
         // We should only have evaluated a single query
         assert_eq!(metrics.delta_queries_evaluated, 1);
-        assert_eq!(metrics.delta_queries_matched, 1);
+        assert_eq!(metrics.delta_queries_matched, 0);
 
         // Insert a new row into `v`
         let metrics = commit_tx(&db, &subs, [], [(v_id, product![2u64, 6u64, 6u64])])?;

crates/execution/src/lib.rs

@@ -1,4 +1,7 @@
-use std::ops::RangeBounds;
+use std::{
+    hash::{Hash, Hasher},
+    ops::RangeBounds,
+};
 
 use anyhow::{anyhow, Result};
 use iter::PlanIter;
@@ -95,12 +98,21 @@ pub trait DeltaStore {
     }
 }
 
-#[derive(Clone)]
+#[derive(Clone, PartialEq, Eq)]
 pub enum Row<'a> {
     Ptr(RowRef<'a>),
     Ref(&'a ProductValue),
 }
 
+impl Hash for Row<'_> {
+    fn hash<H: Hasher>(&self, state: &mut H) {
+        match self {
+            Self::Ptr(x) => x.hash(state),
+            Self::Ref(x) => x.hash(state),
+        }
+    }
+}
+
 impl Row<'_> {
     pub fn to_product_value(&self) -> ProductValue {
         match self {