Fix bug for subcription that was not executing or returning if empty (#6)

* Fix bug for subcription that was not executing or returning if empty

* Clippy

* Filter out deleted rows

* Recover the op_type (DELETED, INSERTED) flag for the row in the subscription

Author: mamcx

crates/core/src/subscription/query.rs

@@ -2,7 +2,8 @@ use crate::db::relational_db::RelationalDB;
 use crate::error::{DBError, SubscriptionError};
 use crate::host::module_host::DatabaseTableUpdate;
 use crate::sql::execute::{compile_sql, execute_single_sql};
-use spacetimedb_sats::relation::MemTable;
+use spacetimedb_sats::relation::{Column, FieldName, MemTable};
+use spacetimedb_sats::AlgebraicType;
 use spacetimedb_vm::expr::{Crud, CrudExpr, DbType, QueryExpr, SourceExpr};
 
 pub enum QueryDef {
@@ -28,16 +29,25 @@ impl Query {
     }
 }
 
+pub const OP_TYPE_FIELD_NAME: &str = "__op_type";
+
+//HACK: To recover the `op_type` of this particular row I add a "hidden" column `OP_TYPE_FIELD_NAME`
 pub fn to_mem_table(of: QueryExpr, data: &DatabaseTableUpdate) -> QueryExpr {
     let mut q = of;
 
     let mut t = match &q.source {
         SourceExpr::MemTable(x) => MemTable::new(&x.head, &[]),
         SourceExpr::DbTable(table) => MemTable::new(&table.head, &[]),
     };
+    t.head.fields.push(Column::new(
+        FieldName::named(&t.head.table_name, OP_TYPE_FIELD_NAME),
+        AlgebraicType::U8,
+    ));
 
     for row in &data.ops {
-        t.data.push(row.row.clone());
+        let mut new = row.row.clone();
+        new.elements.push(row.op_type.into());
+        t.data.push(new);
     }
 
     q.source = SourceExpr::MemTable(t);
@@ -87,10 +97,11 @@ mod tests {
     use crate::subscription::subscription::QuerySet;
     use crate::vm::tests::create_table_from_program;
     use crate::vm::DbProgram;
+    use itertools::Itertools;
     use spacetimedb_lib::data_key::ToDataKey;
     use spacetimedb_lib::error::ResultTest;
     use spacetimedb_sats::relation::FieldName;
-    use spacetimedb_sats::{product, BuiltinType, ProductType, ProductValue};
+    use spacetimedb_sats::{product, BuiltinType, ProductType};
     use spacetimedb_vm::dsl::{db_table, mem_table, scalar};
     use spacetimedb_vm::operator::OpCmp;
 
@@ -102,6 +113,7 @@ mod tests {
         let p = &mut DbProgram::new(&db, &mut tx);
 
         let head = ProductType::from_iter([("inventory_id", BuiltinType::U64), ("name", BuiltinType::String)]);
+
         let row = product!(1u64, "health");
         let table = mem_table(head.clone(), [row.clone()]);
         let table_id = create_table_from_program(p, "inventory", head.clone(), &[row.clone()])?;
@@ -110,7 +122,7 @@ mod tests {
         db.commit_tx(tx)?;
 
         let op = TableOp {
-            op_type: 0,
+            op_type: 1,
             row_pk: vec![],
             row,
         };
@@ -120,7 +132,13 @@ mod tests {
             table_name: "inventory".to_string(),
             ops: vec![op.clone()],
         };
-        let q = QueryExpr::new(db_table((&schema).into(), "inventory", table_id));
+        // For filtering out the hidden field `OP_TYPE_FIELD_NAME`
+        let fields = &[
+            FieldName::named("inventory", "inventory_id").into(),
+            FieldName::named("inventory", "name").into(),
+        ];
+
+        let q = QueryExpr::new(db_table((&schema).into(), "inventory", table_id)).with_project(fields);
 
         let q = to_mem_table(q, &data);
         let result = run_query(&db, &q)?;
@@ -136,16 +154,14 @@ mod tests {
             ops: vec![op],
         };
 
-        let q = QueryExpr::new(db_table((&schema).into(), "inventory", table_id)).with_select_cmp(
-            OpCmp::Eq,
-            FieldName::named("inventory", "inventory_id"),
-            scalar(0),
-        );
+        let q = QueryExpr::new(db_table((&schema).into(), "inventory", table_id))
+            .with_select_cmp(OpCmp::Eq, FieldName::named("inventory", "inventory_id"), scalar(1u64))
+            .with_project(fields);
 
         let q = to_mem_table(q, &data);
         let result = run_query(&db, &q)?;
 
-        let table = mem_table(head, Vec::<ProductValue>::new());
+        let table = mem_table(head, vec![product!(1u64, "health")]);
         assert_eq!(
             Some(table.as_without_table_name()),
             result.first().map(|x| x.as_without_table_name())
@@ -192,8 +208,12 @@ mod tests {
         ]);
 
         let result = s.eval(&db)?;
-        assert_eq!(result.tables.len(), 1, "Must return 1 table");
-        assert_eq!(result.tables[0].ops.len(), 1, "Must return 1 row");
+        assert_eq!(result.tables.len(), 3, "Must return 3 tables");
+        assert_eq!(
+            result.tables.iter().map(|x| x.ops.len()).sum::<usize>(),
+            1,
+            "Must return 1 row"
+        );
         assert_eq!(result.tables[0].ops[0].row, row, "Must return the correct row");
 
         Ok(())
@@ -251,10 +271,77 @@ mod tests {
         let update = DatabaseUpdate { tables: vec![data] };
 
         let result = s.eval_incr(&db, &update)?;
-        assert_eq!(result.tables.len(), 1, "Must return 1 table");
-        assert_eq!(result.tables[0].ops.len(), 1, "Must return 1 row");
+        assert_eq!(result.tables.len(), 3, "Must return 3 tables");
+        assert_eq!(
+            result.tables.iter().map(|x| x.ops.len()).sum::<usize>(),
+            1,
+            "Must return 1 row"
+        );
         assert_eq!(result.tables[0].ops[0].row, row, "Must return the correct row");
 
         Ok(())
     }
+
+    //Check that
+    //```
+    //SELECT * FROM table1
+    //SELECT * FROM table2
+    // =
+    //SELECT * FROM table2
+    //SELECT * FROM table1
+    //```
+    // return just one row irrespective of the order of the queries
+    #[test]
+    fn test_subscribe_commutative() -> ResultTest<()> {
+        let (db, _tmp_dir) = make_test_db()?;
+        let mut tx = db.begin_tx();
+        let p = &mut DbProgram::new(&db, &mut tx);
+
+        let head_1 = ProductType::from_iter([("inventory_id", BuiltinType::U64), ("name", BuiltinType::String)]);
+        let row_1 = product!(1u64, "health");
+        let table_id_1 = create_table_from_program(p, "inventory", head_1.clone(), &[row_1.clone()])?;
+
+        let head_2 = ProductType::from_iter([("player_id", BuiltinType::U64), ("name", BuiltinType::String)]);
+        let row_2 = product!(2u64, "jhon doe");
+        let table_id_2 = create_table_from_program(p, "player", head_2, &[row_2.clone()])?;
+
+        let schema_1 = db.schema_for_table(&tx, table_id_1).unwrap();
+        let schema_2 = db.schema_for_table(&tx, table_id_2).unwrap();
+        db.commit_tx(tx)?;
+
+        let q_1 = QueryExpr::new(db_table((&schema_1).into(), "inventory", table_id_1));
+        let q_2 = QueryExpr::new(db_table((&schema_2).into(), "player", table_id_2));
+
+        let s = QuerySet(vec![
+            Query {
+                queries: vec![q_1.clone()],
+            },
+            Query {
+                queries: vec![q_2.clone()],
+            },
+        ]);
+
+        let result_1 = s.eval(&db)?;
+
+        let s = QuerySet(vec![
+            Query {
+                queries: vec![q_2.clone()],
+            },
+            Query { queries: vec![q_1] },
+        ]);
+
+        let result_2 = s.eval(&db)?;
+        let to_row = |of: DatabaseUpdate| {
+            of.tables
+                .iter()
+                .map(|x| x.ops.iter().map(|x| x.row.clone()))
+                .flatten()
+                .sorted()
+                .collect::<Vec<_>>()
+        };
+
+        assert_eq!(to_row(result_1), to_row(result_2));
+
+        Ok(())
+    }
 }

crates/core/src/subscription/subscription.rs

@@ -1,8 +1,9 @@
+use spacetimedb_sats::{AlgebraicValue, BuiltinValue};
 use std::collections::HashSet;
 
 use super::query::Query;
 use crate::error::DBError;
-use crate::subscription::query::run_query;
+use crate::subscription::query::{run_query, OP_TYPE_FIELD_NAME};
 use crate::{
     client::{ClientActorId, ClientConnectionSender},
     db::relational_db::RelationalDB,
@@ -60,32 +61,35 @@ impl QuerySet {
         let mut seen = HashSet::new();
 
         for query in &self.0 {
-            for table in &database_update.tables {
-                let mut rows = Vec::with_capacity(table.ops.len());
-                for row in &table.ops {
-                    if seen.contains(&(table.table_id, row.row_pk.clone())) {
-                        continue;
-                    }
-                    seen.insert((table.table_id, row.row_pk.clone()));
-                    rows.push(row.clone());
-                }
-
-                if rows.is_empty() {
-                    continue;
-                }
+            for table in database_update.tables.iter().cloned() {
+                for q in query.queries_of_table_id(&table) {
+                    if let Some(result) = run_query(relational_db, &q)?.into_iter().find(|x| !x.data.is_empty()) {
+                        let mut table_row_operations = table.clone();
+                        table_row_operations.ops.clear();
+                        for mut row in result.data {
+                            //Hack: remove the hidden field OP_TYPE_FIELD_NAME. see `to_mem_table`
+                            // needs to be done before calculate the PK
+                            let op_type =
+                                if let Some(AlgebraicValue::Builtin(BuiltinValue::U8(op))) = row.elements.pop() {
+                                    op
+                                } else {
+                                    panic!("Fail to extract {OP_TYPE_FIELD_NAME}")
+                                };
+
+                            let row_pk = RelationalDB::pk_for_row(&row);
+
+                            //Skip rows that are already resolved in a previous subscription...
+                            if seen.contains(&(table.table_id, row_pk)) {
+                                continue;
+                            }
 
-                let table = DatabaseTableUpdate {
-                    table_id: table.table_id,
-                    table_name: table.table_name.clone(),
-                    ops: rows,
-                };
+                            seen.insert((table.table_id, row_pk));
 
-                for q in query.queries_of_table_id(&table) {
-                    let result = run_query(relational_db, &q)?.into_iter().find(|x| !x.data.is_empty());
-                    if result.is_none() {
-                        continue;
+                            let row_pk = row_pk.to_bytes();
+                            table_row_operations.ops.push(TableOp { op_type, row_pk, row });
+                        }
+                        output.tables.push(table_row_operations);
                     }
-                    output.tables.push(table.clone());
                 }
             }
         }
@@ -128,9 +132,9 @@ impl QuerySet {
                                 });
                             }
 
-                            if table_row_operations.is_empty() {
-                                continue;
-                            }
+                            // if table_row_operations.is_empty() {
+                            //     continue;
+                            // }
 
                             database_update.tables.push(DatabaseTableUpdate {
                                 table_id: t.table_id,