perf(832): Remove redundant row deduplication in subscriptions

Closes #832.

The database already operates under set semantics,
so unless multiple queries return rows from the same table,
deduplication of the result set is not necessary.

Author: joshua-spacetime

crates/bench/benches/subscription.rs

@@ -4,6 +4,7 @@ use spacetimedb::error::DBError;
 use spacetimedb::execution_context::ExecutionContext;
 use spacetimedb::host::module_host::{DatabaseTableUpdate, DatabaseUpdate, TableOp};
 use spacetimedb::subscription::query::compile_read_only_query;
+use spacetimedb::subscription::subscription::ExecutionSet;
 use spacetimedb_lib::identity::AuthCtx;
 use spacetimedb_primitives::TableId;
 use spacetimedb_sats::{product, AlgebraicType, AlgebraicValue, ProductValue, ToDataKey};
@@ -110,6 +111,7 @@ fn eval(c: &mut Criterion) {
         let auth = AuthCtx::for_testing();
         let tx = db.begin_tx();
         let query = compile_read_only_query(&db, &tx, &auth, scan).unwrap();
+        let query: ExecutionSet = query.into();
 
         b.iter(|| {
             let out = query.eval(&db, &tx, auth).unwrap();
@@ -131,6 +133,7 @@ fn eval(c: &mut Criterion) {
         let auth = AuthCtx::for_testing();
         let tx = db.begin_tx();
         let query = compile_read_only_query(&db, &tx, &auth, &join).unwrap();
+        let query: ExecutionSet = query.into();
 
         b.iter(|| {
             let out = query.eval(&db, &tx, AuthCtx::for_testing()).unwrap();
@@ -148,9 +151,7 @@ fn eval(c: &mut Criterion) {
         let tx = db.begin_tx();
         let query_lhs = compile_read_only_query(&db, &tx, &auth, select_lhs).unwrap();
         let query_rhs = compile_read_only_query(&db, &tx, &auth, select_rhs).unwrap();
-
-        let mut query = query_lhs;
-        query.extend(query_rhs);
+        let query = ExecutionSet::from_iter(query_lhs.into_iter().chain(query_rhs));
 
         b.iter(|| {
             let out = query.eval_incr(&db, &tx, &update, AuthCtx::for_testing()).unwrap();
@@ -171,6 +172,7 @@ fn eval(c: &mut Criterion) {
         let auth = AuthCtx::for_testing();
         let tx = db.begin_tx();
         let query = compile_read_only_query(&db, &tx, &auth, &join).unwrap();
+        let query: ExecutionSet = query.into();
 
         b.iter(|| {
             let out = query.eval_incr(&db, &tx, &update, AuthCtx::for_testing()).unwrap();

crates/core/src/subscription/module_subscription_actor.rs

@@ -2,7 +2,7 @@ use std::sync::Arc;
 
 use super::{
     query::compile_read_only_query,
-    subscription::{QuerySet, Subscription},
+    subscription::{ExecutionSet, Subscription},
 };
 use crate::client::{
     messages::{CachedMessage, SubscriptionUpdateMessage, TransactionUpdateMessage},
@@ -45,12 +45,15 @@ impl ModuleSubscriptions {
         });
 
         let auth = AuthCtx::new(self.owner_identity, sender.id.identity);
-        let mut queries = QuerySet::new();
+        let mut queries = Vec::new();
         for sql in subscription.query_strings {
             let qset = compile_read_only_query(&self.relational_db, tx, &auth, &sql)?;
             queries.extend(qset);
         }
 
+        let n = queries.len();
+        let queries = ExecutionSet::from_iter(queries);
+
         let database_update = tokio::task::block_in_place(|| queries.eval(&self.relational_db, tx, auth))?;
         // It acquires the subscription lock after `eval`, allowing `add_subscription` to run concurrently.
         // This also makes it possible for `broadcast_event` to get scheduled before the subsequent part here
@@ -63,7 +66,6 @@ impl ModuleSubscriptions {
                 sub
             }
             None => {
-                let n = queries.len();
                 subscriptions.push(Subscription::new(queries, sender));
                 WORKER_METRICS
                     .subscription_queries
@@ -100,7 +102,7 @@ impl ModuleSubscriptions {
                 WORKER_METRICS
                     .subscription_queries
                     .with_label_values(&self.relational_db.address())
-                    .sub(subscription.queries.len() as i64);
+                    .sub(subscription.queries.num_queries() as i64);
             }
             !subscription.subscribers().is_empty()
         });

crates/core/src/subscription/query.rs

@@ -7,7 +7,7 @@ use crate::execution_context::{ExecutionContext, WorkloadType};
 use crate::host::module_host::DatabaseTableUpdate;
 use crate::sql::compiler::compile_sql;
 use crate::sql::execute::execute_single_sql;
-use crate::subscription::subscription::{QuerySet, SupportedQuery};
+use crate::subscription::subscription::SupportedQuery;
 use once_cell::sync::Lazy;
 use regex::Regex;
 use spacetimedb_lib::identity::AuthCtx;
@@ -19,6 +19,8 @@ use spacetimedb_sats::DataKey;
 use spacetimedb_vm::expr;
 use spacetimedb_vm::expr::{Crud, CrudExpr, DbType, QueryExpr};
 
+use super::subscription::get_all;
+
 static WHITESPACE: Lazy<Regex> = Lazy::new(|| Regex::new(r"\s+").unwrap());
 pub const SUBSCRIBE_TO_ALL_QUERY: &str = "SELECT * FROM *";
 
@@ -103,7 +105,7 @@ pub fn compile_read_only_query(
     tx: &Tx,
     auth: &AuthCtx,
     input: &str,
-) -> Result<QuerySet, DBError> {
+) -> Result<Vec<SupportedQuery>, DBError> {
     let input = input.trim();
     if input.is_empty() {
         return Err(SubscriptionError::Empty.into());
@@ -112,7 +114,7 @@ pub fn compile_read_only_query(
     // Remove redundant whitespace, and in particular newlines, for debug info.
     let input = WHITESPACE.replace_all(input, " ");
     if input == SUBSCRIBE_TO_ALL_QUERY {
-        return QuerySet::get_all(relational_db, tx, auth);
+        return get_all(relational_db, tx, auth);
     }
 
     let compiled = compile_sql(relational_db, tx, &input)?;
@@ -204,6 +206,7 @@ mod tests {
     use crate::db::relational_db::MutTx;
     use crate::host::module_host::{DatabaseUpdate, TableOp};
     use crate::sql::execute::run;
+    use crate::subscription::subscription::ExecutionSet;
     use crate::vm::tests::create_table_with_rows;
     use itertools::Itertools;
     use spacetimedb_lib::error::ResultTest;
@@ -359,7 +362,7 @@ mod tests {
     fn check_query_incr(
         db: &RelationalDB,
         tx: &Tx,
-        s: &QuerySet,
+        s: &ExecutionSet,
         update: &DatabaseUpdate,
         total_tables: usize,
         rows: &[ProductValue],
@@ -381,7 +384,7 @@ mod tests {
     fn check_query_eval(
         db: &RelationalDB,
         tx: &Tx,
-        s: &QuerySet,
+        s: &ExecutionSet,
         total_tables: usize,
         rows: &[ProductValue],
     ) -> ResultTest<()> {
@@ -414,7 +417,7 @@ mod tests {
         let table_id = create_table_with_rows(&db, &mut tx, "test", schema.clone(), &[])?;
 
         // select * from test
-        let query: QuerySet = QueryExpr::new(db_table(schema.clone(), table_id)).try_into()?;
+        let query: ExecutionSet = QueryExpr::new(db_table(schema.clone(), table_id)).try_into()?;
 
         let op = TableOp {
             op_type: 0,
@@ -481,7 +484,7 @@ mod tests {
             panic!("unexpected query {:#?}", exp[0]);
         };
 
-        let query = QuerySet::try_from(query)?;
+        let query: ExecutionSet = query.try_into()?;
 
         let result = query.eval_incr(&db, &tx, &update, AuthCtx::for_testing())?;
 
@@ -546,7 +549,8 @@ mod tests {
             panic!("unexpected query {:#?}", exp[0]);
         };
 
-        let query = QuerySet::try_from(query)?;
+        let query: ExecutionSet = query.try_into()?;
+
         db.release_tx(&ExecutionContext::default(), tx);
 
         fn case_env(
@@ -825,7 +829,7 @@ mod tests {
         let s = [q_all, q_id]
             .into_iter()
             .map(TryFrom::try_from)
-            .collect::<Result<QuerySet, _>>()?;
+            .collect::<Result<ExecutionSet, _>>()?;
 
         let row1 = TableOp {
             op_type: 0,
@@ -899,7 +903,8 @@ mod tests {
         let s = [q_all, q_id]
             .into_iter()
             .map(TryFrom::try_from)
-            .collect::<Result<QuerySet, _>>()?;
+            .collect::<Result<ExecutionSet, _>>()?;
+
         db.commit_tx(&ExecutionContext::default(), tx)?;
 
         let tx = db.begin_tx();
@@ -1009,7 +1014,7 @@ mod tests {
         let row_1 = product!(1u64, "health");
         let row_2 = product!(2u64, "jhon doe");
         let tx = db.begin_tx();
-        let s = compile_read_only_query(&db, &tx, &AuthCtx::for_testing(), SUBSCRIBE_TO_ALL_QUERY)?;
+        let s = compile_read_only_query(&db, &tx, &AuthCtx::for_testing(), SUBSCRIBE_TO_ALL_QUERY)?.into();
         check_query_eval(&db, &tx, &s, 2, &[row_1.clone(), row_2.clone()])?;
 
         let row1 = TableOp {
@@ -1078,14 +1083,14 @@ mod tests {
             "SELECT * FROM lhs WHERE id > 5",
         ];
         for scan in scans {
-            let expr = compile_read_only_query(&db, &tx, &auth, scan)?.pop_first().unwrap();
+            let expr = compile_read_only_query(&db, &tx, &auth, scan)?.pop().unwrap();
             assert_eq!(expr.kind(), Supported::Scan, "{scan}\n{expr:#?}");
         }
 
         // Only index semijoins are supported
         let joins = ["SELECT lhs.* FROM lhs JOIN rhs ON lhs.id = rhs.id WHERE rhs.y < 10"];
         for join in joins {
-            let expr = compile_read_only_query(&db, &tx, &auth, join)?.pop_first().unwrap();
+            let expr = compile_read_only_query(&db, &tx, &auth, join)?.pop().unwrap();
             assert_eq!(expr.kind(), Supported::Semijoin, "{join}\n{expr:#?}");
         }