From 7ed14fc14c4380830cc0dfbec07365b1850b5bba Mon Sep 17 00:00:00 2001
From: joshua-spacetime <josh@clockworklabs.io>
Date: Sat, 14 Oct 2023 12:28:07 -0700
Subject: [PATCH] fix(430): Incremental evaluation for semijoins

Fixes #430.

For inserts, under set semantics:

The formula for incrementally evaluating a join before a transaction
has been applied is as follows

{A+ join B} U {A join B+} U {A+ join B+}

The formula for incrementally evaluating a join after a transaction
has been applied is as follows

{A+ join B} U {A join B+}

For deletes, under set semantics:

The formula for incrementally evaluating a join before a transaction
has been applied is as follows

{A- join B} U {A join B-}

The formula for incrementally evaluating a join after a transaction
has been applied is as follows

{A- join B} U {A join B-} U {A- join B-}

Note that SpacetimeDB is operating under the latter model.
Evaluation after the updates have already been applied to the underlying tables.
Therefore the 2nd and 4th formulas apply.
---
 crates/core/src/subscription/subscription.rs | 17 +++--------------
 1 file changed, 3 insertions(+), 14 deletions(-)

diff --git a/crates/core/src/subscription/subscription.rs b/crates/core/src/subscription/subscription.rs
index 48c94e729b..0398e0d1ef 100644
--- a/crates/core/src/subscription/subscription.rs
+++ b/crates/core/src/subscription/subscription.rs
@@ -402,16 +402,9 @@ impl<'a> IncrementalJoin<'a> {
                         }
                     })
                 });
-            // {A+ join B+}
-            let c = eval_incremental(db, tx, auth, &query::to_mem_table(rhs_virt, &self.lhs.inserts()))?;
-
-            // ({A+ join B} U {A join B+}) \ {A+ join B+ }
+            // {A+ join B} U {A join B+}
             let mut set = a.map(|op| (op.row_pk, op)).collect::<HashMap<PrimaryKey, Op>>();
             set.extend(b.map(|op| (op.row_pk, op)));
-            for op in c {
-                set.remove(&op.row_pk);
-            }
-
             set
         };
         let mut deletes = {
@@ -435,14 +428,10 @@ impl<'a> IncrementalJoin<'a> {
                 });
             // {A- join B-}
             let c = eval_incremental(db, tx, auth, &query::to_mem_table(rhs_virt, &self.lhs.deletes()))?;
-
-            // ({A- join B} U {A join B-}) \ {A- join B-}
+            // {A- join B} U {A join B-} U {A- join B-}
             let mut set = a.map(|op| (op.row_pk, op)).collect::<HashMap<PrimaryKey, Op>>();
             set.extend(b.map(|op| (op.row_pk, op)));
-            for op in c {
-                set.remove(&op.row_pk);
-            }
-
+            set.extend(c.map(|op| (op.row_pk, op)));
             set
         };