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

@@ -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
         };