From 15540b0a3171fe4618b3b757d703796a7837693b Mon Sep 17 00:00:00 2001
From: joshua-spacetime <josh@clockworklabs.io>
Date: Tue, 9 Jun 2026 12:59:52 -0700
Subject: [PATCH 1/2] Parameterized query plans

---
 crates/core/src/host/module_host.rs           |   7 +-
 crates/core/src/subscription/delta.rs         |  10 +-
 crates/core/src/subscription/metrics.rs       |   2 +-
 crates/core/src/subscription/mod.rs           |  21 +-
 .../subscription/module_subscription_actor.rs |  15 +-
 .../module_subscription_manager.rs            | 103 +++--
 crates/core/src/subscription/query.rs         |  17 +-
 crates/core/src/subscription/subscription.rs  |   6 +-
 crates/execution/src/pipelined.rs             |  55 ++-
 crates/expr/src/check.rs                      |   3 +-
 crates/expr/src/expr.rs                       |  72 +++-
 crates/expr/src/lib.rs                        |  20 +-
 crates/expr/src/rls.rs                        |   8 +-
 crates/expr/src/statement.rs                  |   4 +-
 crates/physical-plan/src/compile.rs           |   1 +
 crates/physical-plan/src/dml.rs               |  23 +-
 crates/physical-plan/src/plan.rs              | 377 ++++++++++++++++--
 crates/physical-plan/src/rules.rs             |  36 +-
 crates/query/src/lib.rs                       |   8 +-
 crates/subscription/src/lib.rs                | 104 +++--
 20 files changed, 712 insertions(+), 180 deletions(-)

diff --git a/crates/core/src/host/module_host.rs b/crates/core/src/host/module_host.rs
index 9ce704426f7..3be9b535ad3 100644
--- a/crates/core/src/host/module_host.rs
+++ b/crates/core/src/host/module_host.rs
@@ -53,7 +53,7 @@ use spacetimedb_datastore::traits::{IsolationLevel, Program, TxData};
 pub use spacetimedb_durability::{DurabilityExited, DurableOffset};
 use spacetimedb_execution::pipelined::{PipelinedProject, ViewProject};
 use spacetimedb_execution::RelValue;
-use spacetimedb_expr::expr::CollectViews;
+use spacetimedb_expr::expr::{BindEnv, CollectViews};
 use spacetimedb_lib::db::raw_def::v9::Lifecycle;
 use spacetimedb_lib::http::{Request as HttpRequest, Response as HttpResponse};
 use spacetimedb_lib::identity::{AuthCtx, RequestId};
@@ -3272,13 +3272,14 @@ impl ModuleHost {
             plans,
             _,
             table_name,
-            _,
+            requires_sender_binding,
         ) = compile_subscription(query, &schema_tx, auth)?;
+        let bind_env = BindEnv::for_sender_binding(requires_sender_binding, auth.caller());
 
         // Optimize each fragment.
         let optimized = plans
             .into_iter()
-            .map(|plan| plan.optimize(auth))
+            .map(|plan| plan.optimize(auth).map(|plan| plan.bind_params(&bind_env)))
             .collect::<Result<Vec<_>, _>>()?;
 
         check_row_limit(
diff --git a/crates/core/src/subscription/delta.rs b/crates/core/src/subscription/delta.rs
index b2747a933e2..fd7c91088fa 100644
--- a/crates/core/src/subscription/delta.rs
+++ b/crates/core/src/subscription/delta.rs
@@ -2,6 +2,7 @@ use crate::host::module_host::UpdatesRelValue;
 use anyhow::Result;
 use spacetimedb_data_structures::map::{HashCollectionExt as _, HashMap};
 use spacetimedb_execution::{Datastore, DeltaStore, RelValue, Row};
+use spacetimedb_expr::expr::BindEnv;
 use spacetimedb_lib::metrics::ExecutionMetrics;
 use spacetimedb_primitives::ColList;
 use spacetimedb_sats::product_value::InvalidFieldError;
@@ -20,6 +21,7 @@ pub fn eval_delta<'a, Tx: Datastore + DeltaStore>(
     tx: &'a Tx,
     metrics: &mut ExecutionMetrics,
     plan: &SubscriptionPlan,
+    bind_env: &BindEnv,
 ) -> Result<Option<UpdatesRelValue<'a>>> {
     metrics.delta_queries_evaluated += 1;
 
@@ -42,12 +44,12 @@ pub fn eval_delta<'a, Tx: Datastore + DeltaStore>(
     if !plan.is_join() {
         // Single table plans will never return redundant rows,
         // so there's no need to track row counts.
-        plan.for_each_insert(tx, metrics, &mut |row| {
+        plan.for_each_insert(bind_env, tx, metrics, &mut |row| {
             inserts.push(maybe_project(row)?);
             Ok(())
         })?;
 
-        plan.for_each_delete(tx, metrics, &mut |row| {
+        plan.for_each_delete(bind_env, tx, metrics, &mut |row| {
             deletes.push(maybe_project(row)?);
             Ok(())
         })?;
@@ -57,7 +59,7 @@ pub fn eval_delta<'a, Tx: Datastore + DeltaStore>(
         let mut insert_counts = HashMap::new();
         let mut delete_counts = HashMap::new();
 
-        plan.for_each_insert(tx, metrics, &mut |row| {
+        plan.for_each_insert(bind_env, tx, metrics, &mut |row| {
             let row = maybe_project(row)?;
             let n = insert_counts.entry(row).or_default();
             if *n > 0 {
@@ -67,7 +69,7 @@ pub fn eval_delta<'a, Tx: Datastore + DeltaStore>(
             Ok(())
         })?;
 
-        plan.for_each_delete(tx, metrics, &mut |row| {
+        plan.for_each_delete(bind_env, tx, metrics, &mut |row| {
             let row = maybe_project(row)?;
             match insert_counts.get_mut(&row) {
                 // We have not seen an insert for this row.
diff --git a/crates/core/src/subscription/metrics.rs b/crates/core/src/subscription/metrics.rs
index b367a57056b..1247236ad5a 100644
--- a/crates/core/src/subscription/metrics.rs
+++ b/crates/core/src/subscription/metrics.rs
@@ -64,7 +64,7 @@ fn extract_columns(
                 extract_columns(expr, schema, columns);
             }
         }
-        PhysicalExpr::Value(_) => {}
+        PhysicalExpr::Value(_) | PhysicalExpr::Param(..) => {}
     }
 }
 
diff --git a/crates/core/src/subscription/mod.rs b/crates/core/src/subscription/mod.rs
index 40d6d712504..1614e119371 100644
--- a/crates/core/src/subscription/mod.rs
+++ b/crates/core/src/subscription/mod.rs
@@ -256,14 +256,27 @@ pub fn execute_plans<F: BuildableWebsocketFormat>(
 ) -> Result<(ws_v1::DatabaseUpdate<F>, ExecutionMetrics, Vec<QueryMetrics>), DBError> {
     plans
         .par_iter()
-        .flat_map_iter(|plan| plan.plans_fragments().map(|fragment| (plan.sql(), fragment)))
-        .filter(|(_, plan)| {
+        .flat_map_iter(|plan| {
+            plan.plans_fragments()
+                .map(|fragment| (plan.sql(), plan.bind_env(), fragment))
+        })
+        .filter(|(_, _, plan)| {
             // Since subscriptions only support selects and inner joins,
             // we filter out any plans that read from an empty table.
             plan.table_ids().all(|table_id| tx.row_count(table_id) > 0)
         })
-        .map(|(sql, plan)| (sql, plan, plan.subscribed_table_id(), plan.subscribed_table_name()))
-        .map(|(sql, plan, table_id, table_name)| (sql, plan.optimized_physical_plan().clone(), table_id, table_name))
+        .map(|(sql, bind_env, plan)| {
+            (
+                sql,
+                bind_env,
+                plan,
+                plan.subscribed_table_id(),
+                plan.subscribed_table_name(),
+            )
+        })
+        .map(|(sql, bind_env, plan, table_id, table_name)| {
+            (sql, plan.bound_optimized_physical_plan(bind_env), table_id, table_name)
+        })
         .map(|(sql, plan, table_id, table_name)| (sql, plan.optimize(auth), table_id, table_name))
         .map(|(sql, plan, table_id, table_name)| {
             plan.and_then(|plan| {
diff --git a/crates/core/src/subscription/module_subscription_actor.rs b/crates/core/src/subscription/module_subscription_actor.rs
index aede13262df..aa4392c088e 100644
--- a/crates/core/src/subscription/module_subscription_actor.rs
+++ b/crates/core/src/subscription/module_subscription_actor.rs
@@ -427,7 +427,9 @@ impl ModuleSubscriptions {
             tx,
             |plan, tx| {
                 plan.plans_fragments()
-                    .map(|plan_fragment| estimate_rows_scanned(tx, plan_fragment.optimized_physical_plan()))
+                    .map(|plan_fragment| {
+                        estimate_rows_scanned(tx, &plan_fragment.bound_optimized_physical_plan(plan.bind_env()))
+                    })
                     .fold(0, |acc, rows_scanned| acc.saturating_add(rows_scanned))
             },
             auth,
@@ -438,8 +440,7 @@ impl ModuleSubscriptions {
 
         let plans = query
             .plans_fragments()
-            .map(|fragment| fragment.optimized_physical_plan())
-            .cloned()
+            .map(|fragment| fragment.bound_optimized_physical_plan(query.bind_env()))
             .map(|plan| plan.optimize(auth))
             .collect::<Result<Vec<_>, _>>()?;
 
@@ -533,7 +534,9 @@ impl ModuleSubscriptions {
             tx,
             |plan, tx| {
                 plan.plans_fragments()
-                    .map(|plan_fragment| estimate_rows_scanned(tx, plan_fragment.optimized_physical_plan()))
+                    .map(|plan_fragment| {
+                        estimate_rows_scanned(tx, &plan_fragment.bound_optimized_physical_plan(plan.bind_env()))
+                    })
                     .fold(0, |acc, rows_scanned| acc.saturating_add(rows_scanned))
             },
             auth,
@@ -1538,7 +1541,9 @@ impl ModuleSubscriptions {
             &tx,
             |plan, tx| {
                 plan.plans_fragments()
-                    .map(|plan_fragment| estimate_rows_scanned(tx, plan_fragment.optimized_physical_plan()))
+                    .map(|plan_fragment| {
+                        estimate_rows_scanned(tx, &plan_fragment.bound_optimized_physical_plan(plan.bind_env()))
+                    })
                     .fold(0, |acc, rows_scanned| acc.saturating_add(rows_scanned))
             },
             &auth,
diff --git a/crates/core/src/subscription/module_subscription_manager.rs b/crates/core/src/subscription/module_subscription_manager.rs
index 0700846fc74..07907e71862 100644
--- a/crates/core/src/subscription/module_subscription_manager.rs
+++ b/crates/core/src/subscription/module_subscription_manager.rs
@@ -27,7 +27,7 @@ use spacetimedb_data_structures::map::{
 };
 use spacetimedb_datastore::locking_tx_datastore::state_view::StateView;
 use spacetimedb_durability::TxOffset;
-use spacetimedb_expr::expr::CollectViews;
+use spacetimedb_expr::expr::{BindEnv, CollectViews};
 use spacetimedb_lib::metrics::ExecutionMetrics;
 use spacetimedb_lib::{AlgebraicValue, ConnectionId, Identity, ProductValue};
 use spacetimedb_primitives::{ColId, IndexId, TableId, ViewId};
@@ -45,7 +45,7 @@ use tokio::sync::{mpsc, oneshot};
 /// Identity is insufficient because different ConnectionIds can use the same Identity.
 /// TODO: Determine if ConnectionId is sufficient for uniquely identifying a client.
 type ClientId = (Identity, ConnectionId);
-type Query = Arc<Plan>;
+type Query = Arc<PlanInstance>;
 type Client = Arc<ClientConnectionSender>;
 type SwitchedTableUpdate =
     ws_v1::FormatSwitch<ws_v1::TableUpdate<ws_v1::BsatnFormat>, ws_v1::TableUpdate<ws_v1::JsonFormat>>;
@@ -60,14 +60,18 @@ type ClientQueryId = ws_v1::QueryId;
 type SubscriptionId = (ClientId, ClientQueryId);
 type SubscriptionIdV2 = (ClientId, ClientQuerySetId);
 
+/// The unbound, reusable subscription query plan.
+///
+/// Runtime values such as `:sender` are represented as formal parameters inside these plans.
+/// TODO: Intern and share `PlanTemplate`s across bound [`PlanInstance`]s with the same template hash.
+/// Today, [`PlanInstance::new`] still allocates a fresh template for each cached subscription instance.
 #[derive(Debug)]
-pub struct Plan {
-    hash: QueryHash,
+pub struct PlanTemplate {
     sql: String,
     plans: Vec<SubscriptionPlan>,
 }
 
-impl CollectViews for Plan {
+impl CollectViews for PlanTemplate {
     fn collect_views(&self, views: &mut HashSet<ViewId>) {
         for plan in &self.plans {
             plan.collect_views(views);
@@ -75,10 +79,42 @@ impl CollectViews for Plan {
     }
 }
 
-impl Plan {
-    /// Create a new subscription plan to be cached
-    pub fn new(plans: Vec<SubscriptionPlan>, hash: QueryHash, text: String) -> Self {
-        Self { plans, hash, sql: text }
+impl PlanTemplate {
+    pub fn new(plans: Vec<SubscriptionPlan>, text: String) -> Self {
+        Self { sql: text, plans }
+    }
+}
+
+/// A concrete subscription query instance with runtime parameter bindings.
+#[derive(Debug)]
+pub struct PlanInstance {
+    hash: QueryHash,
+    template: Arc<PlanTemplate>,
+    bind_env: BindEnv,
+}
+
+/// A subscription plan tracked by the subscription manager.
+pub type Plan = PlanInstance;
+
+impl CollectViews for PlanInstance {
+    fn collect_views(&self, views: &mut HashSet<ViewId>) {
+        self.template.collect_views(views);
+    }
+}
+
+impl PlanInstance {
+    /// Create a new subscription plan instance to be cached.
+    pub fn new(plans: Vec<SubscriptionPlan>, hash: QueryHash, text: String, bind_env: BindEnv) -> Self {
+        Self::from_template(Arc::new(PlanTemplate::new(plans, text)), hash, bind_env)
+    }
+
+    /// Create a new subscription plan instance from an existing plan template.
+    pub fn from_template(template: Arc<PlanTemplate>, hash: QueryHash, bind_env: BindEnv) -> Self {
+        Self {
+            hash,
+            template,
+            bind_env,
+        }
     }
 
     /// Returns the query hash for this subscription
@@ -89,18 +125,19 @@ impl Plan {
     /// A subscription query return rows from a single table.
     /// This method returns the id of that table.
     pub fn subscribed_table_id(&self) -> TableId {
-        self.plans[0].subscribed_table_id()
+        self.template.plans[0].subscribed_table_id()
     }
 
     /// A subscription query return rows from a single table.
     /// This method returns the name of that table.
     pub fn subscribed_table_name(&self) -> &TableName {
-        self.plans[0].subscribed_table_name()
+        self.template.plans[0].subscribed_table_name()
     }
 
     /// Returns the index ids from which this subscription reads
     pub fn index_ids(&self) -> impl Iterator<Item = (TableId, IndexId)> + use<> {
-        self.plans
+        self.template
+            .plans
             .iter()
             .flat_map(|plan| plan.index_ids())
             .collect::<HashSet<_>>()
@@ -109,7 +146,8 @@ impl Plan {
 
     /// Returns the table ids from which this subscription reads
     pub fn table_ids(&self) -> impl Iterator<Item = TableId> + '_ {
-        self.plans
+        self.template
+            .plans
             .iter()
             .flat_map(|plan| plan.table_ids())
             .collect::<HashSet<_>>()
@@ -120,9 +158,10 @@ impl Plan {
     fn search_args(&self) -> impl Iterator<Item = (TableId, ColId, AlgebraicValue)> + use<> {
         let mut args = HashSet::new();
         for arg in self
+            .template
             .plans
             .iter()
-            .flat_map(|subscription| subscription.optimized_physical_plan().search_args())
+            .flat_map(|subscription| subscription.bound_optimized_physical_plan(&self.bind_env).search_args())
         {
             args.insert(arg);
         }
@@ -132,22 +171,30 @@ impl Plan {
     /// Returns the plan fragments that comprise this subscription.
     /// Will only return one element unless there is a table with multiple RLS rules.
     pub fn plans_fragments(&self) -> impl Iterator<Item = &SubscriptionPlan> + '_ {
-        self.plans.iter()
+        self.template.plans.iter()
     }
 
     /// Returns the join edges for this plan, if any.
     pub fn join_edges(&self) -> impl Iterator<Item = (JoinEdge, AlgebraicValue)> + '_ {
-        self.plans.iter().filter_map(|plan| plan.join_edge())
+        self.template
+            .plans
+            .iter()
+            .filter_map(|plan| plan.bound_join_edge(&self.bind_env))
     }
 
     /// The `SQL` text of this subscription.
     pub fn sql(&self) -> &str {
-        &self.sql
+        &self.template.sql
+    }
+
+    /// Runtime parameter bindings for this subscription instance.
+    pub fn bind_env(&self) -> &BindEnv {
+        &self.bind_env
     }
 
     /// Does this plan return rows from an event table?
     pub fn returns_event_table(&self) -> bool {
-        self.plans.iter().any(|p| p.returns_event_table())
+        self.template.plans.iter().any(|p| p.returns_event_table())
     }
 }
 
@@ -1480,15 +1527,15 @@ impl SubscriptionManager {
             })
             .fold(FoldState::default(), |mut acc, (qstate, plan, _hash)| {
                 let table_name = plan.subscribed_table_name().clone();
-                match eval_delta(tx, &mut acc.metrics, plan) {
+                match eval_delta(tx, &mut acc.metrics, plan, qstate.query.bind_env()) {
                     Err(err) => {
                         tracing::error!(
                             message = "Query errored during tx update",
-                            sql = qstate.query.sql,
+                            sql = qstate.query.sql(),
                             reason = ?err,
                         );
                         let err = DBError::WithSql {
-                            sql: qstate.query.sql.as_str().into(),
+                            sql: qstate.query.sql().into(),
                             error: Box::new(err.into()),
                         }
                         .to_string()
@@ -1637,15 +1684,15 @@ impl SubscriptionManager {
 
                 let clients_for_query = qstate.all_v1_clients();
 
-                match eval_delta(tx, &mut acc.metrics, plan) {
+                match eval_delta(tx, &mut acc.metrics, plan, qstate.query.bind_env()) {
                     Err(err) => {
                         tracing::error!(
                             message = "Query errored during tx update",
-                            sql = qstate.query.sql,
+                            sql = qstate.query.sql(),
                             reason = ?err,
                         );
                         let err = DBError::WithSql {
-                            sql: qstate.query.sql.as_str().into(),
+                            sql: qstate.query.sql().into(),
                             error: Box::new(err.into()),
                         }
                         .to_string()
@@ -2191,6 +2238,7 @@ mod tests {
     use std::{sync::Arc, time::Duration};
 
     use spacetimedb_client_api_messages::websocket::{v1 as ws_v1, v2 as ws_v2};
+    use spacetimedb_expr::expr::BindEnv;
     use spacetimedb_lib::AlgebraicValue;
     use spacetimedb_lib::{error::ResultTest, identity::AuthCtx, AlgebraicType, ConnectionId, Identity, Timestamp};
     use spacetimedb_primitives::{ColId, TableId};
@@ -2231,9 +2279,10 @@ mod tests {
     fn compile_plan_with_auth(db: &RelationalDB, sql: &str, auth: AuthCtx) -> ResultTest<Arc<Plan>> {
         with_read_only(db, |tx| {
             let tx = SchemaViewer::new(&*tx, &auth);
-            let (plans, has_param) = SubscriptionPlan::compile(sql, &tx, &auth).unwrap();
-            let hash = QueryHash::from_string(sql, auth.caller(), has_param);
-            Ok(Arc::new(Plan::new(plans, hash, sql.into())))
+            let (plans, requires_sender_binding) = SubscriptionPlan::compile(sql, &tx, &auth).unwrap();
+            let hash = QueryHash::from_string(sql, auth.caller(), requires_sender_binding);
+            let bind_env = BindEnv::for_sender_binding(requires_sender_binding, auth.caller());
+            Ok(Arc::new(Plan::new(plans, hash, sql.into(), bind_env)))
         })
     }
 
diff --git a/crates/core/src/subscription/query.rs b/crates/core/src/subscription/query.rs
index bea30f96b7f..ba4c6bd243f 100644
--- a/crates/core/src/subscription/query.rs
+++ b/crates/core/src/subscription/query.rs
@@ -7,6 +7,7 @@ use once_cell::sync::Lazy;
 use regex::Regex;
 use spacetimedb_datastore::locking_tx_datastore::state_view::StateView;
 use spacetimedb_execution::Datastore;
+use spacetimedb_expr::expr::BindEnv;
 use spacetimedb_lib::identity::AuthCtx;
 use spacetimedb_subscription::SubscriptionPlan;
 
@@ -31,9 +32,10 @@ pub fn compile_read_only_query(auth: &AuthCtx, tx: &Tx, input: &str) -> Result<P
     }
 
     let tx = SchemaViewer::new(tx, auth);
-    let (plans, has_param) = SubscriptionPlan::compile(input, &tx, auth)?;
-    let hash = QueryHash::from_string(input, auth.caller(), has_param);
-    Ok(Plan::new(plans, hash, input.to_owned()))
+    let (plans, requires_sender_binding) = SubscriptionPlan::compile(input, &tx, auth)?;
+    let hash = QueryHash::from_string(input, auth.caller(), requires_sender_binding);
+    let bind_env = BindEnv::for_sender_binding(requires_sender_binding, auth.caller());
+    Ok(Plan::new(plans, hash, input.to_owned(), bind_env))
 }
 
 /// Compile a string into a single read-only query with externally-computed hashes.
@@ -49,10 +51,11 @@ pub fn compile_query_with_hashes<Tx: Datastore + StateView>(
     }
 
     let tx = SchemaViewer::new(tx, auth);
-    let (plans, has_param) = SubscriptionPlan::compile(input, &tx, auth)?;
+    let (plans, requires_sender_binding) = SubscriptionPlan::compile(input, &tx, auth)?;
+    let bind_env = BindEnv::for_sender_binding(requires_sender_binding, auth.caller());
 
-    if auth.bypass_rls() || has_param {
-        return Ok(Plan::new(plans, hash_with_param, input.to_owned()));
+    if auth.bypass_rls() || requires_sender_binding {
+        return Ok(Plan::new(plans, hash_with_param, input.to_owned(), bind_env));
     }
-    Ok(Plan::new(plans, hash, input.to_owned()))
+    Ok(Plan::new(plans, hash, input.to_owned(), bind_env))
 }
diff --git a/crates/core/src/subscription/subscription.rs b/crates/core/src/subscription/subscription.rs
index a9f3bd12f62..fa59d1f89f2 100644
--- a/crates/core/src/subscription/subscription.rs
+++ b/crates/core/src/subscription/subscription.rs
@@ -4,6 +4,7 @@ use crate::db::relational_db::RelationalDB;
 use crate::error::DBError;
 use crate::sql::ast::SchemaViewer;
 use spacetimedb_datastore::locking_tx_datastore::state_view::StateView;
+use spacetimedb_expr::expr::BindEnv;
 use spacetimedb_lib::db::auth::StTableType;
 use spacetimedb_lib::identity::AuthCtx;
 use spacetimedb_schema::schema::TableSchema;
@@ -27,11 +28,12 @@ where
         .map(|schema| {
             let sql = format!("SELECT * FROM {}", schema.table_name);
             let tx = SchemaViewer::new(tx, auth);
-            SubscriptionPlan::compile(&sql, &tx, auth).map(|(plans, has_param)| {
+            SubscriptionPlan::compile(&sql, &tx, auth).map(|(plans, requires_sender_binding)| {
                 Plan::new(
                     plans,
-                    QueryHash::from_string(&sql, auth.caller(), auth.bypass_rls() || has_param),
+                    QueryHash::from_string(&sql, auth.caller(), auth.bypass_rls() || requires_sender_binding),
                     sql,
+                    BindEnv::for_sender_binding(requires_sender_binding, auth.caller()),
                 )
             })
         })
diff --git a/crates/execution/src/pipelined.rs b/crates/execution/src/pipelined.rs
index c9462fc8cad..2abd30c94a5 100644
--- a/crates/execution/src/pipelined.rs
+++ b/crates/execution/src/pipelined.rs
@@ -8,14 +8,23 @@ use itertools::Either;
 use spacetimedb_expr::expr::AggType;
 use spacetimedb_lib::{metrics::ExecutionMetrics, query::Delta, sats::size_of::SizeOf, AlgebraicValue, ProductValue};
 use spacetimedb_physical_plan::plan::{
-    HashJoin, IxJoin, IxScan, PhysicalExpr, PhysicalPlan, ProjectField, ProjectListPlan, ProjectPlan, Sarg, Semi,
-    TableScan, TupleField,
+    HashJoin, IxJoin, IxScan, PhysicalExpr, PhysicalPlan, ProjectField, ProjectListPlan, ProjectPlan, Sarg, SargValue,
+    Semi, TableScan, TupleField,
 };
 use spacetimedb_primitives::{ColId, ColList, IndexId, TableId};
 use spacetimedb_sats::product;
 
 use crate::{Datastore, DeltaStore, Row, Tuple};
 
+fn expect_bound_sarg_value(value: SargValue) -> AlgebraicValue {
+    match value {
+        SargValue::Literal(value) => value,
+        SargValue::Param(id, _) => panic!(
+            "unbound query parameter {id:?} reached pipelined index scan; bind parameters before constructing a PipelinedProject"
+        ),
+    }
+}
+
 /// An executor for explicit column projections.
 /// Note, this plan can only be constructed from the http api,
 /// which is not considered performance critical.
@@ -579,14 +588,17 @@ impl From<IxScan> for PipelinedIxDeltaScanRange {
                 arg: Sarg::Eq(_, v),
                 delta: Some(delta),
                 ..
-            } => Self {
-                table_id: schema.table_id,
-                index_id,
-                prefix: prefix.into_iter().map(|(_, v)| v).collect(),
-                lower: Bound::Included(v.clone()),
-                upper: Bound::Included(v),
-                delta,
-            },
+            } => {
+                let v = expect_bound_sarg_value(v);
+                Self {
+                    table_id: schema.table_id,
+                    index_id,
+                    prefix: prefix.into_iter().map(|(_, v)| v).collect(),
+                    lower: Bound::Included(v.clone()),
+                    upper: Bound::Included(v),
+                    delta,
+                }
+            }
             IxScan {
                 schema,
                 index_id,
@@ -706,7 +718,7 @@ impl From<IxScan> for PipelinedIxDeltaScanEq {
             } => Self {
                 table_id: schema.table_id,
                 index_id,
-                point: combine_prefix_and_last(prefix, last),
+                point: combine_prefix_and_last(prefix, expect_bound_sarg_value(last)),
                 delta,
             },
             IxScan { .. } => unreachable!(),
@@ -773,14 +785,17 @@ impl From<IxScan> for PipelinedIxScanRange {
                 index_id,
                 prefix,
                 arg: Sarg::Eq(_, v),
-            } => Self {
-                table_id: schema.table_id,
-                index_id,
-                limit,
-                prefix: prefix.into_iter().map(|(_, v)| v).collect(),
-                lower: Bound::Included(v.clone()),
-                upper: Bound::Included(v),
-            },
+            } => {
+                let v = expect_bound_sarg_value(v);
+                Self {
+                    table_id: schema.table_id,
+                    index_id,
+                    limit,
+                    prefix: prefix.into_iter().map(|(_, v)| v).collect(),
+                    lower: Bound::Included(v.clone()),
+                    upper: Bound::Included(v),
+                }
+            }
             IxScan {
                 schema,
                 limit,
@@ -909,7 +924,7 @@ impl From<IxScan> for PipelinedIxScanEq {
                 table_id: schema.table_id,
                 index_id,
                 limit,
-                point: combine_prefix_and_last(prefix, last),
+                point: combine_prefix_and_last(prefix, expect_bound_sarg_value(last)),
             },
             IxScan { .. } => unreachable!(),
         }
diff --git a/crates/expr/src/check.rs b/crates/expr/src/check.rs
index 01e151f4357..ca3bb0304f2 100644
--- a/crates/expr/src/check.rs
+++ b/crates/expr/src/check.rs
@@ -157,10 +157,9 @@ impl TypeChecker for SubChecker {
 }
 
 /// Parse and type check a subscription query
-pub fn parse_and_type_sub(sql: &str, tx: &impl SchemaView, auth: &AuthCtx) -> TypingResult<(ProjectName, bool)> {
+pub fn parse_and_type_sub(sql: &str, tx: &impl SchemaView, _auth: &AuthCtx) -> TypingResult<(ProjectName, bool)> {
     let ast = parse_subscription(sql)?;
     let has_param = ast.has_parameter();
-    let ast = ast.resolve_sender(auth.caller());
     expect_table_type(SubChecker::type_ast(ast, tx)?).map(|plan| (plan, has_param))
 }
 
diff --git a/crates/expr/src/expr.rs b/crates/expr/src/expr.rs
index c9753e5c195..ee6143c419b 100644
--- a/crates/expr/src/expr.rs
+++ b/crates/expr/src/expr.rs
@@ -1,11 +1,73 @@
 use spacetimedb_data_structures::map::HashSet;
-use spacetimedb_lib::{query::Delta, AlgebraicType, AlgebraicValue};
+use spacetimedb_lib::{query::Delta, AlgebraicType, AlgebraicValue, Identity};
 use spacetimedb_primitives::{TableId, ViewId};
 use spacetimedb_sats::raw_identifier::RawIdentifier;
 use spacetimedb_schema::{identifier::Identifier, schema::TableOrViewSchema};
 use spacetimedb_sql_parser::ast::{BinOp, LogOp};
 use std::sync::Arc;
 
+/// A formal parameter slot in a typed query plan.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
+pub struct ParamId(pub u16);
+
+impl ParamId {
+    /// The only parameter currently supported by SQL syntax: `:sender`.
+    pub const SENDER: Self = Self(0);
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+enum ParamBinding {
+    Sender(Identity),
+}
+
+impl ParamBinding {
+    fn value_for_type(&self, ty: &AlgebraicType) -> Option<AlgebraicValue> {
+        match self {
+            Self::Sender(sender) if ty.is_identity() => Some((*sender).into()),
+            // Preserve existing `:sender` behavior for legacy filters that compare it to bytes.
+            Self::Sender(sender) if ty.is_bytes() => Some(AlgebraicValue::Bytes(
+                sender.to_be_byte_array().to_vec().into_boxed_slice(),
+            )),
+            Self::Sender(_) => None,
+        }
+    }
+}
+
+/// Runtime parameter bindings for a parameterized query plan.
+#[derive(Debug, Clone, Default, PartialEq, Eq)]
+pub struct BindEnv {
+    values: Vec<ParamBinding>,
+}
+
+impl BindEnv {
+    pub fn empty() -> Self {
+        Self::default()
+    }
+
+    pub fn sender(sender: Identity) -> Self {
+        Self {
+            values: vec![ParamBinding::Sender(sender)],
+        }
+    }
+
+    pub fn for_sender_binding(requires_sender_binding: bool, sender: Identity) -> Self {
+        if requires_sender_binding {
+            Self::sender(sender)
+        } else {
+            Self::empty()
+        }
+    }
+
+    pub fn get(&self, id: ParamId, ty: &AlgebraicType) -> Option<AlgebraicValue> {
+        self.values.get(id.0 as usize)?.value_for_type(ty)
+    }
+
+    pub fn expect(&self, id: ParamId, ty: &AlgebraicType, context: &str) -> AlgebraicValue {
+        self.get(id, ty)
+            .unwrap_or_else(|| panic!("missing or mistyped binding for query parameter {id:?} while {context}"))
+    }
+}
+
 pub trait CollectViews {
     fn collect_views(&self, views: &mut HashSet<ViewId>);
 }
@@ -383,6 +445,8 @@ pub enum Expr {
     LogOp(LogOp, Box<Expr>, Box<Expr>),
     /// A typed literal expression
     Value(AlgebraicValue, AlgebraicType),
+    /// A typed runtime parameter.
+    Param(ParamId, AlgebraicType),
     /// A field projection
     Field(FieldProject),
 }
@@ -396,7 +460,7 @@ impl Expr {
                 a.visit(f);
                 b.visit(f);
             }
-            Self::Value(..) | Self::Field(..) => {}
+            Self::Value(..) | Self::Param(..) | Self::Field(..) => {}
         }
     }
 
@@ -408,7 +472,7 @@ impl Expr {
                 a.visit_mut(f);
                 b.visit_mut(f);
             }
-            Self::Value(..) | Self::Field(..) => {}
+            Self::Value(..) | Self::Param(..) | Self::Field(..) => {}
         }
     }
 
@@ -426,7 +490,7 @@ impl Expr {
     pub fn ty(&self) -> &AlgebraicType {
         match self {
             Self::BinOp(..) | Self::LogOp(..) => &AlgebraicType::Bool,
-            Self::Value(_, ty) | Self::Field(FieldProject { ty, .. }) => ty,
+            Self::Value(_, ty) | Self::Param(_, ty) | Self::Field(FieldProject { ty, .. }) => ty,
         }
     }
 }
diff --git a/crates/expr/src/lib.rs b/crates/expr/src/lib.rs
index 2a6f02f5a21..6a9fcb9570f 100644
--- a/crates/expr/src/lib.rs
+++ b/crates/expr/src/lib.rs
@@ -9,7 +9,7 @@ use errors::{DuplicateName, InvalidLiteral, InvalidOp, InvalidWildcard, Unexpect
 use ethnum::i256;
 use ethnum::u256;
 use expr::AggType;
-use expr::{Expr, FieldProject, ProjectList, ProjectName, RelExpr};
+use expr::{Expr, FieldProject, ParamId, ProjectList, ProjectName, RelExpr};
 use spacetimedb_data_structures::map::HashCollectionExt as _;
 use spacetimedb_data_structures::map::HashSet;
 use spacetimedb_lib::ser::Serialize;
@@ -19,7 +19,7 @@ use spacetimedb_sats::algebraic_type::fmt::fmt_algebraic_type;
 use spacetimedb_sats::algebraic_value::ser::ValueSerializer;
 use spacetimedb_sats::uuid::Uuid;
 use spacetimedb_schema::schema::ColumnSchema;
-use spacetimedb_sql_parser::ast::{self, BinOp, ProjectElem, SqlExpr, SqlIdent, SqlLiteral};
+use spacetimedb_sql_parser::ast::{self, BinOp, Parameter, ProjectElem, SqlExpr, SqlIdent, SqlLiteral};
 use spacetimedb_sql_parser::parser::recursion;
 use std::{ops::Deref, str::FromStr};
 
@@ -99,6 +99,13 @@ fn _type_expr(vars: &Relvars, expr: SqlExpr, expected: Option<&AlgebraicType>, d
             parse(&v, ty).map_err(|_| InvalidLiteral::new(v.into_string(), ty))?,
             ty.clone(),
         )),
+        (SqlExpr::Param(Parameter::Sender), Some(ty)) if ty.is_identity() || ty.is_bytes() => {
+            Ok(Expr::Param(ParamId::SENDER, ty.clone()))
+        }
+        (SqlExpr::Param(Parameter::Sender), Some(ty)) => {
+            Err(UnexpectedType::new(&AlgebraicType::identity(), ty).into())
+        }
+        (SqlExpr::Param(Parameter::Sender), None) => Err(Unresolved::Literal.into()),
         (SqlExpr::Field(SqlIdent(table), SqlIdent(field)), expected) => {
             let table_type = vars.deref().get(&*table).ok_or_else(|| Unresolved::var(&table))?;
             let ColumnSchema { col_pos, col_type, .. } = table_type
@@ -123,7 +130,9 @@ fn _type_expr(vars: &Relvars, expr: SqlExpr, expected: Option<&AlgebraicType>, d
             let b = _type_expr(vars, *b, Some(&AlgebraicType::Bool), depth + 1)?;
             Ok(Expr::LogOp(op, Box::new(a), Box::new(b)))
         }
-        (SqlExpr::Bin(a, b, op), None | Some(AlgebraicType::Bool)) if matches!(&*a, SqlExpr::Lit(_)) => {
+        (SqlExpr::Bin(a, b, op), None | Some(AlgebraicType::Bool))
+            if matches!(&*a, SqlExpr::Lit(_) | SqlExpr::Param(_)) =>
+        {
             let b = _type_expr(vars, *b, None, depth + 1)?;
             let a = _type_expr(vars, *a, Some(b.ty()), depth + 1)?;
             if !op_supports_type(op, a.ty()) {
@@ -140,9 +149,8 @@ fn _type_expr(vars: &Relvars, expr: SqlExpr, expected: Option<&AlgebraicType>, d
             Ok(Expr::BinOp(op, Box::new(a), Box::new(b)))
         }
         (SqlExpr::Bin(..) | SqlExpr::Log(..), Some(ty)) => Err(UnexpectedType::new(&AlgebraicType::Bool, ty).into()),
-        // Both unqualified names as well as parameters are syntactic constructs.
-        // Unqualified names are qualified and parameters are resolved before type checking.
-        (SqlExpr::Var(_) | SqlExpr::Param(_), _) => unreachable!(),
+        // Unqualified names are syntactic constructs qualified before type checking.
+        (SqlExpr::Var(_), _) => unreachable!(),
     }
 }
 
diff --git a/crates/expr/src/rls.rs b/crates/expr/src/rls.rs
index 7382bd5e630..b7db1f5c92e 100644
--- a/crates/expr/src/rls.rs
+++ b/crates/expr/src/rls.rs
@@ -487,7 +487,7 @@ mod tests {
 
     use crate::{
         check::{parse_and_type_sub, test_utils::build_module_def, SchemaView},
-        expr::{Expr, FieldProject, LeftDeepJoin, ProjectName, RelExpr, Relvar},
+        expr::{Expr, FieldProject, LeftDeepJoin, ParamId, ProjectName, RelExpr, Relvar},
     };
 
     use super::resolve_views_for_sub;
@@ -602,7 +602,7 @@ mod tests {
                             field: 0,
                             ty: AlgebraicType::identity(),
                         })),
-                        Box::new(Expr::Value(Identity::ONE.into(), AlgebraicType::identity()))
+                        Box::new(Expr::Param(ParamId::SENDER, AlgebraicType::identity()))
                     )
                 ),
                 "users".into()
@@ -649,7 +649,7 @@ mod tests {
                                         field: 0,
                                         ty: AlgebraicType::identity(),
                                     })),
-                                    Box::new(Expr::Value(Identity::ONE.into(), AlgebraicType::identity())),
+                                    Box::new(Expr::Param(ParamId::SENDER, AlgebraicType::identity())),
                                 ),
                             )),
                             Expr::BinOp(
@@ -700,7 +700,7 @@ mod tests {
                                     field: 0,
                                     ty: AlgebraicType::identity(),
                                 })),
-                                Box::new(Expr::Value(Identity::ONE.into(), AlgebraicType::identity())),
+                                Box::new(Expr::Param(ParamId::SENDER, AlgebraicType::identity())),
                             ),
                         )),
                         Expr::BinOp(
diff --git a/crates/expr/src/statement.rs b/crates/expr/src/statement.rs
index b7422dd031c..8083d813706 100644
--- a/crates/expr/src/statement.rs
+++ b/crates/expr/src/statement.rs
@@ -450,8 +450,8 @@ impl TypeChecker for SqlChecker {
     }
 }
 
-pub fn parse_and_type_sql(sql: &str, tx: &impl SchemaView, auth: &AuthCtx) -> TypingResult<Statement> {
-    match parse_sql(sql)?.resolve_sender(auth.caller()) {
+pub fn parse_and_type_sql(sql: &str, tx: &impl SchemaView, _auth: &AuthCtx) -> TypingResult<Statement> {
+    match parse_sql(sql)? {
         SqlAst::Select(ast) => Ok(Statement::Select(SqlChecker::type_ast(ast, tx)?)),
         SqlAst::Insert(insert) => Ok(Statement::DML(DML::Insert(type_insert(insert, tx)?))),
         SqlAst::Delete(delete) => Ok(Statement::DML(DML::Delete(type_delete(delete, tx)?))),
diff --git a/crates/physical-plan/src/compile.rs b/crates/physical-plan/src/compile.rs
index 1bda4fd65cd..e5b59cca384 100644
--- a/crates/physical-plan/src/compile.rs
+++ b/crates/physical-plan/src/compile.rs
@@ -21,6 +21,7 @@ fn compile_expr(expr: Expr, var: &mut impl VarLabel) -> PhysicalExpr {
             PhysicalExpr::BinOp(op, a, b)
         }
         Expr::Value(v, _) => PhysicalExpr::Value(v),
+        Expr::Param(id, ty) => PhysicalExpr::Param(id, ty),
         Expr::Field(proj) => PhysicalExpr::Field(compile_field_project(var, proj)),
     }
 }
diff --git a/crates/physical-plan/src/dml.rs b/crates/physical-plan/src/dml.rs
index f6a0aa1aa73..fa6eb88626b 100644
--- a/crates/physical-plan/src/dml.rs
+++ b/crates/physical-plan/src/dml.rs
@@ -2,7 +2,7 @@ use std::sync::Arc;
 
 use anyhow::Result;
 use spacetimedb_expr::{
-    expr::{ProjectName, RelExpr, Relvar},
+    expr::{BindEnv, ProjectName, RelExpr, Relvar},
     statement::{TableDelete, TableInsert, TableUpdate},
 };
 use spacetimedb_lib::{identity::AuthCtx, AlgebraicValue, ProductValue};
@@ -27,6 +27,15 @@ impl MutationPlan {
             Self::Update(plan) => Ok(Self::Update(plan.optimize(auth)?)),
         }
     }
+
+    /// Replace runtime parameters with bound values.
+    pub fn bind_params(self, bind_env: &BindEnv) -> Self {
+        match self {
+            Self::Insert(..) => self,
+            Self::Delete(plan) => Self::Delete(plan.bind_params(bind_env)),
+            Self::Update(plan) => Self::Update(plan.bind_params(bind_env)),
+        }
+    }
 }
 
 /// A plan for inserting rows into a table
@@ -57,6 +66,12 @@ impl DeletePlan {
         Ok(Self { table, filter })
     }
 
+    fn bind_params(self, bind_env: &BindEnv) -> Self {
+        let Self { table, filter } = self;
+        let filter = filter.bind_params(bind_env);
+        Self { table, filter }
+    }
+
     /// Logical to physical conversion
     pub(crate) fn compile(delete: TableDelete) -> Self {
         let TableDelete { table, filter } = delete;
@@ -91,6 +106,12 @@ impl UpdatePlan {
         Ok(Self { columns, table, filter })
     }
 
+    fn bind_params(self, bind_env: &BindEnv) -> Self {
+        let Self { table, columns, filter } = self;
+        let filter = filter.bind_params(bind_env);
+        Self { columns, table, filter }
+    }
+
     /// Logical to physical conversion
     pub(crate) fn compile(update: TableUpdate) -> Self {
         let TableUpdate { table, columns, filter } = update;
diff --git a/crates/physical-plan/src/plan.rs b/crates/physical-plan/src/plan.rs
index 0db3e7910ce..c273924a7ed 100644
--- a/crates/physical-plan/src/plan.rs
+++ b/crates/physical-plan/src/plan.rs
@@ -3,10 +3,12 @@ use derive_more::From;
 use either::Either;
 use spacetimedb_data_structures::map::HashSet;
 use spacetimedb_expr::{
-    expr::{AggType, CollectViews},
+    expr::{AggType, BindEnv, CollectViews, ParamId},
     StatementSource,
 };
-use spacetimedb_lib::{identity::AuthCtx, query::Delta, sats::size_of::SizeOf, AlgebraicValue, ProductValue};
+use spacetimedb_lib::{
+    identity::AuthCtx, query::Delta, sats::size_of::SizeOf, AlgebraicType, AlgebraicValue, ProductValue,
+};
 use spacetimedb_primitives::{ColId, ColSet, IndexId, TableId, ViewId};
 use spacetimedb_schema::schema::{IndexSchema, TableSchema};
 use spacetimedb_sql_parser::ast::{BinOp, LogOp};
@@ -128,6 +130,21 @@ impl ProjectPlan {
             Self::None(plan) | Self::Name(plan, ..) => plan.reads_from_event_table(),
         }
     }
+
+    /// Replace runtime parameters with bound values.
+    pub fn bind_params(self, bind_env: &BindEnv) -> Self {
+        match self {
+            Self::None(plan) => Self::None(plan.bind_params(bind_env)),
+            Self::Name(plan, label, pos) => Self::Name(plan.bind_params(bind_env), label, pos),
+        }
+    }
+
+    /// Returns whether this plan contains a runtime parameter.
+    pub fn requires_param(&self, id: ParamId) -> bool {
+        match self {
+            Self::None(plan) | Self::Name(plan, ..) => plan.requires_param(id),
+        }
+    }
 }
 
 /// Physical plans always terminate with a projection.
@@ -165,6 +182,28 @@ pub enum ProjectListPlan {
 }
 
 impl ProjectListPlan {
+    /// Replace runtime parameters with bound values.
+    pub fn bind_params(self, bind_env: &BindEnv) -> Self {
+        match self {
+            Self::Name(plans) => Self::Name(plans.into_iter().map(|plan| plan.bind_params(bind_env)).collect()),
+            Self::List(plans, fields) => Self::List(
+                plans.into_iter().map(|plan| plan.bind_params(bind_env)).collect(),
+                fields,
+            ),
+            Self::Limit(plan, n) => Self::Limit(Box::new((*plan).bind_params(bind_env)), n),
+            Self::Agg(plans, agg) => Self::Agg(plans.into_iter().map(|plan| plan.bind_params(bind_env)).collect(), agg),
+        }
+    }
+
+    /// Returns whether this plan contains a runtime parameter.
+    pub fn requires_param(&self, id: ParamId) -> bool {
+        match self {
+            Self::Name(plans) => plans.iter().any(|plan| plan.requires_param(id)),
+            Self::List(plans, _) | Self::Agg(plans, _) => plans.iter().any(|plan| plan.requires_param(id)),
+            Self::Limit(plan, _) => plan.requires_param(id),
+        }
+    }
+
     pub fn optimize(self, auth: &AuthCtx) -> Result<Self> {
         match self {
             Self::Name(plan) => Ok(Self::Name(
@@ -339,6 +378,15 @@ impl PhysicalPlan {
         ok
     }
 
+    /// Returns whether this plan contains a runtime parameter.
+    pub fn requires_param(&self, id: ParamId) -> bool {
+        self.any(&|plan| match plan {
+            Self::IxScan(scan, _) => scan.arg.requires_param(id),
+            Self::Filter(_, expr) => expr.requires_param(id),
+            Self::TableScan(..) | Self::IxJoin(..) | Self::HashJoin(..) | Self::NLJoin(..) => false,
+        })
+    }
+
     /// Applies `f` recursively to all subplans
     pub fn map(self, f: &impl Fn(Self) -> Self) -> Self {
         match f(self) {
@@ -363,6 +411,69 @@ impl PhysicalPlan {
         }
     }
 
+    /// Replace runtime parameters with bound values.
+    pub fn bind_params(self, bind_env: &BindEnv) -> Self {
+        match self {
+            Self::TableScan(..) => self,
+            Self::IxScan(mut scan, label) => {
+                scan.arg = scan.arg.bind_params(bind_env);
+                Self::IxScan(scan, label)
+            }
+            Self::IxJoin(
+                IxJoin {
+                    lhs,
+                    rhs,
+                    rhs_label,
+                    rhs_index,
+                    rhs_prefix,
+                    rhs_field,
+                    unique,
+                    lhs_field,
+                    rhs_delta,
+                },
+                semi,
+            ) => Self::IxJoin(
+                IxJoin {
+                    lhs: Box::new(lhs.bind_params(bind_env)),
+                    rhs,
+                    rhs_label,
+                    rhs_index,
+                    rhs_prefix,
+                    rhs_field,
+                    unique,
+                    lhs_field,
+                    rhs_delta,
+                },
+                semi,
+            ),
+            Self::HashJoin(
+                HashJoin {
+                    lhs,
+                    rhs,
+                    lhs_field,
+                    rhs_field,
+                    unique,
+                },
+                semi,
+            ) => Self::HashJoin(
+                HashJoin {
+                    lhs: Box::new(lhs.bind_params(bind_env)),
+                    rhs: Box::new(rhs.bind_params(bind_env)),
+                    lhs_field,
+                    rhs_field,
+                    unique,
+                },
+                semi,
+            ),
+            Self::NLJoin(lhs, rhs) => {
+                Self::NLJoin(Box::new(lhs.bind_params(bind_env)), Box::new(rhs.bind_params(bind_env)))
+            }
+            Self::Filter(input, expr) => {
+                Self::Filter(Box::new(input.bind_params(bind_env)), expr.bind_params(bind_env))
+            }
+        }
+    }
+
     /// Applies `f` to a subplan if `ok` returns a match.
     /// Recurses until an `ok` match is found.
     pub fn map_if<Info>(
@@ -474,9 +585,9 @@ impl PhysicalPlan {
     /// 3. Turn filters into index scans if possible
     /// 4. Determine index and semijoins
     /// 5. Compute positions for tuple labels
-    pub fn optimize(self, auth: &AuthCtx, reqs: Vec<Label>) -> Result<Self> {
+    pub fn optimize(self, _auth: &AuthCtx, reqs: Vec<Label>) -> Result<Self> {
         let optimized = self
-            .expand_views(auth)
+            .expand_views()
             .map(&Self::canonicalize)
             .apply_rec::<PushConstAnd>()?
             .apply_rec::<PushConstEq>()?
@@ -555,18 +666,18 @@ impl PhysicalPlan {
     /// ```sql
     /// SELECT * FROM my_view WHERE sender = :sender
     /// ```
-    fn expand_views(self, auth: &AuthCtx) -> Self {
+    fn expand_views(self) -> Self {
         match self {
             Self::TableScan(scan, label) if scan.schema.is_view() && !scan.schema.is_anonymous_view() => Self::Filter(
                 Box::new(Self::TableScan(scan, label)),
                 PhysicalExpr::BinOp(
                     BinOp::Eq,
-                    Box::new(PhysicalExpr::Value(auth.caller().into())),
                     Box::new(PhysicalExpr::Field(TupleField {
                         label,
                         label_pos: None,
                         field_pos: 0,
                     })),
+                    Box::new(PhysicalExpr::Param(ParamId::SENDER, AlgebraicType::identity())),
                 ),
             ),
             Self::IxJoin(
@@ -584,7 +695,7 @@ impl PhysicalPlan {
                 semi,
             ) => Self::IxJoin(
                 IxJoin {
-                    lhs: Box::new(lhs.expand_views(auth)),
+                    lhs: Box::new(lhs.expand_views()),
                     rhs,
                     rhs_label,
                     rhs_index,
@@ -607,16 +718,16 @@ impl PhysicalPlan {
                 semi,
             ) => Self::HashJoin(
                 HashJoin {
-                    lhs: Box::new(lhs.expand_views(auth)),
-                    rhs: Box::new(rhs.expand_views(auth)),
+                    lhs: Box::new(lhs.expand_views()),
+                    rhs: Box::new(rhs.expand_views()),
                     lhs_field,
                     rhs_field,
                     unique,
                 },
                 semi,
             ),
-            Self::Filter(input, expr) => Self::Filter(Box::new(input.expand_views(auth)), expr),
-            Self::NLJoin(lhs, rhs) => Self::NLJoin(Box::new(lhs.expand_views(auth)), Box::new(rhs.expand_views(auth))),
+            Self::Filter(input, expr) => Self::Filter(Box::new(input.expand_views()), expr),
+            Self::NLJoin(lhs, rhs) => Self::NLJoin(Box::new(lhs.expand_views()), Box::new(rhs.expand_views())),
             Self::TableScan(..) | Self::IxScan(..) => self,
         }
     }
@@ -686,7 +797,7 @@ impl PhysicalPlan {
             Self::Filter(input, expr) => {
                 let move_value_to_rhs = |expr| match expr {
                     PhysicalExpr::BinOp(op, value, expr)
-                        if matches!(&*value, PhysicalExpr::Value(_)) && matches!(&*expr, PhysicalExpr::Field(..)) =>
+                        if value.is_sarg_value() && matches!(&*expr, PhysicalExpr::Field(..)) =>
                     {
                         match op {
                             BinOp::Eq => PhysicalExpr::BinOp(BinOp::Eq, expr, value),
@@ -957,7 +1068,8 @@ impl PhysicalPlan {
                 let mut cols: Vec<_> = cols.iter().collect();
                 expr.visit(&mut |plan| {
                     if let PhysicalExpr::BinOp(BinOp::Eq, expr, value) = plan
-                        && let (PhysicalExpr::Field(proj), PhysicalExpr::Value(..)) = (&**expr, &**value)
+                        && let PhysicalExpr::Field(proj) = &**expr
+                        && value.is_sarg_value()
                         && proj.label == *label
                     {
                         cols.push(proj.field_pos.into());
@@ -1103,8 +1215,8 @@ impl PhysicalPlan {
                     ..
                 },
                 _,
-            ) if scan.prefix.is_empty() => {
-                args.push((scan.schema.table_id, *col_id, value.clone()));
+            ) if scan.prefix.is_empty() && value.as_literal().is_some() => {
+                args.push((scan.schema.table_id, *col_id, value.as_literal().unwrap().clone()));
             }
             PhysicalPlan::Filter(input, PhysicalExpr::BinOp(BinOp::Eq, a, b)) => {
                 if let (PhysicalExpr::Field(field), PhysicalExpr::Value(value)) = (&**a, &**b) {
@@ -1214,7 +1326,7 @@ pub struct IxScan {
 /// An index [S]earch [arg]ument
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub enum Sarg {
-    Eq(ColId, AlgebraicValue),
+    Eq(ColId, SargValue),
     /// NOTE(centril): We currently never construct this variant.
     /// We do have non-ranged hash indices.
     /// This means that when we get around to using this variant,
@@ -1229,6 +1341,65 @@ pub enum Sarg {
     Range(ColId, Bound<AlgebraicValue>, Bound<AlgebraicValue>),
 }
 
+impl Sarg {
+    pub fn requires_param(&self, id: ParamId) -> bool {
+        match self {
+            Self::Eq(_, value) => value.requires_param(id),
+            Self::Range(..) => false,
+        }
+    }
+
+    pub fn bind_params(self, bind_env: &BindEnv) -> Self {
+        match self {
+            Self::Eq(col, value) => Self::Eq(col, value.bind_params(bind_env)),
+            Self::Range(..) => self,
+        }
+    }
+}
+
+/// A search argument value can be a literal or a runtime parameter.
+///
+/// `Param` is optimizer/template state only; executor construction and concrete
+/// search-arg extraction require it to be bound to a literal first.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum SargValue {
+    Literal(AlgebraicValue),
+    Param(ParamId, AlgebraicType),
+}
+
+impl SargValue {
+    pub fn requires_param(&self, id: ParamId) -> bool {
+        matches!(self, Self::Param(param_id, _) if *param_id == id)
+    }
+
+    pub fn bind_params(self, bind_env: &BindEnv) -> Self {
+        match self {
+            Self::Literal(value) => Self::Literal(value),
+            Self::Param(id, ty) => Self::Literal(bind_env.expect(id, &ty, "binding index search argument")),
+        }
+    }
+
+    pub fn into_literal(self) -> Option<AlgebraicValue> {
+        match self {
+            Self::Literal(value) => Some(value),
+            Self::Param(..) => None,
+        }
+    }
+
+    pub fn as_literal(&self) -> Option<&AlgebraicValue> {
+        match self {
+            Self::Literal(value) => Some(value),
+            Self::Param(..) => None,
+        }
+    }
+}
+
+impl From<AlgebraicValue> for SargValue {
+    fn from(value: AlgebraicValue) -> Self {
+        Self::Literal(value)
+    }
+}
+
 /// A join of two relations on a single equality condition.
 /// It builds a hash table for the rhs and streams the lhs.
 #[derive(Debug, Clone, PartialEq, Eq)]
@@ -1286,6 +1457,8 @@ pub enum PhysicalExpr {
     BinOp(BinOp, Box<PhysicalExpr>, Box<PhysicalExpr>),
     /// A constant algebraic value
     Value(AlgebraicValue),
+    /// A runtime parameter.
+    Param(ParamId, AlgebraicType),
     /// A field projection expression
     Field(TupleField),
 }
@@ -1311,6 +1484,48 @@ impl ProjectField for &'_ ProductValue {
 }
 
 impl PhysicalExpr {
+    /// Can this expression be used as the constant side of an index search argument?
+    pub fn is_sarg_value(&self) -> bool {
+        matches!(self, Self::Value(_) | Self::Param(..))
+    }
+
+    pub fn into_sarg_value(self) -> Option<SargValue> {
+        match self {
+            Self::Value(value) => Some(SargValue::Literal(value)),
+            Self::Param(id, ty) => Some(SargValue::Param(id, ty)),
+            _ => None,
+        }
+    }
+
+    pub fn as_literal(&self) -> Option<&AlgebraicValue> {
+        match self {
+            Self::Value(value) => Some(value),
+            Self::Param(..) | Self::Field(..) | Self::BinOp(..) | Self::LogOp(..) => None,
+        }
+    }
+
+    pub fn requires_param(&self, id: ParamId) -> bool {
+        let mut found = false;
+        self.visit(&mut |expr| {
+            found = found || matches!(expr, Self::Param(param_id, _) if *param_id == id);
+        });
+        found
+    }
+
+    /// Replace runtime parameters with bound values.
+    pub fn bind_params(self, bind_env: &BindEnv) -> Self {
+        match self {
+            Self::Param(id, ty) => Self::Value(bind_env.expect(id, &ty, "binding scalar expression")),
+            Self::BinOp(op, a, b) => {
+                Self::BinOp(op, Box::new(a.bind_params(bind_env)), Box::new(b.bind_params(bind_env)))
+            }
+            Self::LogOp(op, exprs) => {
+                Self::LogOp(op, exprs.into_iter().map(|expr| expr.bind_params(bind_env)).collect())
+            }
+            Self::Field(..) | Self::Value(..) => self,
+        }
+    }
+
     /// Walks the expression tree and calls `f` on every subexpression
     pub fn visit(&self, f: &mut impl FnMut(&Self)) {
         f(self);
@@ -1349,6 +1564,7 @@ impl PhysicalExpr {
     pub fn map(self, f: &impl Fn(Self) -> Self) -> Self {
         match f(self) {
             value @ Self::Value(..) => value,
+            param @ Self::Param(..) => param,
             field @ Self::Field(..) => field,
             Self::BinOp(op, a, b) => Self::BinOp(op, Box::new(a.map(f)), Box::new(b.map(f))),
             Self::LogOp(op, exprs) => Self::LogOp(op, exprs.into_iter().map(|expr| expr.map(f)).collect()),
@@ -1410,6 +1626,9 @@ impl PhysicalExpr {
                 Cow::Owned(value)
             }
             Self::Value(v) => Cow::Borrowed(v),
+            Self::Param(id, _) => panic!(
+                "unbound query parameter {id:?} reached scalar evaluation; bind parameters before executing a physical plan"
+            ),
         }
     }
 
@@ -1428,7 +1647,7 @@ impl PhysicalExpr {
                     .collect(),
             ),
             Self::BinOp(op, a, b) => Self::BinOp(op, Box::new(a.flatten()), Box::new(b.flatten())),
-            Self::Field(..) | Self::Value(..) => self,
+            Self::Field(..) | Self::Value(..) | Self::Param(..) => self,
         }
     }
 }
@@ -1447,13 +1666,13 @@ mod tests {
     use pretty_assertions::assert_eq;
     use spacetimedb_expr::{
         check::{SchemaView, TypingResult},
-        expr::ProjectName,
+        expr::{BindEnv, ParamId, ProjectName},
         statement::{parse_and_type_sql, Statement},
     };
     use spacetimedb_lib::{
         db::auth::{StAccess, StTableType},
         identity::AuthCtx,
-        AlgebraicType, AlgebraicValue,
+        AlgebraicType, AlgebraicValue, Identity,
     };
     use spacetimedb_primitives::{ColId, ColList, ColSet, TableId};
     use spacetimedb_schema::{
@@ -1627,6 +1846,94 @@ mod tests {
         };
     }
 
+    #[test]
+    fn sender_param_binds_after_planning() {
+        let t_id = TableId(1);
+        let t = Arc::new(schema(
+            t_id,
+            "t",
+            &[("identity", AlgebraicType::identity())],
+            &[&[0]],
+            &[&[0]],
+            Some(0),
+        ));
+        let db = SchemaViewer { schemas: vec![t] };
+        let auth = AuthCtx::for_testing();
+        let lp = parse_and_type_sub("select * from t where identity = :sender", &db).unwrap();
+        let pp = compile_select(lp).optimize(&auth).unwrap();
+
+        match pp.physical_plan() {
+            PhysicalPlan::IxScan(
+                IxScan {
+                    arg: Sarg::Eq(ColId(0), value),
+                    ..
+                },
+                _,
+            ) => {
+                assert!(matches!(
+                    value,
+                    super::SargValue::Param(ParamId::SENDER, ty) if ty.is_identity()
+                ));
+            }
+            plan => panic!("unexpected plan: {plan:#?}"),
+        }
+
+        let bind_sender = |sender| match pp.clone().bind_params(&BindEnv::sender(sender)) {
+            ProjectPlan::None(PhysicalPlan::IxScan(
+                IxScan {
+                    arg: Sarg::Eq(ColId(0), value),
+                    ..
+                },
+                _,
+            )) => value,
+            plan => panic!("unexpected bound plan: {plan:#?}"),
+        };
+
+        assert_eq!(
+            bind_sender(Identity::ZERO),
+            super::SargValue::Literal(Identity::ZERO.into())
+        );
+        assert_eq!(
+            bind_sender(Identity::ONE),
+            super::SargValue::Literal(Identity::ONE.into())
+        );
+    }
+
+    #[test]
+    fn sender_param_can_bind_as_bytes() {
+        let t_id = TableId(1);
+        let t = Arc::new(schema(
+            t_id,
+            "t",
+            &[("bytes", AlgebraicType::bytes())],
+            &[&[0]],
+            &[&[0]],
+            Some(0),
+        ));
+        let db = SchemaViewer { schemas: vec![t] };
+        let auth = AuthCtx::for_testing();
+        let lp = parse_and_type_sub("select * from t where bytes = :sender", &db).unwrap();
+        let pp = compile_select(lp).optimize(&auth).unwrap();
+
+        let value = match pp.bind_params(&BindEnv::sender(Identity::ONE)) {
+            ProjectPlan::None(PhysicalPlan::IxScan(
+                IxScan {
+                    arg: Sarg::Eq(ColId(0), value),
+                    ..
+                },
+                _,
+            )) => value,
+            plan => panic!("unexpected bound plan: {plan:#?}"),
+        };
+
+        assert_eq!(
+            value,
+            super::SargValue::Literal(AlgebraicValue::Bytes(
+                Identity::ONE.to_be_byte_array().to_vec().into_boxed_slice()
+            ))
+        );
+    }
+
     /// Given the following operator notation:
     ///
     /// x:  join
@@ -1800,13 +2107,11 @@ mod tests {
         match plan {
             PhysicalPlan::IxScan(
                 IxScan {
-                    schema,
-                    prefix,
-                    arg: Sarg::Eq(ColId(0), AlgebraicValue::U64(5)),
-                    ..
+                    schema, prefix, arg, ..
                 },
                 _,
             ) => {
+                assert_eq!(arg, Sarg::Eq(ColId(0), AlgebraicValue::U64(5).into()));
                 assert!(prefix.is_empty());
                 assert_eq!(schema.table_id, u_id);
             }
@@ -1973,13 +2278,11 @@ mod tests {
         match rhs {
             PhysicalPlan::IxScan(
                 IxScan {
-                    schema,
-                    prefix,
-                    arg: Sarg::Eq(ColId(0), AlgebraicValue::U64(5)),
-                    ..
+                    schema, prefix, arg, ..
                 },
                 _,
             ) => {
+                assert_eq!(arg, Sarg::Eq(ColId(0), AlgebraicValue::U64(5).into()));
                 assert!(prefix.is_empty());
                 assert_eq!(schema.table_id, w_id);
             }
@@ -2036,13 +2339,11 @@ mod tests {
         match plan {
             PhysicalPlan::IxScan(
                 IxScan {
-                    schema,
-                    prefix,
-                    arg: Sarg::Eq(ColId(0), AlgebraicValue::U64(5)),
-                    ..
+                    schema, prefix, arg, ..
                 },
                 _,
             ) => {
+                assert_eq!(arg, Sarg::Eq(ColId(0), AlgebraicValue::U64(5).into()));
                 assert!(prefix.is_empty());
                 assert_eq!(schema.table_id, m_id);
             }
@@ -2087,7 +2388,7 @@ mod tests {
                 _,
             )) => {
                 assert_eq!(schema.table_id, t_id);
-                assert_eq!(arg, Sarg::Eq(ColId(3), AlgebraicValue::U8(5)));
+                assert_eq!(arg, Sarg::Eq(ColId(3), AlgebraicValue::U8(5).into()));
                 assert_eq!(
                     prefix,
                     vec![(ColId(1), AlgebraicValue::U8(3)), (ColId(2), AlgebraicValue::U8(4))]
@@ -2109,7 +2410,7 @@ mod tests {
                 _,
             )) => {
                 assert_eq!(schema.table_id, t_id);
-                assert_eq!(arg, Sarg::Eq(ColId(3), AlgebraicValue::U8(5)));
+                assert_eq!(arg, Sarg::Eq(ColId(3), AlgebraicValue::U8(5).into()));
                 assert_eq!(
                     prefix,
                     vec![(ColId(1), AlgebraicValue::U8(3)), (ColId(2), AlgebraicValue::U8(4))]
@@ -2140,7 +2441,7 @@ mod tests {
                 _,
             ) => {
                 assert_eq!(schema.table_id, t_id);
-                assert_eq!(arg, Sarg::Eq(ColId(1), AlgebraicValue::U8(3)));
+                assert_eq!(arg, Sarg::Eq(ColId(1), AlgebraicValue::U8(3).into()));
                 assert!(prefix.is_empty());
             }
             plan => panic!("unexpected plan: {plan:#?}"),
@@ -2168,7 +2469,7 @@ mod tests {
                 _,
             ) => {
                 assert_eq!(schema.table_id, t_id);
-                assert_eq!(arg, Sarg::Eq(ColId(1), AlgebraicValue::U8(4)));
+                assert_eq!(arg, Sarg::Eq(ColId(1), AlgebraicValue::U8(4).into()));
                 assert!(prefix.is_empty());
             }
             plan => panic!("unexpected plan: {plan:#?}"),
@@ -2201,7 +2502,7 @@ mod tests {
                 _,
             )) => {
                 assert_eq!(schema.table_id, t_id);
-                assert_eq!(arg, Sarg::Eq(ColId(3), AlgebraicValue::U8(2)));
+                assert_eq!(arg, Sarg::Eq(ColId(3), AlgebraicValue::U8(2).into()));
                 assert_eq!(prefix, vec![(ColId(2), AlgebraicValue::U8(1))]);
             }
             proj => panic!("unexpected plan: {proj:#?}"),
@@ -2220,7 +2521,7 @@ mod tests {
                 _,
             )) => {
                 assert_eq!(schema.table_id, t_id);
-                assert_eq!(arg, Sarg::Eq(ColId(3), AlgebraicValue::U8(2)));
+                assert_eq!(arg, Sarg::Eq(ColId(3), AlgebraicValue::U8(2).into()));
                 assert_eq!(prefix, vec![(ColId(2), AlgebraicValue::U8(1))]);
             }
             proj => panic!("unexpected plan: {proj:#?}"),
@@ -2248,7 +2549,7 @@ mod tests {
                 _,
             ) => {
                 assert_eq!(schema.table_id, t_id);
-                assert_eq!(arg, Sarg::Eq(ColId(3), AlgebraicValue::U8(3)));
+                assert_eq!(arg, Sarg::Eq(ColId(3), AlgebraicValue::U8(3).into()));
                 assert_eq!(prefix, vec![(ColId(2), AlgebraicValue::U8(2))]);
             }
             plan => panic!("unexpected plan: {plan:#?}"),
diff --git a/crates/physical-plan/src/rules.rs b/crates/physical-plan/src/rules.rs
index c553671b6f1..bfcd7e42548 100644
--- a/crates/physical-plan/src/rules.rs
+++ b/crates/physical-plan/src/rules.rs
@@ -274,7 +274,8 @@ impl RewriteRule for PushConstEq {
             !plan.any(&|plan| !matches!(plan, PhysicalPlan::TableScan(..) | PhysicalPlan::Filter(..)))
         };
         if let PhysicalPlan::Filter(input, PhysicalExpr::BinOp(_, expr, value)) = plan
-            && let (PhysicalExpr::Field(TupleField { label, .. }), PhysicalExpr::Value(_)) = (&**expr, &**value)
+            && let PhysicalExpr::Field(TupleField { label, .. }) = &**expr
+            && value.is_sarg_value()
         {
             return (input.has_table_scan(Some(label)) && !is_filter(input)).then_some(*label);
         }
@@ -342,7 +343,8 @@ impl RewriteRule for PushConstAnd {
         if let PhysicalPlan::Filter(input, PhysicalExpr::LogOp(LogOp::And, exprs)) = plan {
             return exprs.iter().find_map(|expr| {
                 if let PhysicalExpr::BinOp(_, expr, value) = expr
-                    && let (PhysicalExpr::Field(TupleField { label, .. }), PhysicalExpr::Value(_)) = (&**expr, &**value)
+                    && let PhysicalExpr::Field(TupleField { label, .. }) = &**expr
+                    && value.is_sarg_value()
                 {
                     return (input.has_table_scan(Some(label)) && !is_filter(input)).then_some(*label);
                 }
@@ -359,8 +361,8 @@ impl RewriteRule for PushConstAnd {
                 let mut root_exprs = vec![];
                 for expr in exprs {
                     if let PhysicalExpr::BinOp(_, lhs, value) = &expr
-                        && let (PhysicalExpr::Field(TupleField { label: var, .. }), PhysicalExpr::Value(_)) =
-                            (&**lhs, &**value)
+                        && let PhysicalExpr::Field(TupleField { label: var, .. }) = &**lhs
+                        && value.is_sarg_value()
                         && var == &relvar
                     {
                         leaf_exprs.push(expr);
@@ -428,8 +430,8 @@ impl RewriteRule for IxScanEq {
                 },
                 _,
             ) = &**input
-            && let (PhysicalExpr::Field(TupleField { field_pos: pos, .. }), PhysicalExpr::Value(_)) =
-                (&**expr, &**value)
+            && let PhysicalExpr::Field(TupleField { field_pos: pos, .. }) = &**expr
+            && value.is_sarg_value()
         {
             return schema.indexes.iter().find_map(
                 |IndexSchema {
@@ -453,7 +455,7 @@ impl RewriteRule for IxScanEq {
     fn rewrite(plan: PhysicalPlan, (index_id, col_id): Self::Info) -> Result<PhysicalPlan> {
         if let PhysicalPlan::Filter(input, PhysicalExpr::BinOp(BinOp::Eq, _, value)) = plan
             && let PhysicalPlan::TableScan(TableScan { schema, limit, delta }, var) = *input
-            && let PhysicalExpr::Value(v) = *value
+            && let Some(v) = (*value).into_sarg_value()
         {
             return Ok(PhysicalPlan::IxScan(
                 IxScan {
@@ -499,8 +501,8 @@ impl RewriteRule for IxScanAnd {
         {
             return exprs.iter().enumerate().find_map(|(i, expr)| {
                 if let PhysicalExpr::BinOp(BinOp::Eq, lhs, value) = expr
-                    && let (PhysicalExpr::Field(TupleField { field_pos: pos, .. }), PhysicalExpr::Value(_)) =
-                        (&**lhs, &**value)
+                    && let PhysicalExpr::Field(TupleField { field_pos: pos, .. }) = &**lhs
+                    && value.is_sarg_value()
                 {
                     return schema.indexes.iter().find_map(
                         |IndexSchema {
@@ -527,7 +529,7 @@ impl RewriteRule for IxScanAnd {
         if let PhysicalPlan::Filter(input, PhysicalExpr::LogOp(LogOp::And, mut exprs)) = plan
             && let PhysicalPlan::TableScan(TableScan { schema, limit, delta }, label) = *input
             && let PhysicalExpr::BinOp(BinOp::Eq, _, value) = exprs.swap_remove(i)
-            && let PhysicalExpr::Value(v) = *value
+            && let Some(v) = (*value).into_sarg_value()
         {
             return Ok(PhysicalPlan::Filter(
                 Box::new(PhysicalPlan::IxScan(
@@ -646,7 +648,7 @@ impl RewriteRule for IxScanEq2Col {
                                 delta,
                                 index_id: info.index_id,
                                 prefix: vec![(*a, u.clone())],
-                                arg: Sarg::Eq(*b, v.clone()),
+                                arg: Sarg::Eq(*b, v.clone().into()),
                             },
                             label,
                         ),
@@ -663,7 +665,7 @@ impl RewriteRule for IxScanEq2Col {
                                     delta,
                                     index_id: info.index_id,
                                     prefix: vec![(*a, u.clone())],
-                                    arg: Sarg::Eq(*b, v.clone()),
+                                    arg: Sarg::Eq(*b, v.clone().into()),
                                 },
                                 label,
                             )),
@@ -685,7 +687,7 @@ impl RewriteRule for IxScanEq2Col {
                                     delta,
                                     index_id: info.index_id,
                                     prefix: vec![(*a, u.clone())],
-                                    arg: Sarg::Eq(*b, v.clone()),
+                                    arg: Sarg::Eq(*b, v.clone().into()),
                                 },
                                 label,
                             )),
@@ -824,7 +826,7 @@ impl RewriteRule for IxScanEq3Col {
                                 delta,
                                 index_id: info.index_id,
                                 prefix: vec![(*a, u.clone()), (*b, v.clone())],
-                                arg: Sarg::Eq(*c, w.clone()),
+                                arg: Sarg::Eq(*c, w.clone().into()),
                             },
                             label,
                         ),
@@ -841,7 +843,7 @@ impl RewriteRule for IxScanEq3Col {
                                     delta,
                                     index_id: info.index_id,
                                     prefix: vec![(*a, u.clone()), (*b, v.clone())],
-                                    arg: Sarg::Eq(*c, w.clone()),
+                                    arg: Sarg::Eq(*c, w.clone().into()),
                                 },
                                 label,
                             )),
@@ -863,7 +865,7 @@ impl RewriteRule for IxScanEq3Col {
                                     delta,
                                     index_id: info.index_id,
                                     prefix: vec![(*a, u.clone()), (*b, v.clone())],
-                                    arg: Sarg::Eq(*c, w.clone()),
+                                    arg: Sarg::Eq(*c, w.clone().into()),
                                 },
                                 label,
                             )),
@@ -1195,7 +1197,7 @@ fn rhs_eq_constants_from_ixscan(scan: &IxScan) -> Option<Vec<EqConstFilterTerm>>
     let Sarg::Eq(col, value) = &scan.arg else {
         return None;
     };
-    constants.push((*col, value.clone()));
+    constants.push((*col, value.as_literal()?.clone()));
     Some(constants)
 }
 
diff --git a/crates/query/src/lib.rs b/crates/query/src/lib.rs
index 4a1bea08c9c..6bbf1796a29 100644
--- a/crates/query/src/lib.rs
+++ b/crates/query/src/lib.rs
@@ -6,7 +6,7 @@ use spacetimedb_execution::{
 };
 use spacetimedb_expr::{
     check::{parse_and_type_sub, SchemaView},
-    expr::ProjectList,
+    expr::{BindEnv, ProjectList},
     rls::{resolve_views_for_sql, resolve_views_for_sub},
     statement::{parse_and_type_sql, Statement, DML},
 };
@@ -76,7 +76,8 @@ pub fn execute_select_stmt<Tx: Datastore + DeltaStore>(
     metrics: &mut ExecutionMetrics,
     check_row_limit: impl Fn(ProjectListPlan) -> Result<ProjectListPlan>,
 ) -> Result<Vec<ProductValue>> {
-    let plan = compile_select_list(stmt).optimize(auth)?;
+    let bind_env = BindEnv::sender(auth.caller());
+    let plan = compile_select_list(stmt).optimize(auth)?.bind_params(&bind_env);
     let plan = check_row_limit(plan)?;
     let plan = ProjectListExecutor::from(plan);
     let mut rows = vec![];
@@ -94,7 +95,8 @@ pub fn execute_dml_stmt<Tx: MutDatastore>(
     tx: &mut Tx,
     metrics: &mut ExecutionMetrics,
 ) -> Result<()> {
-    let plan = compile_dml_plan(stmt).optimize(auth)?;
+    let bind_env = BindEnv::sender(auth.caller());
+    let plan = compile_dml_plan(stmt).optimize(auth)?.bind_params(&bind_env);
     let plan = MutExecutor::from(plan);
     plan.execute(tx, metrics)
 }
diff --git a/crates/subscription/src/lib.rs b/crates/subscription/src/lib.rs
index 00ac42ec1c9..8c3bc4006a8 100644
--- a/crates/subscription/src/lib.rs
+++ b/crates/subscription/src/lib.rs
@@ -1,13 +1,10 @@
 use anyhow::{bail, Result};
 use spacetimedb_data_structures::map::{HashCollectionExt as _, HashSet};
-use spacetimedb_execution::{
-    pipelined::{
-        PipelinedExecutor, PipelinedIxDeltaJoin, PipelinedIxDeltaScanEq, PipelinedIxDeltaScanRange, PipelinedIxJoin,
-        PipelinedIxScanEq, PipelinedIxScanRange, PipelinedProject,
-    },
-    Datastore, DeltaStore, Row,
+use spacetimedb_execution::{pipelined::PipelinedProject, Datastore, DeltaStore, Row};
+use spacetimedb_expr::{
+    check::SchemaView,
+    expr::{BindEnv, CollectViews, ParamId},
 };
-use spacetimedb_expr::{check::SchemaView, expr::CollectViews};
 use spacetimedb_lib::{identity::AuthCtx, metrics::ExecutionMetrics, query::Delta, AlgebraicValue};
 use spacetimedb_physical_plan::plan::{IxJoin, IxScan, Label, PhysicalPlan, ProjectPlan, Sarg, TableScan, TupleField};
 use spacetimedb_primitives::{ColId, ColList, IndexId, TableId, ViewId};
@@ -19,39 +16,46 @@ use std::ops::RangeBounds;
 /// How do we incrementally maintain that view?
 /// These are the query fragments that are required.
 /// See [Self::compile_from_plan] for how to generate them.
+///
+/// Fragments stay as physical plans because they can contain runtime params.
+/// Each subscription instance binds them before constructing pipelined executors.
 #[derive(Debug)]
 struct Fragments {
     /// Plan fragments that return rows to insert.
     /// For joins there will be 4 fragments,
     /// but for selects only one.
-    insert_plans: Vec<PipelinedProject>,
+    insert_plans: Vec<ProjectPlan>,
     /// Plan fragments that return rows to delete.
     /// For joins there will be 4 fragments,
     /// but for selects only one.
-    delete_plans: Vec<PipelinedProject>,
+    delete_plans: Vec<ProjectPlan>,
 }
 
 impl Fragments {
+    fn requires_param(&self, id: ParamId) -> bool {
+        self.insert_plans
+            .iter()
+            .chain(&self.delete_plans)
+            .any(|plan| plan.requires_param(id))
+    }
+
     /// Returns the index ids from which this fragment reads.
     fn index_ids(&self) -> impl Iterator<Item = (TableId, IndexId)> + use<> {
         let mut index_ids = HashSet::new();
         for plan in self.insert_plans.iter().chain(self.delete_plans.iter()) {
             plan.visit(&mut |plan| match plan {
-                PipelinedExecutor::IxScanEq(PipelinedIxScanEq { table_id, index_id, .. })
-                | PipelinedExecutor::IxScanRange(PipelinedIxScanRange { table_id, index_id, .. })
-                | PipelinedExecutor::IxDeltaScanEq(PipelinedIxDeltaScanEq { table_id, index_id, .. })
-                | PipelinedExecutor::IxDeltaScanRange(PipelinedIxDeltaScanRange { table_id, index_id, .. })
-                | PipelinedExecutor::IxJoin(PipelinedIxJoin {
-                    rhs_table: table_id,
-                    rhs_index: index_id,
-                    ..
-                })
-                | PipelinedExecutor::IxDeltaJoin(PipelinedIxDeltaJoin {
-                    rhs_table: table_id,
-                    rhs_index: index_id,
-                    ..
-                }) => {
-                    index_ids.insert((*table_id, *index_id));
+                PhysicalPlan::IxScan(IxScan { schema, index_id, .. }, _) => {
+                    index_ids.insert((schema.table_id, *index_id));
+                }
+                PhysicalPlan::IxJoin(
+                    IxJoin {
+                        rhs,
+                        rhs_index: index_id,
+                        ..
+                    },
+                    _,
+                ) => {
+                    index_ids.insert((rhs.table_id, *index_id));
                 }
                 _ => {}
             });
@@ -64,11 +68,16 @@ impl Fragments {
     /// and evaluate a closure over each one.
     fn for_each_insert<'a, Tx: Datastore + DeltaStore>(
         &self,
+        bind_env: &BindEnv,
         tx: &'a Tx,
         metrics: &mut ExecutionMetrics,
         f: &mut dyn FnMut(Row<'a>) -> Result<()>,
     ) -> Result<()> {
         for plan in &self.insert_plans {
+            // TODO: This rebuilds the executor for every delta evaluation. That is the
+            // temporary cost of keeping parameterized fragments as physical-plan templates;
+            // a later pass should bind/cache executable fragments per subscription instance.
+            let plan = PipelinedProject::from(plan.clone().bind_params(bind_env));
             if !plan.is_empty(tx) {
                 plan.execute(tx, metrics, f)?;
             }
@@ -81,11 +90,16 @@ impl Fragments {
     /// and evaluate a closure over each one.
     fn for_each_delete<'a, Tx: Datastore + DeltaStore>(
         &self,
+        bind_env: &BindEnv,
         tx: &'a Tx,
         metrics: &mut ExecutionMetrics,
         f: &mut dyn FnMut(Row<'a>) -> Result<()>,
     ) -> Result<()> {
         for plan in &self.delete_plans {
+            // TODO: This rebuilds the executor for every delta evaluation. That is the
+            // temporary cost of keeping parameterized fragments as physical-plan templates;
+            // a later pass should bind/cache executable fragments per subscription instance.
+            let plan = PipelinedProject::from(plan.clone().bind_params(bind_env));
             if !plan.is_empty(tx) {
                 plan.execute(tx, metrics, f)?;
             }
@@ -181,12 +195,12 @@ impl Fragments {
         }
 
         /// Return a new plan with delta scans for the given tables
-        fn new_plan(plan: &ProjectPlan, tables: &[(Label, Delta)], auth: &AuthCtx) -> Result<PipelinedProject> {
+        fn new_plan(plan: &ProjectPlan, tables: &[(Label, Delta)], auth: &AuthCtx) -> Result<ProjectPlan> {
             let mut plan = plan.clone();
             for (alias, delta) in tables {
                 mut_plan(&mut plan, *alias, *delta);
             }
-            plan.optimize(auth).map(PipelinedProject::from)
+            plan.optimize(auth)
         }
 
         match tables {
@@ -383,6 +397,19 @@ impl SubscriptionPlan {
         &self.plan_opt
     }
 
+    /// The optimized plan with this subscription instance's runtime bindings applied.
+    ///
+    /// TODO: Prefer binding directly into an executable form so callers do not observe an
+    /// intermediate physical plan that happens to have all params replaced with literals.
+    pub fn bound_optimized_physical_plan(&self, bind_env: &BindEnv) -> ProjectPlan {
+        self.optimized_physical_plan().clone().bind_params(bind_env)
+    }
+
+    /// Returns whether this subscription requires a runtime parameter binding.
+    pub fn requires_param(&self, id: ParamId) -> bool {
+        self.plan_opt.requires_param(id) || self.fragments.requires_param(id)
+    }
+
     /// From which indexes does this plan read?
     pub fn index_ids(&self) -> impl Iterator<Item = (TableId, IndexId)> + use<> {
         self.fragments.index_ids()
@@ -393,11 +420,12 @@ impl SubscriptionPlan {
     /// and evaluate a closure over each one.
     pub fn for_each_insert<'a, Tx: Datastore + DeltaStore>(
         &self,
+        bind_env: &BindEnv,
         tx: &'a Tx,
         metrics: &mut ExecutionMetrics,
         f: &mut dyn FnMut(Row<'a>) -> Result<()>,
     ) -> Result<()> {
-        self.fragments.for_each_insert(tx, metrics, f)
+        self.fragments.for_each_insert(bind_env, tx, metrics, f)
     }
 
     /// A subscription is just a view of a particular table.
@@ -405,11 +433,12 @@ impl SubscriptionPlan {
     /// and evaluate a closure over each one.
     pub fn for_each_delete<'a, Tx: Datastore + DeltaStore>(
         &self,
+        bind_env: &BindEnv,
         tx: &'a Tx,
         metrics: &mut ExecutionMetrics,
         f: &mut dyn FnMut(Row<'a>) -> Result<()>,
     ) -> Result<()> {
-        self.fragments.for_each_delete(tx, metrics, f)
+        self.fragments.for_each_delete(bind_env, tx, metrics, f)
     }
 
     /// Returns a join edge for this query if it has one.
@@ -419,11 +448,24 @@ impl SubscriptionPlan {
     /// 2. Single column index lookup on the rhs table
     /// 3. No self joins
     pub fn join_edge(&self) -> Option<(JoinEdge, AlgebraicValue)> {
+        self.join_edge_in_plan(&self.plan_opt)
+    }
+
+    /// Returns a join edge for this query after applying runtime bindings.
+    ///
+    /// Subscription pruning indexes store concrete values, so param RHS values like `:sender`
+    /// must be bound before indexing a query instance.
+    pub fn bound_join_edge(&self, bind_env: &BindEnv) -> Option<(JoinEdge, AlgebraicValue)> {
+        let plan = self.bound_optimized_physical_plan(bind_env);
+        self.join_edge_in_plan(&plan)
+    }
+
+    fn join_edge_in_plan(&self, plan: &ProjectPlan) -> Option<(JoinEdge, AlgebraicValue)> {
         if !self.is_join() {
             return None;
         }
         let mut join_edge = None;
-        self.plan_opt.visit(&mut |op| match op {
+        plan.visit(&mut |op| match op {
             PhysicalPlan::IxJoin(
                 IxJoin {
                     lhs,
@@ -453,7 +495,9 @@ impl SubscriptionPlan {
                     let lhs_table = self.return_id;
                     let rhs_table = schema.table_id;
                     let rhs_col = *rhs_col;
-                    let rhs_val = rhs_val.clone();
+                    let Some(rhs_val) = rhs_val.as_literal().cloned() else {
+                        return;
+                    };
                     let lhs_join_col = *lhs_join_col;
                     let rhs_join_col = (*rhs_join_col).into();
                     let edge = JoinEdge {

From c72b1a875435e1c9d20fddd52c9a9669b9054f5a Mon Sep 17 00:00:00 2001
From: joshua-spacetime <josh@clockworklabs.io>
Date: Tue, 9 Jun 2026 15:05:26 -0700
Subject: [PATCH 2/2] multi-col index params

---
 crates/execution/src/pipelined.rs | 16 +++----
 crates/physical-plan/src/plan.rs  | 54 +++++++++++++++++++-----
 crates/physical-plan/src/rules.rs | 70 +++++++++++++++++--------------
 3 files changed, 89 insertions(+), 51 deletions(-)

diff --git a/crates/execution/src/pipelined.rs b/crates/execution/src/pipelined.rs
index 2abd30c94a5..db97a5ec42d 100644
--- a/crates/execution/src/pipelined.rs
+++ b/crates/execution/src/pipelined.rs
@@ -362,7 +362,7 @@ impl From<PhysicalPlan> for PipelinedExecutor {
                 lhs: Box::new(Self::from(*lhs)),
                 rhs_table: rhs.table_id,
                 rhs_index,
-                rhs_prefix,
+                rhs_prefix: rhs_prefix.into_iter().map(expect_bound_sarg_value).collect(),
                 rhs_field,
                 lhs_field,
                 unique,
@@ -385,7 +385,7 @@ impl From<PhysicalPlan> for PipelinedExecutor {
                 lhs: Box::new(Self::from(*lhs)),
                 rhs_table: rhs.table_id,
                 rhs_index,
-                rhs_prefix,
+                rhs_prefix: rhs_prefix.into_iter().map(expect_bound_sarg_value).collect(),
                 rhs_field,
                 rhs_delta,
                 lhs_field,
@@ -593,7 +593,7 @@ impl From<IxScan> for PipelinedIxDeltaScanRange {
                 Self {
                     table_id: schema.table_id,
                     index_id,
-                    prefix: prefix.into_iter().map(|(_, v)| v).collect(),
+                    prefix: prefix.into_iter().map(|(_, v)| expect_bound_sarg_value(v)).collect(),
                     lower: Bound::Included(v.clone()),
                     upper: Bound::Included(v),
                     delta,
@@ -609,7 +609,7 @@ impl From<IxScan> for PipelinedIxDeltaScanRange {
             } => Self {
                 table_id: schema.table_id,
                 index_id,
-                prefix: prefix.into_iter().map(|(_, v)| v).collect(),
+                prefix: prefix.into_iter().map(|(_, v)| expect_bound_sarg_value(v)).collect(),
                 lower,
                 upper,
                 delta,
@@ -791,7 +791,7 @@ impl From<IxScan> for PipelinedIxScanRange {
                     table_id: schema.table_id,
                     index_id,
                     limit,
-                    prefix: prefix.into_iter().map(|(_, v)| v).collect(),
+                    prefix: prefix.into_iter().map(|(_, v)| expect_bound_sarg_value(v)).collect(),
                     lower: Bound::Included(v.clone()),
                     upper: Bound::Included(v),
                 }
@@ -807,7 +807,7 @@ impl From<IxScan> for PipelinedIxScanRange {
                 table_id: schema.table_id,
                 index_id,
                 limit,
-                prefix: prefix.into_iter().map(|(_, v)| v).collect(),
+                prefix: prefix.into_iter().map(|(_, v)| expect_bound_sarg_value(v)).collect(),
                 lower,
                 upper,
             },
@@ -931,12 +931,12 @@ impl From<IxScan> for PipelinedIxScanEq {
     }
 }
 
-fn combine_prefix_and_last(prefix: Vec<(ColId, AlgebraicValue)>, last: AlgebraicValue) -> AlgebraicValue {
+fn combine_prefix_and_last(prefix: Vec<(ColId, SargValue)>, last: AlgebraicValue) -> AlgebraicValue {
     if prefix.is_empty() {
         last
     } else {
         let mut elems = Vec::with_capacity(prefix.len() + 1);
-        elems.extend(prefix.into_iter().map(|(_, v)| v));
+        elems.extend(prefix.into_iter().map(|(_, v)| expect_bound_sarg_value(v)));
         elems.push(last);
         AlgebraicValue::product(elems)
     }
diff --git a/crates/physical-plan/src/plan.rs b/crates/physical-plan/src/plan.rs
index c273924a7ed..4450ff91085 100644
--- a/crates/physical-plan/src/plan.rs
+++ b/crates/physical-plan/src/plan.rs
@@ -381,9 +381,12 @@ impl PhysicalPlan {
     /// Returns whether this plan contains a runtime parameter.
     pub fn requires_param(&self, id: ParamId) -> bool {
         self.any(&|plan| match plan {
-            Self::IxScan(scan, _) => scan.arg.requires_param(id),
+            Self::IxScan(scan, _) => {
+                scan.arg.requires_param(id) || scan.prefix.iter().any(|(_, value)| value.requires_param(id))
+            }
+            Self::IxJoin(join, _) => join.rhs_prefix.iter().any(|value| value.requires_param(id)),
             Self::Filter(_, expr) => expr.requires_param(id),
-            Self::TableScan(..) | Self::IxJoin(..) | Self::HashJoin(..) | Self::NLJoin(..) => false,
+            Self::TableScan(..) | Self::HashJoin(..) | Self::NLJoin(..) => false,
         })
     }
 
@@ -416,6 +419,11 @@ impl PhysicalPlan {
         match self {
             Self::TableScan(..) => self,
             Self::IxScan(mut scan, label) => {
+                scan.prefix = scan
+                    .prefix
+                    .into_iter()
+                    .map(|(col, value)| (col, value.bind_params(bind_env)))
+                    .collect();
                 scan.arg = scan.arg.bind_params(bind_env);
                 Self::IxScan(scan, label)
             }
@@ -438,7 +446,10 @@ impl PhysicalPlan {
                     rhs,
                     rhs_label,
                     rhs_index,
-                    rhs_prefix,
+                    rhs_prefix: rhs_prefix
+                        .into_iter()
+                        .map(|value| value.bind_params(bind_env))
+                        .collect(),
                     rhs_field,
                     unique,
                     lhs_field,
@@ -1317,8 +1328,11 @@ pub struct IxScan {
     pub delta: Option<Delta>,
     /// The index id
     pub index_id: IndexId,
-    /// An equality prefix for multi-column scans
-    pub prefix: Vec<(ColId, AlgebraicValue)>,
+    /// An equality prefix for multi-column scans.
+    ///
+    /// Prefix values may be runtime params while this is still an optimizer/template plan.
+    /// They must be bound before executor construction.
+    pub prefix: Vec<(ColId, SargValue)>,
     /// The index argument
     pub arg: Sarg,
 }
@@ -1426,7 +1440,10 @@ pub struct IxJoin {
     /// The index id
     pub rhs_index: IndexId,
     /// Optional constant prefix values for multi-column index probes.
-    pub rhs_prefix: Vec<AlgebraicValue>,
+    ///
+    /// Prefix values may be runtime params while this is still an optimizer/template plan.
+    /// They must be bound before executor construction.
+    pub rhs_prefix: Vec<SargValue>,
     /// The index field
     pub rhs_field: ColId,
     /// Is the index a unique constraint index?
@@ -1497,6 +1514,15 @@ impl PhysicalExpr {
         }
     }
 
+    /// Converts a literal or runtime parameter expression into an index search value.
+    pub fn as_sarg_value(&self) -> Option<SargValue> {
+        match self {
+            Self::Value(value) => Some(SargValue::Literal(value.clone())),
+            Self::Param(id, ty) => Some(SargValue::Param(*id, ty.clone())),
+            _ => None,
+        }
+    }
+
     pub fn as_literal(&self) -> Option<&AlgebraicValue> {
         match self {
             Self::Value(value) => Some(value),
@@ -2391,7 +2417,10 @@ mod tests {
                 assert_eq!(arg, Sarg::Eq(ColId(3), AlgebraicValue::U8(5).into()));
                 assert_eq!(
                     prefix,
-                    vec![(ColId(1), AlgebraicValue::U8(3)), (ColId(2), AlgebraicValue::U8(4))]
+                    vec![
+                        (ColId(1), AlgebraicValue::U8(3).into()),
+                        (ColId(2), AlgebraicValue::U8(4).into())
+                    ]
                 );
             }
             proj => panic!("unexpected plan: {proj:#?}"),
@@ -2413,7 +2442,10 @@ mod tests {
                 assert_eq!(arg, Sarg::Eq(ColId(3), AlgebraicValue::U8(5).into()));
                 assert_eq!(
                     prefix,
-                    vec![(ColId(1), AlgebraicValue::U8(3)), (ColId(2), AlgebraicValue::U8(4))]
+                    vec![
+                        (ColId(1), AlgebraicValue::U8(3).into()),
+                        (ColId(2), AlgebraicValue::U8(4).into())
+                    ]
                 );
             }
             proj => panic!("unexpected plan: {proj:#?}"),
@@ -2503,7 +2535,7 @@ mod tests {
             )) => {
                 assert_eq!(schema.table_id, t_id);
                 assert_eq!(arg, Sarg::Eq(ColId(3), AlgebraicValue::U8(2).into()));
-                assert_eq!(prefix, vec![(ColId(2), AlgebraicValue::U8(1))]);
+                assert_eq!(prefix, vec![(ColId(2), AlgebraicValue::U8(1).into())]);
             }
             proj => panic!("unexpected plan: {proj:#?}"),
         };
@@ -2522,7 +2554,7 @@ mod tests {
             )) => {
                 assert_eq!(schema.table_id, t_id);
                 assert_eq!(arg, Sarg::Eq(ColId(3), AlgebraicValue::U8(2).into()));
-                assert_eq!(prefix, vec![(ColId(2), AlgebraicValue::U8(1))]);
+                assert_eq!(prefix, vec![(ColId(2), AlgebraicValue::U8(1).into())]);
             }
             proj => panic!("unexpected plan: {proj:#?}"),
         };
@@ -2550,7 +2582,7 @@ mod tests {
             ) => {
                 assert_eq!(schema.table_id, t_id);
                 assert_eq!(arg, Sarg::Eq(ColId(3), AlgebraicValue::U8(3).into()));
-                assert_eq!(prefix, vec![(ColId(2), AlgebraicValue::U8(2))]);
+                assert_eq!(prefix, vec![(ColId(2), AlgebraicValue::U8(2).into())]);
             }
             plan => panic!("unexpected plan: {plan:#?}"),
         };
diff --git a/crates/physical-plan/src/rules.rs b/crates/physical-plan/src/rules.rs
index bfcd7e42548..e1f7420bc4d 100644
--- a/crates/physical-plan/src/rules.rs
+++ b/crates/physical-plan/src/rules.rs
@@ -27,14 +27,13 @@
 //! * [UniqueHashJoinRule]
 //!   Mark hash join as unique
 use anyhow::{bail, Result};
-use spacetimedb_lib::AlgebraicValue;
 use spacetimedb_primitives::{ColId, ColSet, IndexId};
 use spacetimedb_schema::schema::{IndexSchema, TableSchema};
 use spacetimedb_sql_parser::ast::{BinOp, LogOp};
 
 use crate::plan::{
-    HashJoin, IxJoin, IxScan, Label, PhysicalExpr, PhysicalPlan, ProjectListPlan, ProjectPlan, Sarg, Semi, TableScan,
-    TupleField,
+    HashJoin, IxJoin, IxScan, Label, PhysicalExpr, PhysicalPlan, ProjectListPlan, ProjectPlan, Sarg, SargValue, Semi,
+    TableScan, TupleField,
 };
 
 /// A rewrite will only fail due to an internal logic bug.
@@ -591,11 +590,13 @@ impl RewriteRule for IxScanEq2Col {
                     continue;
                 };
 
-                let (PhysicalExpr::Field(u), PhysicalExpr::Value(_), PhysicalExpr::Field(v), PhysicalExpr::Value(_)) =
-                    (&**a, &**u, &**b, &**v)
+                let (PhysicalExpr::Field(u), value_u, PhysicalExpr::Field(v), value_v) = (&**a, &**u, &**b, &**v)
                 else {
                     continue;
                 };
+                if !value_u.is_sarg_value() || !value_v.is_sarg_value() {
+                    continue;
+                }
 
                 if let Some(scan) = schema
                     .indexes
@@ -633,7 +634,7 @@ impl RewriteRule for IxScanEq2Col {
                     && let PhysicalPlan::TableScan(TableScan { schema, limit, delta }, label) = *input
                     && let (Some(PhysicalExpr::BinOp(BinOp::Eq, _, u)), Some(PhysicalExpr::BinOp(BinOp::Eq, _, v))) =
                         (exprs.get(*i), exprs.get(*j))
-                    && let (PhysicalExpr::Value(u), PhysicalExpr::Value(v)) = (&**u, &**v)
+                    && let (Some(u), Some(v)) = (u.as_sarg_value(), v.as_sarg_value())
                 {
                     return Ok(match exprs.len() {
                         n @ 0 | n @ 1 => {
@@ -648,7 +649,7 @@ impl RewriteRule for IxScanEq2Col {
                                 delta,
                                 index_id: info.index_id,
                                 prefix: vec![(*a, u.clone())],
-                                arg: Sarg::Eq(*b, v.clone().into()),
+                                arg: Sarg::Eq(*b, v.clone()),
                             },
                             label,
                         ),
@@ -665,7 +666,7 @@ impl RewriteRule for IxScanEq2Col {
                                     delta,
                                     index_id: info.index_id,
                                     prefix: vec![(*a, u.clone())],
-                                    arg: Sarg::Eq(*b, v.clone().into()),
+                                    arg: Sarg::Eq(*b, v.clone()),
                                 },
                                 label,
                             )),
@@ -687,7 +688,7 @@ impl RewriteRule for IxScanEq2Col {
                                     delta,
                                     index_id: info.index_id,
                                     prefix: vec![(*a, u.clone())],
-                                    arg: Sarg::Eq(*b, v.clone().into()),
+                                    arg: Sarg::Eq(*b, v.clone()),
                                 },
                                 label,
                             )),
@@ -757,15 +758,18 @@ impl RewriteRule for IxScanEq3Col {
 
                     let (
                         PhysicalExpr::Field(u),
-                        PhysicalExpr::Value(_),
+                        value_u,
                         PhysicalExpr::Field(v),
-                        PhysicalExpr::Value(_),
+                        value_v,
                         PhysicalExpr::Field(w),
-                        PhysicalExpr::Value(_),
+                        value_w,
                     ) = (&**a, &**u, &**b, &**v, &**c, &**w)
                     else {
                         continue;
                     };
+                    if !value_u.is_sarg_value() || !value_v.is_sarg_value() || !value_w.is_sarg_value() {
+                        continue;
+                    }
 
                     if let Some(scan) = schema
                         .indexes
@@ -811,7 +815,7 @@ impl RewriteRule for IxScanEq3Col {
                         Some(PhysicalExpr::BinOp(BinOp::Eq, _, v)),
                         Some(PhysicalExpr::BinOp(BinOp::Eq, _, w)),
                     ) = (exprs.get(*i), exprs.get(*j), exprs.get(*k))
-                    && let (PhysicalExpr::Value(u), PhysicalExpr::Value(v), PhysicalExpr::Value(w)) = (&**u, &**v, &**w)
+                    && let (Some(u), Some(v), Some(w)) = (u.as_sarg_value(), v.as_sarg_value(), w.as_sarg_value())
                 {
                     return Ok(match exprs.len() {
                         n @ 0 | n @ 1 | n @ 2 => {
@@ -826,7 +830,7 @@ impl RewriteRule for IxScanEq3Col {
                                 delta,
                                 index_id: info.index_id,
                                 prefix: vec![(*a, u.clone()), (*b, v.clone())],
-                                arg: Sarg::Eq(*c, w.clone().into()),
+                                arg: Sarg::Eq(*c, w.clone()),
                             },
                             label,
                         ),
@@ -843,7 +847,7 @@ impl RewriteRule for IxScanEq3Col {
                                     delta,
                                     index_id: info.index_id,
                                     prefix: vec![(*a, u.clone()), (*b, v.clone())],
-                                    arg: Sarg::Eq(*c, w.clone().into()),
+                                    arg: Sarg::Eq(*c, w.clone()),
                                 },
                                 label,
                             )),
@@ -865,7 +869,7 @@ impl RewriteRule for IxScanEq3Col {
                                     delta,
                                     index_id: info.index_id,
                                     prefix: vec![(*a, u.clone()), (*b, v.clone())],
-                                    arg: Sarg::Eq(*c, w.clone().into()),
+                                    arg: Sarg::Eq(*c, w.clone()),
                                 },
                                 label,
                             )),
@@ -1064,8 +1068,8 @@ impl RewriteRule for PullFilterAboveHashJoin {
 /// Always prefer an index join to a hash join
 pub(crate) struct HashToIxJoin;
 
-/// A filter term of the form `rhs_col = constant`.
-type EqConstFilterTerm = (ColId, AlgebraicValue);
+/// A filter term of the form `rhs_col = constant_or_runtime_param`.
+type EqConstFilterTerm = (ColId, SargValue);
 
 /// Planning metadata derived while proving `HashJoin -> IxJoin` is valid.
 #[derive(Clone)]
@@ -1077,27 +1081,26 @@ pub(crate) struct HashToIxJoinInfo {
     rhs_field: ColId,
     /// Constant leading key components for a multi-column index probe.
     /// For an index `(a, b, c)` and join on `c`, this stores `[a_const, b_const]`.
-    rhs_prefix: Vec<AlgebraicValue>,
+    /// Runtime params are allowed here while this is still an optimizer/template plan.
+    rhs_prefix: Vec<SargValue>,
     /// RHS filter terms consumed into `rhs_prefix` and therefore removable
     /// from the residual filter.
     consumed_filter_terms: Vec<EqConstFilterTerm>,
 }
 
-/// Collect RHS predicates of the form `rhs.col = constant`.
+/// Collect RHS predicates of the form `rhs.col = constant_or_runtime_param`.
 ///
 /// This is intentionally narrow: only equality predicates over RHS fields are
 /// useful for building exact prefix probes into multi-column indexes.
 fn rhs_eq_constants(expr: &PhysicalExpr, rhs_label: Label) -> Vec<EqConstFilterTerm> {
     match expr {
         PhysicalExpr::BinOp(BinOp::Eq, lhs, rhs)
-            if matches!(
-                (&**lhs, &**rhs),
-                (PhysicalExpr::Field(TupleField { label, .. }), PhysicalExpr::Value(_)) if *label == rhs_label
-            ) =>
+            if matches!(&**lhs, PhysicalExpr::Field(TupleField { label, .. }) if *label == rhs_label)
+                && rhs.is_sarg_value() =>
         {
-            match (&**lhs, &**rhs) {
-                (PhysicalExpr::Field(TupleField { field_pos, .. }), PhysicalExpr::Value(value)) => {
-                    vec![(ColId(*field_pos as u16), value.clone())]
+            match (&**lhs, rhs.as_sarg_value()) {
+                (PhysicalExpr::Field(TupleField { field_pos, .. }), Some(value)) => {
+                    vec![(ColId(*field_pos as u16), value)]
                 }
                 _ => vec![],
             }
@@ -1166,7 +1169,7 @@ fn ix_join_candidate(
             return None;
         };
 
-        let mut rhs_prefix: Vec<AlgebraicValue> = vec![];
+        let mut rhs_prefix: Vec<SargValue> = vec![];
         let mut consumed_filter_terms: Vec<EqConstFilterTerm> = vec![];
         for col in cols.iter().take(cols.len().saturating_sub(1).into()) {
             let (_, value) = constants.iter().find(|(candidate, _)| *candidate == col)?;
@@ -1197,7 +1200,7 @@ fn rhs_eq_constants_from_ixscan(scan: &IxScan) -> Option<Vec<EqConstFilterTerm>>
     let Sarg::Eq(col, value) = &scan.arg else {
         return None;
     };
-    constants.push((*col, value.as_literal()?.clone()));
+    constants.push((*col, value.clone()));
     Some(constants)
 }
 
@@ -1216,7 +1219,7 @@ fn remove_consumed_filter_terms(
 ) -> Option<PhysicalExpr> {
     // These terms are now represented by `rhs_prefix`; keeping them in a
     // residual filter is redundant work.
-    let is_consumed = |col_id: ColId, value: &AlgebraicValue| {
+    let is_consumed = |col_id: ColId, value: &SargValue| {
         consumed_filter_terms
             .iter()
             .any(|(consumed_col, consumed_val)| consumed_col == &col_id && consumed_val == value)
@@ -1226,8 +1229,11 @@ fn remove_consumed_filter_terms(
         PhysicalExpr::BinOp(BinOp::Eq, lhs, rhs)
             if matches!(
                 (&*lhs, &*rhs),
-                (PhysicalExpr::Field(TupleField { label, field_pos, .. }), PhysicalExpr::Value(value))
-                    if *label == rhs_label && is_consumed(ColId(*field_pos as u16), value)
+                (PhysicalExpr::Field(TupleField { label, field_pos, .. }), value)
+                    if *label == rhs_label
+                        && value
+                            .as_sarg_value()
+                            .is_some_and(|value| is_consumed(ColId(*field_pos as u16), &value))
             ) =>
         {
             None