Call parameterized views from sql #3489, plan and execution

Author: mamcx

crates/core/src/db/relational_db.rs

@@ -1122,6 +1122,10 @@ impl RelationalDB {
         Ok(tx.create_view(module_def, view_def)?)
     }
 
+    pub fn create_or_get_params(&self, tx: &mut MutTx, params: &ProductValue) -> Result<ArgId, DBError> {
+        Ok(tx.create_or_get_params(params)?)
+    }
+
     pub fn drop_view(&self, tx: &mut MutTx, view_id: ViewId) -> Result<(), DBError> {
         Ok(tx.drop_view(view_id)?)
     }
@@ -2217,6 +2221,7 @@ pub mod tests_utils {
         db: &RelationalDB,
         name: &str,
         schema: &[(&str, AlgebraicType)],
+        params: ProductType,
         is_anonymous: bool,
     ) -> Result<(ViewId, TableId), DBError> {
         let mut builder = RawModuleDefV9Builder::new();
@@ -2234,7 +2239,7 @@ pub mod tests_utils {
             0,
             true,
             is_anonymous,
-            ProductType::unit(),
+            params,
             AlgebraicType::array(AlgebraicType::Ref(type_ref)),
         );
 

crates/core/src/sql/ast.rs

@@ -4,11 +4,11 @@ use anyhow::Context;
 use spacetimedb_data_structures::map::{HashCollectionExt as _, IntMap};
 use spacetimedb_datastore::locking_tx_datastore::state_view::StateView;
 use spacetimedb_datastore::system_tables::{StRowLevelSecurityFields, ST_ROW_LEVEL_SECURITY_ID};
-use spacetimedb_expr::check::SchemaView;
+use spacetimedb_expr::check::{SchemaView, TypingResult};
 use spacetimedb_expr::statement::compile_sql_stmt;
 use spacetimedb_lib::identity::AuthCtx;
-use spacetimedb_primitives::{ColId, TableId};
-use spacetimedb_sats::{AlgebraicType, AlgebraicValue};
+use spacetimedb_primitives::{ArgId, ColId, TableId};
+use spacetimedb_sats::{AlgebraicType, AlgebraicValue, ProductValue};
 use spacetimedb_schema::def::error::RelationError;
 use spacetimedb_schema::relation::{ColExpr, FieldName};
 use spacetimedb_schema::schema::{ColumnSchema, TableOrViewSchema, TableSchema};
@@ -477,7 +477,7 @@ fn compile_where(table: &From, filter: Option<SqlExpr>) -> Result<Option<Selecti
 }
 
 pub struct SchemaViewer<'a, T> {
-    tx: &'a T,
+    pub(crate) tx: &'a T,
     auth: &'a AuthCtx,
 }
 

crates/core/src/sql/execute.rs

@@ -1,7 +1,8 @@
+use std::sync::Arc;
 use std::time::Duration;
 
 use super::ast::SchemaViewer;
-use crate::db::relational_db::{RelationalDB, Tx};
+use crate::db::relational_db::{MutTx, RelationalDB, Tx};
 use crate::energy::EnergyQuanta;
 use crate::error::DBError;
 use crate::estimation::estimate_rows_scanned;
@@ -19,13 +20,18 @@ use anyhow::anyhow;
 use spacetimedb_datastore::execution_context::Workload;
 use spacetimedb_datastore::locking_tx_datastore::state_view::StateView;
 use spacetimedb_datastore::traits::IsolationLevel;
+use spacetimedb_expr::check::SchemaView;
+use spacetimedb_expr::errors::TypingError;
+use spacetimedb_expr::expr::CallParams;
 use spacetimedb_expr::statement::Statement;
 use spacetimedb_lib::identity::AuthCtx;
 use spacetimedb_lib::metrics::ExecutionMetrics;
 use spacetimedb_lib::Timestamp;
 use spacetimedb_lib::{AlgebraicType, ProductType, ProductValue};
+use spacetimedb_primitives::{ArgId, TableId};
 use spacetimedb_query::{compile_sql_stmt, execute_dml_stmt, execute_select_stmt};
 use spacetimedb_schema::relation::FieldName;
+use spacetimedb_schema::schema::TableOrViewSchema;
 use spacetimedb_vm::eval::run_ast;
 use spacetimedb_vm::expr::{CodeResult, CrudExpr, Expr};
 use spacetimedb_vm::relation::MemTable;
@@ -185,6 +191,19 @@ pub struct SqlResult {
     pub metrics: ExecutionMetrics,
 }
 
+struct DbParams<'a> {
+    db: &'a RelationalDB,
+    tx: &'a mut MutTx,
+}
+
+impl CallParams for DbParams<'_> {
+    fn create_or_get_param(&mut self, param: &ProductValue) -> Result<ArgId, TypingError> {
+        self.db
+            .create_or_get_params(self.tx, &param)
+            .map_err(|err| TypingError::Other(err.into()))
+    }
+}
+
 /// Run the `SQL` string using the `auth` credentials
 pub async fn run(
     db: &RelationalDB,
@@ -196,9 +215,20 @@ pub async fn run(
 ) -> Result<SqlResult, DBError> {
     // We parse the sql statement in a mutable transaction.
     // If it turns out to be a query, we downgrade the tx.
-    let (tx, stmt) = db.with_auto_rollback(db.begin_mut_tx(IsolationLevel::Serializable, Workload::Sql), |tx| {
-        compile_sql_stmt(sql_text, &SchemaViewer::new(tx, &auth), &auth)
-    })?;
+    let (tx, stmt) =
+        db.with_auto_rollback(
+            db.begin_mut_tx(IsolationLevel::Serializable, Workload::Sql),
+            |tx| match compile_sql_stmt(sql_text, &SchemaViewer::new(tx, &auth), &auth) {
+                Ok(Statement::Select(mut stmt)) => {
+                    stmt.for_each_fun_call(&mut |param| {
+                        db.create_or_get_params(tx, &param)
+                            .map_err(|err| TypingError::Other(err.into()))
+                    })?;
+                    Ok(Statement::Select(stmt))
+                }
+                result => result,
+            },
+        )?;
 
     let mut metrics = ExecutionMetrics::default();
 
@@ -345,7 +375,8 @@ pub(crate) mod tests {
     use itertools::Itertools;
     use pretty_assertions::assert_eq;
     use spacetimedb_datastore::system_tables::{
-        StRowLevelSecurityRow, StTableFields, ST_ROW_LEVEL_SECURITY_ID, ST_TABLE_ID, ST_TABLE_NAME,
+        StRowLevelSecurityRow, StTableFields, ST_RESERVED_SEQUENCE_RANGE, ST_ROW_LEVEL_SECURITY_ID, ST_TABLE_ID,
+        ST_TABLE_NAME,
     };
     use spacetimedb_lib::bsatn::ToBsatn;
     use spacetimedb_lib::db::auth::{StAccess, StTableType};
@@ -958,7 +989,7 @@ pub(crate) mod tests {
         let db = TestDB::in_memory()?;
 
         let schema = [("a", AlgebraicType::U8), ("b", AlgebraicType::U8)];
-        let (_, table_id) = tests_utils::create_view_for_test(&db, "my_view", &schema, false)?;
+        let (_, table_id) = tests_utils::create_view_for_test(&db, "my_view", &schema, ProductType::unit(), false)?;
 
         with_auto_commit(&db, |tx| -> Result<_, DBError> {
             tests_utils::insert_into_view(&db, tx, table_id, Some(identity_from_u8(1)), product![0u8, 1u8])?;
@@ -979,7 +1010,7 @@ pub(crate) mod tests {
         let db = TestDB::in_memory()?;
 
         let schema = [("a", AlgebraicType::U8), ("b", AlgebraicType::U8)];
-        let (_, table_id) = tests_utils::create_view_for_test(&db, "my_view", &schema, true)?;
+        let (_, table_id) = tests_utils::create_view_for_test(&db, "my_view", &schema, ProductType::unit(), true)?;
 
         with_auto_commit(&db, |tx| -> Result<_, DBError> {
             tests_utils::insert_into_view(&db, tx, table_id, None, product![0u8, 1u8])?;
@@ -1000,7 +1031,7 @@ pub(crate) mod tests {
         let db = TestDB::in_memory()?;
 
         let schema = [("a", AlgebraicType::U8), ("b", AlgebraicType::U8)];
-        let (_, v_id) = tests_utils::create_view_for_test(&db, "v", &schema, false)?;
+        let (_, v_id) = tests_utils::create_view_for_test(&db, "v", &schema, ProductType::unit(), false)?;
 
         let schema = [("c", AlgebraicType::U8), ("d", AlgebraicType::U8)];
         let t_id = db.create_table_for_test("t", &schema, &[0.into()])?;
@@ -1060,10 +1091,10 @@ pub(crate) mod tests {
         let db = TestDB::in_memory()?;
 
         let schema = [("a", AlgebraicType::U8), ("b", AlgebraicType::U8)];
-        let (_, u_id) = tests_utils::create_view_for_test(&db, "u", &schema, false)?;
+        let (_, u_id) = tests_utils::create_view_for_test(&db, "u", &schema, ProductType::unit(), false)?;
 
         let schema = [("c", AlgebraicType::U8), ("d", AlgebraicType::U8)];
-        let (_, v_id) = tests_utils::create_view_for_test(&db, "v", &schema, false)?;
+        let (_, v_id) = tests_utils::create_view_for_test(&db, "v", &schema, ProductType::unit(), false)?;
 
         with_auto_commit(&db, |tx| -> Result<_, DBError> {
             tests_utils::insert_into_view(&db, tx, u_id, Some(identity_from_u8(1)), product![0u8, 1u8])?;
@@ -1574,4 +1605,37 @@ pub(crate) mod tests {
 
         Ok(())
     }
+
+    // Verify calling views with params
+    #[test]
+    fn test_view_params() -> ResultTest<()> {
+        let db = TestDB::durable()?;
+        let schema = [("a", AlgebraicType::U8), ("b", AlgebraicType::I64)];
+        let (_view_id, table_id) = tests_utils::create_view_for_test(
+            &db,
+            "my_view",
+            &schema,
+            ProductType::from([("x", AlgebraicType::U8)]),
+            true,
+        )?;
+        let arg_id = ST_RESERVED_SEQUENCE_RANGE as u64;
+
+        with_auto_commit(&db, |tx| -> Result<_, DBError> {
+            tests_utils::insert_into_view(&db, tx, table_id, None, product![arg_id + 1, 0u8, 1i64])?;
+            tests_utils::insert_into_view(&db, tx, table_id, None, product![arg_id, 1u8, 2i64])?;
+            Ok(())
+        })?;
+
+        assert_eq!(
+            run_for_testing(&db, "select * from my_view(1)")?,
+            vec![product![1u8, 2i64]]
+        );
+
+        assert_eq!(
+            run_for_testing(&db, "select * from my_view(2)")?,
+            vec![product![0u8, 1i64]]
+        );
+
+        Ok(())
+    }
 }

crates/core/src/subscription/module_subscription_actor.rs

@@ -327,7 +327,7 @@ impl ModuleSubscriptions {
         let view_info = plans
             .first()
             .and_then(|plan| plan.return_table())
-            .and_then(|schema| schema.view_info);
+            .and_then(|schema| schema.view_info.clone());
 
         let num_cols = plans
             .first()

crates/core/src/subscription/subscription.rs

@@ -609,7 +609,7 @@ impl AuthAccess for ExecutionSet {
     }
 }
 
-/// Querieshttps://github.com/clockworklabs/SpacetimeDBPrivate/pull/2207 all the [`StTableType::User`] tables *right now*
+/// Queries https://github.com/clockworklabs/SpacetimeDBPrivate/pull/2207 all the [`StTableType::User`] tables *right now*
 /// and turns them into [`QueryExpr`],
 /// the moral equivalent of `SELECT * FROM table`.
 pub(crate) fn get_all<T, F, I>(

crates/datastore/src/locking_tx_datastore/mut_tx.rs

@@ -8,6 +8,8 @@ use super::{
     tx_state::{IndexIdMap, PendingSchemaChange, TxState, TxTableForInsertion},
     SharedMutexGuard, SharedWriteGuard,
 };
+use crate::error::DatastoreError;
+use crate::system_tables::{StViewArgFields, StViewArgRow, ST_VIEW_ARG_ID};
 use crate::{
     error::ViewError,
     locking_tx_datastore::state_view::EqOnColumn,
@@ -48,6 +50,7 @@ use spacetimedb_lib::{
 use spacetimedb_primitives::{
     col_list, ArgId, ColId, ColList, ColSet, ConstraintId, IndexId, ScheduleId, SequenceId, TableId, ViewFnPtr, ViewId,
 };
+use spacetimedb_sats::bsatn::ToBsatn;
 use spacetimedb_sats::{
     bsatn::{self, to_writer, DecodeError, Deserializer},
     de::{DeserializeSeed, WithBound},
@@ -589,6 +592,33 @@ impl MutTxId {
         Ok(())
     }
 
+    pub fn get_params(&self, params: ProductValue) -> Result<Option<ArgId>> {
+        self.iter_by_col_eq(ST_VIEW_ARG_ID, StViewArgFields::Bytes, &AlgebraicValue::Product(params))?
+            .next()
+            .map(|row| Ok(ArgId(StViewArgRow::try_from(row)?.id)))
+            .transpose()
+    }
+
+    /// Create parameters for a view, storing the values in `st_view_arg`.
+    pub fn create_or_get_params(&mut self, params: &ProductValue) -> Result<ArgId> {
+        if let Some(arg_id) = self.get_params(params.clone())? {
+            return Ok(arg_id);
+        }
+        let row = StViewArgRow {
+            id: ArgId::SENTINEL.0,
+            bytes: params
+                .to_bsatn_vec()
+                .map_err(|err| DatastoreError::ReadViaBsatnError(err.into()))?
+                .into(),
+        };
+        let arg_id = self
+            .insert_via_serialize_bsatn(ST_VIEW_ARG_ID, &row)?
+            .1
+            .collapse()
+            .read_col::<u64>(StViewArgFields::Id)?;
+        Ok(ArgId(arg_id))
+    }
+
     /// Create a table.
     ///
     /// Requires:

crates/datastore/src/locking_tx_datastore/state_view.rs

@@ -5,15 +5,17 @@ use crate::locking_tx_datastore::mut_tx::{IndexScanPoint, IndexScanRanged};
 use crate::system_tables::{
     ConnectionIdViaU128, StColumnFields, StColumnRow, StConnectionCredentialsFields, StConnectionCredentialsRow,
     StConstraintFields, StConstraintRow, StIndexFields, StIndexRow, StScheduledFields, StScheduledRow,
-    StSequenceFields, StSequenceRow, StTableFields, StTableRow, StViewFields, StViewParamFields, StViewRow,
-    SystemTable, ST_COLUMN_ID, ST_CONNECTION_CREDENTIALS_ID, ST_CONSTRAINT_ID, ST_INDEX_ID, ST_SCHEDULED_ID,
+    StSequenceFields, StSequenceRow, StTableFields, StTableRow, StViewFields, StViewParamFields, StViewParamRow,
+    StViewRow, SystemTable, ST_COLUMN_ID, ST_CONNECTION_CREDENTIALS_ID, ST_CONSTRAINT_ID, ST_INDEX_ID, ST_SCHEDULED_ID,
     ST_SEQUENCE_ID, ST_TABLE_ID, ST_VIEW_ID, ST_VIEW_PARAM_ID,
 };
 use anyhow::anyhow;
 use core::ops::RangeBounds;
 use spacetimedb_lib::ConnectionId;
 use spacetimedb_primitives::{ColList, TableId};
 use spacetimedb_sats::AlgebraicValue;
+use spacetimedb_schema::def::ViewParamDefSimple;
+use spacetimedb_schema::identifier::Identifier;
 use spacetimedb_schema::schema::{ColumnSchema, TableSchema, ViewDefInfo};
 use spacetimedb_table::table::IndexScanPointIter;
 use spacetimedb_table::{
@@ -133,14 +135,20 @@ pub trait StateView {
             .map(|mut iter| {
                 iter.next().map(|row| -> Result<_> {
                     let row = StViewRow::try_from(row)?;
-                    let has_args = self
+                    let args: Vec<_> = self
                         .iter_by_col_eq(ST_VIEW_PARAM_ID, StViewParamFields::ViewId, &row.view_id.into())?
-                        .next()
-                        .is_some();
+                        .map(|param_row| {
+                            let param_row = StViewParamRow::try_from(param_row)?;
+                            Ok(ViewParamDefSimple {
+                                name: Identifier::new(param_row.param_name).expect("valid identifier"),
+                                ty: param_row.param_type.0,
+                            })
+                        })
+                        .collect::<Result<Vec<_>>>()?;
 
                     Ok(ViewDefInfo {
                         view_id: row.view_id,
-                        has_args,
+                        args,
                         is_anonymous: row.is_anonymous,
                     })
                 })

crates/datastore/src/system_tables.rs

@@ -991,6 +991,13 @@ pub struct StViewParamRow {
     pub param_type: AlgebraicTypeViaBytes,
 }
 
+impl TryFrom<RowRef<'_>> for StViewParamRow {
+    type Error = DatastoreError;
+    fn try_from(row: RowRef<'_>) -> Result<Self, DatastoreError> {
+        read_via_bsatn(row)
+    }
+}
+
 /// System table [ST_VIEW_SUB_NAME]
 ///
 /// | view_id | arg_id | identity | num_subscribers | has_subscribers | last_called |