fix expression canonicalization

crates/core/src/subscription/module_subscription_actor.rs

@@ -744,6 +744,7 @@ mod tests {
         CompressableQueryUpdate, Compression, FormatSwitch, QueryId, Subscribe, SubscribeMulti, SubscribeSingle,
         Unsubscribe, UnsubscribeMulti,
     };
+    use spacetimedb_execution::dml::MutDatastore;
     use spacetimedb_lib::bsatn::ToBsatn;
     use spacetimedb_lib::db::auth::StAccess;
     use spacetimedb_lib::identity::AuthCtx;
@@ -994,6 +995,7 @@ mod tests {
                         .filter(|(_, n)| n > &&0)
                         .map(|(row, _)| row)
                         .cloned()
+                        .sorted()
                         .collect::<Vec<_>>(),
                     inserts.into_iter().sorted().collect::<Vec<_>>()
                 );
@@ -1003,6 +1005,7 @@ mod tests {
                         .filter(|(_, n)| n < &&0)
                         .map(|(row, _)| row)
                         .cloned()
+                        .sorted()
                         .collect::<Vec<_>>(),
                     deletes.into_iter().sorted().collect::<Vec<_>>()
                 );
@@ -1012,6 +1015,27 @@ mod tests {
         }
     }
 
+    /// Commit a set of row updates and broadcast to subscribers
+    fn commit_tx(
+        db: &RelationalDB,
+        subs: &ModuleSubscriptions,
+        deletes: impl IntoIterator<Item = (TableId, ProductValue)>,
+        inserts: impl IntoIterator<Item = (TableId, ProductValue)>,
+    ) -> anyhow::Result<()> {
+        let mut tx = db.begin_mut_tx(IsolationLevel::Serializable, Workload::ForTests);
+        for (table_id, row) in deletes {
+            tx.delete_product_value(table_id, &row)?;
+        }
+        for (table_id, row) in inserts {
+            db.insert(&mut tx, table_id, &bsatn::to_vec(&row)?)?;
+        }
+        assert!(matches!(
+            subs.commit_and_broadcast_event(None, module_event(), tx),
+            Ok(Ok(_))
+        ));
+        Ok(())
+    }
+
     #[test]
     fn test_subscribe_metrics() -> anyhow::Result<()> {
         let client_id = client_id_from_u8(1);
@@ -1456,6 +1480,119 @@ mod tests {
         Ok(())
     }
 
+    /// In this test we subscribe to a join query, update the lhs table,
+    /// and assert that the server sends the correct delta to the client.
+    #[tokio::test]
+    async fn test_update_for_join() -> anyhow::Result<()> {
+        async fn test_subscription_updates(queries: &[&'static str]) -> anyhow::Result<()> {
+            // Establish a client connection
+            let (sender, mut rx) = client_connection(client_id_from_u8(1));
+
+            let db = relational_db()?;
+            let subs = module_subscriptions(db.clone());
+
+            let p_schema = [("id", AlgebraicType::U64), ("signed_in", AlgebraicType::Bool)];
+            let l_schema = [
+                ("id", AlgebraicType::U64),
+                ("x", AlgebraicType::U64),
+                ("z", AlgebraicType::U64),
+            ];
+
+            let p_id = db.create_table_for_test("p", &p_schema, &[0.into()])?;
+            let l_id = db.create_table_for_test("l", &l_schema, &[0.into()])?;
+
+            subscribe_multi(&subs, queries, sender, &mut 0)?;
+
+            assert!(matches!(rx.recv().await, Some(SerializableMessage::Subscription(_))));
+
+            // Insert two matching player rows
+            commit_tx(
+                &db,
+                &subs,
+                [],
+                [
+                    (p_id, product![1_u64, true]),
+                    (p_id, product![2_u64, true]),
+                    (l_id, product![1_u64, 2_u64, 2_u64]),
+                    (l_id, product![2_u64, 3_u64, 3_u64]),
+                ],
+            )?;
+
+            let schema = ProductType::from(p_schema);
+
+            // We should receive both matching player rows
+            assert_tx_update_for_table(
+                &mut rx,
+                p_id,
+                &schema,
+                [product![1_u64, true], product![2_u64, true]],
+                [],
+            )
+            .await;
+
+            // Update one of the matching player rows
+            commit_tx(
+                &db,
+                &subs,
+                [(p_id, product![2_u64, true])],
+                [(p_id, product![2_u64, false])],
+            )?;
+
+            // We should receive an update for it because it is still matching
+            assert_tx_update_for_table(
+                &mut rx,
+                p_id,
+                &schema,
+                [product![2_u64, false]],
+                [product![2_u64, true]],
+            )
+            .await;
+
+            // Update the the same matching player row
+            commit_tx(
+                &db,
+                &subs,
+                [(p_id, product![2_u64, false])],
+                [(p_id, product![2_u64, true])],
+            )?;
+
+            // We should receive an update for it because it is still matching
+            assert_tx_update_for_table(
+                &mut rx,
+                p_id,
+                &schema,
+                [product![2_u64, true]],
+                [product![2_u64, false]],
+            )
+            .await;
+
+            Ok(())
+        }
+
+        test_subscription_updates(&[
+            "select * from p where id = 1",
+            "select p.* from p join l on p.id = l.id where l.x > 0 and l.x < 5 and l.z > 0 and l.z < 5",
+        ])
+        .await?;
+        test_subscription_updates(&[
+            "select * from p where id = 1",
+            "select p.* from p join l on p.id = l.id where 0 < l.x and l.x < 5 and 0 < l.z and l.z < 5",
+        ])
+        .await?;
+        test_subscription_updates(&[
+            "select * from p where id = 1",
+            "select p.* from p join l on p.id = l.id where l.x > 0 and l.x < 5 and l.x > 0 and l.z < 5 and l.id != 1",
+        ])
+        .await?;
+        test_subscription_updates(&[
+            "select * from p where id = 1",
+            "select p.* from p join l on p.id = l.id where 0 < l.x and l.x < 5 and 0 < l.z and l.z < 5 and l.id != 1",
+        ])
+        .await?;
+
+        Ok(())
+    }
+
     /// Asserts that a subscription holds a tx handle for the entire length of its evaluation.
     #[test]
     fn test_tx_subscription_ordering() -> ResultTest<()> {

crates/expr/src/lib.rs

@@ -121,16 +121,22 @@ pub(crate) fn type_expr(vars: &Relvars, expr: SqlExpr, expected: Option<&Algebra
             let b = type_expr(vars, *b, Some(&AlgebraicType::Bool))?;
             Ok(Expr::LogOp(op, Box::new(a), Box::new(b)))
         }
-        (SqlExpr::Bin(a, b, op), None | Some(AlgebraicType::Bool)) => match (*a, *b) {
-            (a, b @ SqlExpr::Lit(_)) | (b @ SqlExpr::Lit(_), a) | (a, b) => {
-                let a = type_expr(vars, a, None)?;
-                let b = type_expr(vars, b, Some(a.ty()))?;
-                if !op_supports_type(op, a.ty()) {
-                    return Err(InvalidOp::new(op, a.ty()).into());
-                }
-                Ok(Expr::BinOp(op, Box::new(a), Box::new(b)))
+        (SqlExpr::Bin(a, b, op), None | Some(AlgebraicType::Bool)) if matches!(&*a, SqlExpr::Lit(_)) => {
+            let b = type_expr(vars, *b, None)?;
+            let a = type_expr(vars, *a, Some(b.ty()))?;
+            if !op_supports_type(op, a.ty()) {
+                return Err(InvalidOp::new(op, a.ty()).into());
+            }
+            Ok(Expr::BinOp(op, Box::new(a), Box::new(b)))
+        }
+        (SqlExpr::Bin(a, b, op), None | Some(AlgebraicType::Bool)) => {
+            let a = type_expr(vars, *a, None)?;
+            let b = type_expr(vars, *b, Some(a.ty()))?;
+            if !op_supports_type(op, a.ty()) {
+                return Err(InvalidOp::new(op, a.ty()).into());
             }
-        },
+            Ok(Expr::BinOp(op, Box::new(a), Box::new(b)))
+        }
         (SqlExpr::Bin(..) | SqlExpr::Log(..), Some(ty)) => Err(UnexpectedType::new(&AlgebraicType::Bool, ty).into()),
         // Both unqualified names as well as parameters are syntactic constructs.
         // Unqualified names are qualified and parameters are resolved before type checking.

crates/physical-plan/src/plan.rs

@@ -517,7 +517,14 @@ impl PhysicalPlan {
                     PhysicalExpr::BinOp(op, value, expr)
                         if matches!(&*value, PhysicalExpr::Value(_)) && matches!(&*expr, PhysicalExpr::Field(..)) =>
                     {
-                        PhysicalExpr::BinOp(op, expr, value)
+                        match op {
+                            BinOp::Eq => PhysicalExpr::BinOp(BinOp::Eq, expr, value),
+                            BinOp::Ne => PhysicalExpr::BinOp(BinOp::Ne, expr, value),
+                            BinOp::Lt => PhysicalExpr::BinOp(BinOp::Gt, expr, value),
+                            BinOp::Gt => PhysicalExpr::BinOp(BinOp::Lt, expr, value),
+                            BinOp::Lte => PhysicalExpr::BinOp(BinOp::Gte, expr, value),
+                            BinOp::Gte => PhysicalExpr::BinOp(BinOp::Lte, expr, value),
+                        }
                     }
                     _ => expr,
                 };

crates/sdk/tests/test-client/src/main.rs

@@ -123,6 +123,8 @@ fn main() {
         "row-deduplication" => exec_row_deduplication(),
         "row-deduplication-join-r-and-s" => exec_row_deduplication_join_r_and_s(),
         "row-deduplication-r-join-s-and-r-joint" => exec_row_deduplication_r_join_s_and_r_join_t(),
+        "test-lhs-join-update" => test_lhs_join_update(),
+        "test-lhs-join-update-disjoint-queries" => test_lhs_join_update_disjoint_queries(),
         "test-intra-query-bag-semantics-for-join" => test_intra_query_bag_semantics_for_join(),
         "two-different-compression-algos" => exec_two_different_compression_algos(),
         "test-parameterized-subscription" => test_parameterized_subscription(),
@@ -2094,6 +2096,118 @@ fn exec_row_deduplication_r_join_s_and_r_join_t() {
     assert_eq!(count_unique_u32_on_insert.load(Ordering::SeqCst), 1);
 }
 
+/// This test asserts that the correct callbacks are invoked when updating the lhs table of a join
+fn test_lhs_join_update() {
+    let insert_counter = TestCounter::new();
+    let update_counter = TestCounter::new();
+    let mut on_update_1 = Some(update_counter.add_test("on_update_1"));
+    let mut on_update_2 = Some(update_counter.add_test("on_update_2"));
+    let mut on_insert_1 = Some(insert_counter.add_test("on_insert_1"));
+    let mut on_insert_2 = Some(insert_counter.add_test("on_insert_2"));
+
+    let conn = Arc::new(connect_then(&update_counter, {
+        move |ctx| {
+            subscribe_these_then(
+                ctx,
+                &[
+                    "SELECT p.* FROM pk_u32 p WHERE n = 1",
+                    "SELECT p.* FROM pk_u32 p JOIN unique_u32 u ON p.n = u.n WHERE u.data > 0 AND u.data < 5",
+                ],
+                |_| {},
+            );
+        }
+    }));
+
+    conn.reducers.on_insert_pk_u_32(move |_, n, data| {
+        if *n == 1 && *data == 0 {
+            return put_result(&mut on_insert_1, Ok(()));
+        }
+        if *n == 2 && *data == 0 {
+            return put_result(&mut on_insert_2, Ok(()));
+        }
+        panic!("unexpected insert: pk_u32(n: {n}, data: {data})");
+    });
+
+    conn.reducers.on_update_pk_u_32(move |ctx, n, data| {
+        if *n == 2 && *data == 1 {
+            PkU32::update(ctx, 2, 0);
+            return put_result(&mut on_update_1, Ok(()));
+        }
+        if *n == 2 && *data == 0 {
+            return put_result(&mut on_update_2, Ok(()));
+        }
+        panic!("unexpected update: pk_u32(n: {n}, data: {data})");
+    });
+
+    // Add two pk_u32 rows to the subscription
+    conn.reducers.insert_pk_u_32(1, 0).unwrap();
+    conn.reducers.insert_pk_u_32(2, 0).unwrap();
+    conn.reducers.insert_unique_u_32(1, 3).unwrap();
+    conn.reducers.insert_unique_u_32(2, 4).unwrap();
+
+    // Wait for the subscription to be updated,
+    // then update one of the pk_u32 rows.
+    insert_counter.wait_for_all();
+    conn.reducers.update_pk_u_32(2, 1).unwrap();
+
+    // Wait for the second row update for pk_u32
+    update_counter.wait_for_all();
+}
+
+/// This test asserts that the correct callbacks are invoked when updating the lhs table of a join
+fn test_lhs_join_update_disjoint_queries() {
+    let insert_counter = TestCounter::new();
+    let update_counter = TestCounter::new();
+    let mut on_update_1 = Some(update_counter.add_test("on_update_1"));
+    let mut on_update_2 = Some(update_counter.add_test("on_update_2"));
+    let mut on_insert_1 = Some(insert_counter.add_test("on_insert_1"));
+    let mut on_insert_2 = Some(insert_counter.add_test("on_insert_2"));
+
+    let conn = Arc::new(connect_then(&update_counter, {
+        move |ctx| {
+            subscribe_these_then(ctx, &[
+                "SELECT p.* FROM pk_u32 p WHERE n = 1",
+                "SELECT p.* FROM pk_u32 p JOIN unique_u32 u ON p.n = u.n WHERE u.data > 0 AND u.data < 5 AND u.n != 1",
+            ], |_| {});
+        }
+    }));
+
+    conn.reducers.on_insert_pk_u_32(move |_, n, data| {
+        if *n == 1 && *data == 0 {
+            return put_result(&mut on_insert_1, Ok(()));
+        }
+        if *n == 2 && *data == 0 {
+            return put_result(&mut on_insert_2, Ok(()));
+        }
+        panic!("unexpected insert: pk_u32(n: {n}, data: {data})");
+    });
+
+    conn.reducers.on_update_pk_u_32(move |ctx, n, data| {
+        if *n == 2 && *data == 1 {
+            PkU32::update(ctx, 2, 0);
+            return put_result(&mut on_update_1, Ok(()));
+        }
+        if *n == 2 && *data == 0 {
+            return put_result(&mut on_update_2, Ok(()));
+        }
+        panic!("unexpected update: pk_u32(n: {n}, data: {data})");
+    });
+
+    // Add two pk_u32 rows to the subscription
+    conn.reducers.insert_pk_u_32(1, 0).unwrap();
+    conn.reducers.insert_pk_u_32(2, 0).unwrap();
+    conn.reducers.insert_unique_u_32(1, 3).unwrap();
+    conn.reducers.insert_unique_u_32(2, 4).unwrap();
+
+    // Wait for the subscription to be updated,
+    // then update one of the pk_u32 rows.
+    insert_counter.wait_for_all();
+    conn.reducers.update_pk_u_32(2, 1).unwrap();
+
+    // Wait for the second row update for pk_u32
+    update_counter.wait_for_all();
+}
+
 /// Test that when subscribing to a single join query,
 /// the server returns a bag of rows to the client - not a set.
 ///

crates/sdk/tests/test.rs

@@ -216,6 +216,16 @@ macro_rules! declare_tests_with_suffix {
                 make_test("row-deduplication-r-join-s-and-r-joint").run();
             }
 
+            #[test]
+            fn test_lhs_join_update() {
+                make_test("test-lhs-join-update").run()
+            }
+
+            #[test]
+            fn test_lhs_join_update_disjoint_queries() {
+                make_test("test-lhs-join-update-disjoint-queries").run()
+            }
+
             #[test]
             fn test_intra_query_bag_semantics_for_join() {
                 make_test("test-intra-query-bag-semantics-for-join").run()