clarify cost of dropping the updates in eval_incr (#1192)

crates/bench/benches/subscription.rs

@@ -3,7 +3,7 @@ use spacetimedb::client::Protocol;
 use spacetimedb::db::relational_db::RelationalDB;
 use spacetimedb::error::DBError;
 use spacetimedb::execution_context::ExecutionContext;
-use spacetimedb::host::module_host::{DatabaseTableUpdate, DatabaseUpdate};
+use spacetimedb::host::module_host::DatabaseTableUpdate;
 use spacetimedb::subscription::query::compile_read_only_query;
 use spacetimedb::subscription::subscription::ExecutionSet;
 use spacetimedb::util::slow::SlowQueryConfig;
@@ -97,10 +97,7 @@ fn eval(c: &mut Criterion) {
 
     let ins_lhs = insert_op(lhs, "footprint", new_lhs_row);
     let ins_rhs = insert_op(rhs, "location", new_rhs_row);
-
-    let update = DatabaseUpdate {
-        tables: vec![ins_lhs, ins_rhs],
-    };
+    let update = [&ins_lhs, &ins_rhs];
 
     let bench_eval = |c: &mut Criterion, name, sql| {
         c.bench_function(name, |b| {

crates/core/Cargo.toml

@@ -27,6 +27,7 @@ spacetimedb-table.workspace = true
 spacetimedb-vm.workspace = true
 
 anyhow = { workspace = true, features = ["backtrace"] }
+arrayvec.workspace = true
 async-trait.workspace = true
 backtrace.workspace = true
 base64.workspace = true

crates/core/src/subscription/module_subscription_manager.rs

@@ -5,9 +5,9 @@ use crate::db::relational_db::RelationalDB;
 use crate::execution_context::ExecutionContext;
 use crate::host::module_host::{DatabaseTableUpdate, ModuleEvent, ProtocolDatabaseUpdate};
 use crate::json::client_api::{TableRowOperationJson, TableUpdateJson};
+use arrayvec::ArrayVec;
 use itertools::Either;
-use rayon::iter::{IntoParallelIterator, ParallelIterator};
-use smallvec::SmallVec;
+use rayon::iter::{IntoParallelRefIterator, ParallelIterator};
 use spacetimedb_client_api_messages::client_api::{TableRowOperation, TableUpdate};
 use spacetimedb_data_structures::map::{Entry, HashMap, HashSet, IntMap};
 use spacetimedb_lib::{Address, Identity};
@@ -128,11 +128,11 @@ impl SubscriptionManager {
             // Collect the delta tables for each query.
             // For selects this is just a single table.
             // For joins it's two tables.
-            let mut units: HashMap<_, SmallVec<[_; 2]>> = HashMap::new();
+            let mut units: HashMap<_, ArrayVec<_, 2>> = HashMap::new();
             for table @ DatabaseTableUpdate { table_id, .. } in tables {
                 if let Some(hashes) = self.tables.get(table_id) {
                     for hash in hashes {
-                        units.entry(hash).or_insert_with(SmallVec::new).push(table);
+                        units.entry(hash).or_insert_with(ArrayVec::new).push(table);
                     }
                 }
             }
@@ -141,10 +141,10 @@ impl SubscriptionManager {
             let ctx = ExecutionContext::incremental_update(db.address(), slow);
             let tx = &tx.deref().into();
             let eval = units
-                .into_par_iter()
-                .filter_map(|(hash, tables)| self.queries.get(hash).map(|unit| (hash, tables, unit)))
-                .filter_map(|(hash, tables, unit)| {
-                    match unit.eval_incr(&ctx, db, tx, &unit.sql, tables.into_iter()) {
+                .par_iter()
+                .filter_map(|(&hash, tables)| {
+                    let unit = self.queries.get(hash)?;
+                    match unit.eval_incr(&ctx, db, tx, &unit.sql, tables.iter().copied()) {
                         Ok(None) => None,
                         Ok(Some(table)) => Some((hash, table)),
                         Err(err) => {

crates/core/src/subscription/query.rs

@@ -280,7 +280,8 @@ mod tests {
     ) -> ResultTest<()> {
         let ctx = &ExecutionContext::incremental_update(db.address(), SlowQueryConfig::default());
         let tx = &tx.into();
-        let result = s.eval_incr(ctx, db, tx, update)?;
+        let update = update.tables.iter().collect::<Vec<_>>();
+        let result = s.eval_incr(ctx, db, tx, &update)?;
         assert_eq!(
             result.tables.len(),
             total_tables,
@@ -370,6 +371,7 @@ mod tests {
 
         let ctx = &ExecutionContext::incremental_update(db.address(), SlowQueryConfig::default());
         let tx = (&tx).into();
+        let update = update.tables.iter().collect::<Vec<_>>();
         let result = query.eval_incr(ctx, &db, &tx, &update)?;
 
         assert_eq!(result.tables.len(), 1);
@@ -726,6 +728,7 @@ mod tests {
         let update = DatabaseUpdate { tables };
         db.with_read_only(ctx, |tx| {
             let tx = (&*tx).into();
+            let update = update.tables.iter().collect::<Vec<_>>();
             let result = query.eval_incr(ctx, db, &tx, &update)?;
             let tables = result
                 .tables

crates/core/src/subscription/subscription.rs

@@ -29,9 +29,7 @@ use crate::client::Protocol;
 use crate::db::relational_db::{RelationalDB, Tx};
 use crate::error::{DBError, SubscriptionError};
 use crate::execution_context::ExecutionContext;
-use crate::host::module_host::{
-    DatabaseTableUpdate, DatabaseUpdate, DatabaseUpdateRelValue, ProtocolDatabaseUpdate, UpdatesRelValue,
-};
+use crate::host::module_host::{DatabaseTableUpdate, DatabaseUpdateRelValue, ProtocolDatabaseUpdate, UpdatesRelValue};
 use crate::json::client_api::TableUpdateJson;
 use crate::vm::{build_query, TxMode};
 use anyhow::Context;
@@ -568,11 +566,11 @@ impl ExecutionSet {
         ctx: &'a ExecutionContext,
         db: &'a RelationalDB,
         tx: &'a TxMode<'a>,
-        database_update: &'a DatabaseUpdate,
-    ) -> Result<DatabaseUpdateRelValue<'_>, DBError> {
+        database_update: &'a [&'a DatabaseTableUpdate],
+    ) -> Result<DatabaseUpdateRelValue<'a>, DBError> {
         let mut tables = Vec::new();
         for unit in &self.exec_units {
-            if let Some(table) = unit.eval_incr(ctx, db, tx, &unit.sql, database_update.tables.iter())? {
+            if let Some(table) = unit.eval_incr(ctx, db, tx, &unit.sql, database_update.iter().copied())? {
                 tables.push(table);
             }
         }