wip

Author: joshua-spacetime

crates/bench/benches/subscription.rs

@@ -11,7 +11,7 @@ use spacetimedb_bench::database::BenchDatabase as _;
 use spacetimedb_bench::spacetime_raw::SpacetimeRaw;
 use spacetimedb_client_api_messages::websocket::v1::BsatnFormat;
 use spacetimedb_datastore::execution_context::Workload;
-use spacetimedb_execution::pipelined::PipelinedProject;
+use spacetimedb_execution::{pipelined::PipelinedProject, ExecutionParams};
 use spacetimedb_primitives::{col_list, TableId};
 use spacetimedb_query::compile_subscription;
 use spacetimedb_sats::{bsatn, product, AlgebraicType, AlgebraicValue};
@@ -113,6 +113,7 @@ fn eval(c: &mut Criterion) {
             let auth = AuthCtx::for_testing();
             let schema_viewer = &SchemaViewer::new(&tx, &auth);
             let (plans, table_id, table_name, _) = compile_subscription(sql, schema_viewer, &auth).unwrap();
+            let params = ExecutionParams::from_auth(&auth);
             let plans = plans
                 .into_iter()
                 .map(|plan| plan.optimize(&auth).unwrap())
@@ -126,6 +127,7 @@ fn eval(c: &mut Criterion) {
                     table_id,
                     table_name.clone(),
                     &tx,
+                    &params,
                     TableUpdateType::Subscribe,
                     &bsatn_rlb_pool,
                 ));

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, execute_plan_for_view};
+use crate::subscription::{execute_plan_for_view_with_params, execute_plan_with_params};
 use crate::util::jobs::{AllocatedJobCore, SingleThreadedExecutor};
 use crate::worker_metrics::WORKER_METRICS;
 use anyhow::Context;
@@ -52,6 +52,7 @@ use spacetimedb_datastore::locking_tx_datastore::{MutTxId, ViewCallInfo};
 use spacetimedb_datastore::traits::{IsolationLevel, Program, TxData};
 pub use spacetimedb_durability::{DurabilityExited, DurableOffset};
 use spacetimedb_execution::pipelined::PipelinedProject;
+use spacetimedb_execution::ExecutionParams;
 use spacetimedb_execution::RelValue;
 use spacetimedb_expr::expr::CollectViews;
 use spacetimedb_lib::db::raw_def::v9::Lifecycle;
@@ -3306,10 +3307,18 @@ impl ModuleHost {
 
         let table_name = table_name.into();
         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::<F>(optimized.iter(), num_cols, num_private_cols, &delta_tx, rlb_pool)
+            execute_plan_for_view_with_params::<F>(
+                optimized.iter(),
+                num_cols,
+                num_private_cols,
+                &delta_tx,
+                &params,
+                rlb_pool,
+            )
         } else {
-            execute_plan::<F>(optimized.iter(), &delta_tx, rlb_pool)
+            execute_plan_with_params::<F>(optimized.iter(), &delta_tx, &params, rlb_pool)
         }
         .context("One-off queries are not allowed to modify the database")?;
 

crates/core/src/subscription/mod.rs

@@ -10,8 +10,9 @@ 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, Row};
+use spacetimedb_execution::{pipelined::PipelinedProject, Datastore, DeltaStore, ExecutionParams, Row};
 use spacetimedb_lib::{metrics::ExecutionMetrics, Identity};
+use spacetimedb_physical_plan::plan::ParamResolver;
 use spacetimedb_primitives::{ColList, TableId};
 use spacetimedb_sats::bsatn::ToBsatn;
 use spacetimedb_sats::Serialize;
@@ -104,13 +105,35 @@ pub fn execute_plan_for_view<'p, F>(
     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,
+    tx: &(impl Datastore + DeltaStore),
+    params: &impl ParamResolver,
+    rlb_pool: &impl RowListBuilderSource<F>,
+) -> Result<(F::List, u64, ExecutionMetrics)>
 where
     F: BuildableWebsocketFormat,
 {
     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(tx, metrics, &mut |row| match row {
+            fragment.execute_with_params(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)?),
             })?;
@@ -125,12 +148,25 @@ pub fn execute_plan<'p, F>(
     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,
+    rlb_pool: &impl RowListBuilderSource<F>,
+) -> Result<(F::List, u64, ExecutionMetrics)>
 where
     F: BuildableWebsocketFormat,
 {
     build_list_with_executor(rlb_pool, |metrics, add| {
         for fragment in plan_fragments {
-            fragment.execute(tx, metrics, add)?;
+            fragment.execute_with_params(tx, params, metrics, add)?;
         }
         Ok(())
     })
@@ -208,14 +244,15 @@ pub fn collect_table_update_for_view<'p, Tx, F>(
     table_id: TableId,
     table_name: TableName,
     tx: &Tx,
+    params: &impl ParamResolver,
     update_type: TableUpdateType,
     rlb_pool: &impl RowListBuilderSource<F>,
 ) -> Result<(ws_v1::TableUpdate<F>, ExecutionMetrics)>
 where
     Tx: Datastore + DeltaStore,
     F: BuildableWebsocketFormat,
 {
-    execute_plan_for_view::<F>(plan_fragments, num_cols, num_private_cols, tx, rlb_pool).map(
+    execute_plan_for_view_with_params::<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),
     )
 }
@@ -226,13 +263,14 @@ pub fn collect_table_update<'p, F>(
     table_id: TableId,
     table_name: TableName,
     tx: &(impl Datastore + DeltaStore),
+    params: &impl ParamResolver,
     update_type: TableUpdateType,
     rlb_pool: &impl RowListBuilderSource<F>,
 ) -> Result<(ws_v1::TableUpdate<F>, ExecutionMetrics)>
 where
     F: BuildableWebsocketFormat,
 {
-    execute_plan::<F>(plan_fragments, tx, rlb_pool).map(|(rows, num_rows, metrics)| {
+    execute_plan_with_params::<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)
     })
 }
@@ -265,6 +303,7 @@ pub fn execute_plans<F: BuildableWebsocketFormat>(
                         table_id,
                         table_name.clone(),
                         tx,
+                        plan.params(),
                         update_type,
                         rlb_pool,
                     )?
@@ -274,6 +313,7 @@ pub fn execute_plans<F: BuildableWebsocketFormat>(
                         table_id,
                         table_name.clone(),
                         tx,
+                        plan.params(),
                         update_type,
                         rlb_pool,
                     )?

crates/core/src/subscription/module_subscription_actor.rs

@@ -35,6 +35,7 @@ use spacetimedb_datastore::locking_tx_datastore::datastore::TxMetrics;
 use spacetimedb_datastore::locking_tx_datastore::{MutTxId, TxId};
 use spacetimedb_datastore::traits::{IsolationLevel, TxData};
 use spacetimedb_durability::TxOffset;
+use spacetimedb_execution::ExecutionParams;
 use spacetimedb_expr::expr::CollectViews;
 use spacetimedb_lib::identity::RequestId;
 use spacetimedb_lib::metrics::ExecutionMetrics;
@@ -467,6 +468,7 @@ impl ModuleSubscriptions {
             .unwrap_or_default();
 
         let tx = DeltaTx::from(tx);
+        let params = ExecutionParams::from_sender(sender.id.identity);
 
         // TODO: See the comment on `collect_table_update_for_view`.
         // The following view and non-view branches should be merged together,
@@ -479,6 +481,7 @@ impl ModuleSubscriptions {
                 table_id,
                 table_name.clone(),
                 &tx,
+                &params,
                 update_type,
                 &self.bsatn_rlb_pool,
             )
@@ -488,6 +491,7 @@ impl ModuleSubscriptions {
                 table_id,
                 table_name.clone(),
                 &tx,
+                &params,
                 update_type,
                 &self.bsatn_rlb_pool,
             )
@@ -499,6 +503,7 @@ impl ModuleSubscriptions {
                 table_id,
                 table_name,
                 &tx,
+                &params,
                 update_type,
                 &JsonRowListBuilderFakePool,
             )
@@ -508,6 +513,7 @@ impl ModuleSubscriptions {
                 table_id,
                 table_name,
                 &tx,
+                &params,
                 update_type,
                 &JsonRowListBuilderFakePool,
             )

crates/execution/src/dml.rs

@@ -1,10 +1,11 @@
 use anyhow::Result;
 use spacetimedb_lib::{metrics::ExecutionMetrics, AlgebraicValue, ProductValue};
 use spacetimedb_physical_plan::dml::{DeletePlan, InsertPlan, MutationPlan, UpdatePlan};
+use spacetimedb_physical_plan::plan::ParamResolver;
 use spacetimedb_primitives::{ColId, TableId};
 use spacetimedb_sats::size_of::SizeOf;
 
-use crate::{pipelined::PipelinedProject, Datastore, DeltaStore};
+use crate::{pipelined::PipelinedProject, Datastore, DeltaStore, ExecutionParams};
 
 /// A mutable datastore can read as well as insert and delete rows
 pub trait MutDatastore: Datastore + DeltaStore {
@@ -31,10 +32,19 @@ 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>(
+        &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(tx, metrics),
-            Self::Update(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),
         }
     }
 }
@@ -84,10 +94,15 @@ impl From<DeletePlan> for DeleteExecutor {
 }
 
 impl DeleteExecutor {
-    fn execute<Tx: MutDatastore>(&self, tx: &mut Tx, metrics: &mut ExecutionMetrics) -> Result<()> {
+    fn execute_with_params<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(tx, metrics, &mut |row| {
+        self.filter.execute_with_params(tx, params, metrics, &mut |row| {
             deletes.push(row.to_product_value());
             Ok(())
         })?;
@@ -122,9 +137,14 @@ impl From<UpdatePlan> for UpdateExecutor {
 }
 
 impl UpdateExecutor {
-    fn execute<Tx: MutDatastore>(&self, tx: &mut Tx, metrics: &mut ExecutionMetrics) -> Result<()> {
+    fn execute_with_params<Tx: MutDatastore>(
+        &self,
+        tx: &mut Tx,
+        params: &impl ParamResolver,
+        metrics: &mut ExecutionMetrics,
+    ) -> Result<()> {
         let mut deletes = vec![];
-        self.filter.execute(tx, metrics, &mut |row| {
+        self.filter.execute_with_params(tx, params, metrics, &mut |row| {
             deletes.push(row.to_product_value());
             Ok(())
         })?;

crates/execution/src/lib.rs

@@ -1,18 +1,53 @@
 use anyhow::Result;
 use core::hash::{Hash, Hasher};
 use core::ops::RangeBounds;
-use spacetimedb_lib::query::Delta;
-use spacetimedb_physical_plan::plan::{ProjectField, TupleField};
+use spacetimedb_lib::{identity::AuthCtx, query::Delta, AlgebraicType, Identity};
+use spacetimedb_physical_plan::plan::{ParamResolver, ProjectField, TupleField};
 use spacetimedb_primitives::{ColList, IndexId, TableId};
 use spacetimedb_sats::bsatn::{BufReservedFill, EncodeError, ToBsatn};
 use spacetimedb_sats::buffer::BufWriter;
 use spacetimedb_sats::product_value::InvalidFieldError;
 use spacetimedb_sats::{impl_serialize, AlgebraicValue, ProductValue};
+use spacetimedb_sql_parser::ast::Parameter;
 use spacetimedb_table::{static_assert_size, table::RowRef};
 
 pub mod dml;
 pub mod pipelined;
 
+#[derive(Debug, Clone, Copy, Default)]
+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) }
+    }
+
+    pub fn from_auth(auth: &AuthCtx) -> Self {
+        Self::from_sender(auth.caller())
+    }
+}
+
+impl ParamResolver for ExecutionParams {
+    fn resolve_param(&self, param: Parameter, ty: &AlgebraicType) -> AlgebraicValue {
+        match param {
+            Parameter::Sender if ty.is_identity() => self.sender.expect("missing sender binding for :sender").into(),
+            Parameter::Sender if ty.is_bytes() => AlgebraicValue::Bytes(
+                self.sender
+                    .expect("missing sender binding for :sender")
+                    .to_be_byte_array()
+                    .into(),
+            ),
+            Parameter::Sender => panic!("unsupported type for :sender: {ty:?}"),
+        }
+    }
+}
+
 pub trait Datastore {
     /// Iterator type for table scans
     type TableIter<'a>: Iterator<Item = RowRef<'a>> + 'a