simplify

Author: joshua-spacetime

crates/core/src/host/module_host.rs

@@ -27,7 +27,7 @@ pub use crate::subscription::module_subscription_manager::TransactionOffset;
 use crate::subscription::row_list_builder_pool::{BsatnRowListBuilderPool, JsonRowListBuilderFakePool};
 use crate::subscription::tx::DeltaTx;
 use crate::subscription::websocket_building::{BuildableWebsocketFormat, RowListBuilderSource};
-use crate::subscription::{execute_plan_for_view_with_params, execute_plan_with_params};
+use crate::subscription::{execute_plan, execute_plan_for_view};
 use crate::util::jobs::{AllocatedJobCore, SingleThreadedExecutor};
 use crate::worker_metrics::WORKER_METRICS;
 use anyhow::Context;
@@ -3309,7 +3309,7 @@ impl ModuleHost {
         let delta_tx = DeltaTx::from(tx);
         let params = ExecutionParams::from_auth(auth);
         let (rows, _, metrics) = if returns_view_table && num_private_cols > 0 {
-            execute_plan_for_view_with_params::<F>(
+            execute_plan_for_view::<F>(
                 optimized.iter(),
                 num_cols,
                 num_private_cols,
@@ -3318,7 +3318,7 @@ impl ModuleHost {
                 rlb_pool,
             )
         } else {
-            execute_plan_with_params::<F>(optimized.iter(), &delta_tx, &params, rlb_pool)
+            execute_plan::<F>(optimized.iter(), &delta_tx, &params, rlb_pool)
         }
         .context("One-off queries are not allowed to modify the database")?;
 

crates/core/src/host/wasm_common/module_host_actor.rs

@@ -39,6 +39,7 @@ use spacetimedb_datastore::error::{DatastoreError, ViewError};
 use spacetimedb_datastore::execution_context::{self, ReducerContext, Workload};
 use spacetimedb_datastore::locking_tx_datastore::{FuncCallType, MutTxId, ViewCallInfo};
 use spacetimedb_datastore::traits::{IsolationLevel, Program};
+use spacetimedb_execution::ExecutionParams;
 use spacetimedb_lib::buffer::DecodeError;
 use spacetimedb_lib::db::raw_def::v9::{Lifecycle, ViewResultHeader};
 use spacetimedb_lib::de::DeserializeSeed;
@@ -213,10 +214,11 @@ pub(crate) fn run_query_for_view(
             tx.record_table_scan(&op, table_id);
         }
 
-        plan.base_plan().execute(&*tx, &mut metrics, &mut |row| {
-            rows.push(row.to_product_value());
-            Ok(())
-        })?;
+        plan.base_plan()
+            .execute(&*tx, &ExecutionParams::from_auth(&auth), &mut metrics, &mut |row| {
+                rows.push(row.to_product_value());
+                Ok(())
+            })?;
     }
 
     Ok(rows)

crates/core/src/subscription/mod.rs

@@ -10,7 +10,7 @@ use spacetimedb_client_api_messages::websocket::v1::{self as ws_v1};
 use spacetimedb_datastore::{
     db_metrics::DB_METRICS, execution_context::WorkloadType, locking_tx_datastore::datastore::MetricsRecorder,
 };
-use spacetimedb_execution::{pipelined::PipelinedProject, Datastore, DeltaStore, ExecutionParams, Row};
+use spacetimedb_execution::{pipelined::PipelinedProject, Datastore, DeltaStore, Row};
 use spacetimedb_lib::{metrics::ExecutionMetrics, Identity};
 use spacetimedb_physical_plan::plan::ParamResolver;
 use spacetimedb_primitives::{ColList, TableId};
@@ -99,27 +99,6 @@ impl MetricsRecorder for ExecutionCounters {
 
 /// Execute subscription query fragments over a view.
 pub fn execute_plan_for_view<'p, F>(
-    plan_fragments: impl IntoIterator<Item = &'p PipelinedProject>,
-    num_cols: usize,
-    num_private_cols: usize,
-    tx: &(impl Datastore + DeltaStore),
-    rlb_pool: &impl RowListBuilderSource<F>,
-) -> Result<(F::List, u64, ExecutionMetrics)>
-where
-    F: BuildableWebsocketFormat,
-{
-    execute_plan_for_view_with_params(
-        plan_fragments,
-        num_cols,
-        num_private_cols,
-        tx,
-        &ExecutionParams::empty(),
-        rlb_pool,
-    )
-}
-
-/// Execute subscription query fragments over a view with runtime parameters.
-pub fn execute_plan_for_view_with_params<'p, F>(
     plan_fragments: impl IntoIterator<Item = &'p PipelinedProject>,
     num_cols: usize,
     num_private_cols: usize,
@@ -133,7 +112,7 @@ where
     build_list_with_executor(rlb_pool, |metrics, add| {
         let col_list = ColList::from_iter(num_private_cols..num_cols);
         for fragment in plan_fragments {
-            fragment.execute_with_params(tx, params, metrics, &mut |row| match row {
+            fragment.execute(tx, params, metrics, &mut |row| match row {
                 Row::Ptr(ptr) => add(ptr.project_product(&col_list)?),
                 Row::Ref(val) => add(val.project_product(&col_list)?),
             })?;
@@ -144,18 +123,6 @@ where
 
 /// Execute a subscription query
 pub fn execute_plan<'p, F>(
-    plan_fragments: impl IntoIterator<Item = &'p PipelinedProject>,
-    tx: &(impl Datastore + DeltaStore),
-    rlb_pool: &impl RowListBuilderSource<F>,
-) -> Result<(F::List, u64, ExecutionMetrics)>
-where
-    F: BuildableWebsocketFormat,
-{
-    execute_plan_with_params(plan_fragments, tx, &ExecutionParams::empty(), rlb_pool)
-}
-
-/// Execute a subscription query with runtime parameters.
-pub fn execute_plan_with_params<'p, F>(
     plan_fragments: impl IntoIterator<Item = &'p PipelinedProject>,
     tx: &(impl Datastore + DeltaStore),
     params: &impl ParamResolver,
@@ -166,7 +133,7 @@ where
 {
     build_list_with_executor(rlb_pool, |metrics, add| {
         for fragment in plan_fragments {
-            fragment.execute_with_params(tx, params, metrics, add)?;
+            fragment.execute(tx, params, metrics, add)?;
         }
         Ok(())
     })
@@ -252,7 +219,7 @@ where
     Tx: Datastore + DeltaStore,
     F: BuildableWebsocketFormat,
 {
-    execute_plan_for_view_with_params::<F>(plan_fragments, num_cols, num_private_cols, tx, params, rlb_pool).map(
+    execute_plan_for_view::<F>(plan_fragments, num_cols, num_private_cols, tx, params, rlb_pool).map(
         |(rows, num_rows, metrics)| table_update_from_rows(rows, num_rows, metrics, table_id, table_name, update_type),
     )
 }
@@ -270,7 +237,7 @@ pub fn collect_table_update<'p, F>(
 where
     F: BuildableWebsocketFormat,
 {
-    execute_plan_with_params::<F>(plan_fragments, tx, params, rlb_pool).map(|(rows, num_rows, metrics)| {
+    execute_plan::<F>(plan_fragments, tx, params, rlb_pool).map(|(rows, num_rows, metrics)| {
         table_update_from_rows(rows, num_rows, metrics, table_id, table_name, update_type)
     })
 }

crates/execution/src/dml.rs

@@ -5,7 +5,7 @@ use spacetimedb_physical_plan::plan::ParamResolver;
 use spacetimedb_primitives::{ColId, TableId};
 use spacetimedb_sats::size_of::SizeOf;
 
-use crate::{pipelined::PipelinedProject, Datastore, DeltaStore, ExecutionParams};
+use crate::{pipelined::PipelinedProject, Datastore, DeltaStore};
 
 /// A mutable datastore can read as well as insert and delete rows
 pub trait MutDatastore: Datastore + DeltaStore {
@@ -31,20 +31,16 @@ impl From<MutationPlan> for MutExecutor {
 }
 
 impl MutExecutor {
-    pub fn execute<Tx: MutDatastore>(&self, tx: &mut Tx, metrics: &mut ExecutionMetrics) -> Result<()> {
-        self.execute_with_params(tx, &ExecutionParams::empty(), metrics)
-    }
-
-    pub fn execute_with_params<Tx: MutDatastore>(
+    pub fn execute<Tx: MutDatastore>(
         &self,
         tx: &mut Tx,
         params: &impl ParamResolver,
         metrics: &mut ExecutionMetrics,
     ) -> Result<()> {
         match self {
             Self::Insert(exec) => exec.execute(tx, metrics),
-            Self::Delete(exec) => exec.execute_with_params(tx, params, metrics),
-            Self::Update(exec) => exec.execute_with_params(tx, params, metrics),
+            Self::Delete(exec) => exec.execute(tx, params, metrics),
+            Self::Update(exec) => exec.execute(tx, params, metrics),
         }
     }
 }
@@ -94,15 +90,15 @@ impl From<DeletePlan> for DeleteExecutor {
 }
 
 impl DeleteExecutor {
-    fn execute_with_params<Tx: MutDatastore>(
+    fn execute<Tx: MutDatastore>(
         &self,
         tx: &mut Tx,
         params: &impl ParamResolver,
         metrics: &mut ExecutionMetrics,
     ) -> Result<()> {
         // TODO: Delete by row id instead of product value
         let mut deletes = vec![];
-        self.filter.execute_with_params(tx, params, metrics, &mut |row| {
+        self.filter.execute(tx, params, metrics, &mut |row| {
             deletes.push(row.to_product_value());
             Ok(())
         })?;
@@ -137,14 +133,14 @@ impl From<UpdatePlan> for UpdateExecutor {
 }
 
 impl UpdateExecutor {
-    fn execute_with_params<Tx: MutDatastore>(
+    fn execute<Tx: MutDatastore>(
         &self,
         tx: &mut Tx,
         params: &impl ParamResolver,
         metrics: &mut ExecutionMetrics,
     ) -> Result<()> {
         let mut deletes = vec![];
-        self.filter.execute_with_params(tx, params, metrics, &mut |row| {
+        self.filter.execute(tx, params, metrics, &mut |row| {
             deletes.push(row.to_product_value());
             Ok(())
         })?;

crates/execution/src/lib.rs

@@ -14,16 +14,12 @@ use spacetimedb_table::{static_assert_size, table::RowRef};
 pub mod dml;
 pub mod pipelined;
 
-#[derive(Debug, Clone, Copy, Default)]
+#[derive(Debug, Clone, Copy)]
 pub struct ExecutionParams {
     sender: Option<Identity>,
 }
 
 impl ExecutionParams {
-    pub fn empty() -> Self {
-        Self { sender: None }
-    }
-
     pub fn from_sender(sender: Identity) -> Self {
         Self { sender: Some(sender) }
     }