Materialize views for sql http calls

Author: joshua-spacetime

Cargo.lock

@@ -49,7 +49,7 @@ use spacetimedb_lib::identity::{AuthCtx, RequestId};
 use spacetimedb_lib::metrics::ExecutionMetrics;
 use spacetimedb_lib::ConnectionId;
 use spacetimedb_lib::Timestamp;
-use spacetimedb_primitives::{ProcedureId, TableId, ViewDatabaseId, ViewId};
+use spacetimedb_primitives::{ProcedureId, TableId, ViewDatabaseId};
 use spacetimedb_query::compile_subscription;
 use spacetimedb_sats::{AlgebraicTypeRef, ProductValue};
 use spacetimedb_schema::auto_migrate::{AutoMigrateError, MigrationPolicy};
@@ -541,7 +541,7 @@ pub struct CallViewParams {
     pub timestamp: Timestamp,
     pub caller_identity: Identity,
     pub caller_connection_id: Option<ConnectionId>,
-    pub view_id: ViewId,
+    pub view_id: ViewDatabaseId,
     pub view_db_id: ViewDatabaseId,
     pub args: ArgsTuple,
 
@@ -1531,7 +1531,7 @@ impl ModuleHost {
     async fn call_view_inner(
         &self,
         tx: MutTxId,
-        view_id: ViewId,
+        view_id: ViewDatabaseId,
         view_def: &ViewDef,
         args: ArgsTuple,
         caller_identity: Identity,

crates/core/src/host/module_host.rs

@@ -4,7 +4,6 @@ use spacetimedb_lib::db::raw_def::v9::Lifecycle;
 use spacetimedb_lib::de::DeserializeSeed as _;
 use spacetimedb_primitives::ProcedureId;
 use spacetimedb_primitives::ViewDatabaseId;
-use spacetimedb_primitives::ViewId;
 use spacetimedb_schema::auto_migrate::{MigratePlan, MigrationPolicy, MigrationPolicyError};
 use std::future::Future;
 use std::sync::Arc;
@@ -992,7 +991,7 @@ fn commit_and_broadcast_event(
 /// Describes a view call in a cheaply shareable way.
 #[derive(Clone, Debug)]
 pub struct ViewOp<'a> {
-    pub id: ViewId,
+    pub id: ViewDatabaseId,
     pub db_id: ViewDatabaseId,
     pub name: &'a str,
     pub args: &'a ArgsTuple,
@@ -1003,7 +1002,7 @@ pub struct ViewOp<'a> {
 /// Describes an anonymous view call in a cheaply shareable way.
 #[derive(Clone, Debug)]
 pub struct AnonymousViewOp<'a> {
-    pub id: ViewId,
+    pub id: ViewDatabaseId,
     pub db_id: ViewDatabaseId,
     pub name: &'a str,
     pub args: &'a ArgsTuple,

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

@@ -233,8 +233,9 @@ pub async fn run(
 
     match stmt {
         Statement::Select(stmt) => {
-            // Up to this point, the tx has been read-only,
-            // and hence there are no deltas to process.
+            // Materialize views before we downgrade to a read-only transaction
+            tx.materialize_views(&stmt, auth.caller)?;
+
             let (tx_data, tx_metrics_mut, tx) = tx.commit_downgrade(Workload::Sql);
 
             let (tx_offset_send, tx_offset) = oneshot::channel();

crates/core/src/sql/execute.rs

@@ -19,6 +19,7 @@ spacetimedb-schema.workspace = true
 spacetimedb-table.workspace = true
 spacetimedb-snapshot.workspace = true
 spacetimedb-execution.workspace = true
+spacetimedb-expr.workspace = true
 
 anyhow = { workspace = true, features = ["backtrace"] }
 bytes.workspace = true

crates/datastore/Cargo.toml

@@ -35,6 +35,7 @@ use smallvec::SmallVec;
 use spacetimedb_data_structures::map::{IntMap, IntSet};
 use spacetimedb_durability::TxOffset;
 use spacetimedb_execution::{dml::MutDatastore, Datastore, DeltaStore, Row};
+use spacetimedb_expr::expr::CollectViews;
 use spacetimedb_lib::{bsatn::ToBsatn as _, db::raw_def::v9::RawSql, metrics::ExecutionMetrics, Timestamp};
 use spacetimedb_lib::{
     db::{auth::StAccess, raw_def::SEQUENCE_ALLOCATION_STEP},
@@ -65,7 +66,7 @@ use spacetimedb_table::{
     table_index::TableIndex,
 };
 use std::{
-    collections::HashMap,
+    collections::{HashMap, HashSet},
     sync::Arc,
     time::{Duration, Instant},
 };
@@ -1849,6 +1850,19 @@ impl<'a, I: Iterator<Item = RowRef<'a>>> Iterator for FilterDeleted<'a, I> {
 }
 
 impl MutTxId {
+    /// Materialize views for `sender`, collected from `view_collector`.
+    pub fn materialize_views(&mut self, view_collector: &impl CollectViews, sender: Identity) -> Result<()> {
+        let mut view_ids = HashSet::new();
+        view_collector.collect_views(&mut view_ids);
+        for view_id in view_ids {
+            if !self.is_view_materialized(view_id, ArgId::SENTINEL, sender)? {
+                // TODO: __call_view__
+            }
+            self.st_view_sub_update_or_insert_last_called(view_id, ArgId::SENTINEL, sender)?;
+        }
+        Ok(())
+    }
+
     /// Does this caller have an entry for `view_id` in `st_view_sub`?
     pub fn is_view_materialized(&self, view_id: ViewDatabaseId, arg_id: ArgId, sender: Identity) -> Result<bool> {
         use StViewSubFields::*;

crates/datastore/src/locking_tx_datastore/mut_tx.rs

@@ -1,10 +1,28 @@
-use std::sync::Arc;
+use std::{collections::HashSet, sync::Arc};
 
 use spacetimedb_lib::{query::Delta, AlgebraicType, AlgebraicValue};
-use spacetimedb_primitives::TableId;
+use spacetimedb_primitives::{TableId, ViewDatabaseId};
 use spacetimedb_schema::schema::TableOrViewSchema;
 use spacetimedb_sql_parser::ast::{BinOp, LogOp};
 
+pub trait CollectViews {
+    fn collect_views(&self, views: &mut HashSet<ViewDatabaseId>);
+}
+
+impl<T: CollectViews> CollectViews for Arc<T> {
+    fn collect_views(&self, views: &mut HashSet<ViewDatabaseId>) {
+        self.as_ref().collect_views(views);
+    }
+}
+
+impl<T: CollectViews> CollectViews for Vec<T> {
+    fn collect_views(&self, views: &mut HashSet<ViewDatabaseId>) {
+        for item in self {
+            item.collect_views(views);
+        }
+    }
+}
+
 /// A projection is the root of any relational expression.
 /// This type represents a projection that returns relvars.
 ///
@@ -25,6 +43,14 @@ pub enum ProjectName {
     Some(RelExpr, Box<str>),
 }
 
+impl CollectViews for ProjectName {
+    fn collect_views(&self, views: &mut HashSet<ViewDatabaseId>) {
+        match self {
+            Self::None(expr) | Self::Some(expr, _) => expr.collect_views(views),
+        }
+    }
+}
+
 impl ProjectName {
     /// Unwrap the outer projection, returning the inner expression
     pub fn unwrap(self) -> RelExpr {
@@ -146,6 +172,26 @@ pub enum AggType {
     Count,
 }
 
+impl CollectViews for ProjectList {
+    fn collect_views(&self, views: &mut HashSet<ViewDatabaseId>) {
+        match self {
+            Self::Limit(proj, _) => {
+                proj.collect_views(views);
+            }
+            Self::Name(exprs) => {
+                for expr in exprs {
+                    expr.collect_views(views);
+                }
+            }
+            Self::List(exprs, _) | Self::Agg(exprs, ..) => {
+                for expr in exprs {
+                    expr.collect_views(views);
+                }
+            }
+        }
+    }
+}
+
 impl ProjectList {
     /// Does this expression project a single relvar?
     /// If so, we return it's [`TableOrViewSchema`].
@@ -212,6 +258,18 @@ pub struct Relvar {
     pub delta: Option<Delta>,
 }
 
+impl CollectViews for RelExpr {
+    fn collect_views(&self, views: &mut HashSet<ViewDatabaseId>) {
+        self.visit(&mut |expr| {
+            if let Self::RelVar(Relvar { schema, .. }) = expr {
+                if let Some(info) = schema.view_info {
+                    views.insert(info.view_id);
+                }
+            }
+        });
+    }
+}
+
 impl RelExpr {
     /// Walk the expression tree and call `f` on each node
     pub fn visit(&self, f: &mut impl FnMut(&Self)) {

crates/expr/src/expr.rs

@@ -1,15 +1,19 @@
 use std::{
     borrow::Cow,
+    collections::HashSet,
     ops::{Bound, Deref, DerefMut},
     sync::Arc,
 };
 
 use anyhow::{bail, Result};
 use derive_more::From;
 use either::Either;
-use spacetimedb_expr::{expr::AggType, StatementSource};
+use spacetimedb_expr::{
+    expr::{AggType, CollectViews},
+    StatementSource,
+};
 use spacetimedb_lib::{identity::AuthCtx, query::Delta, sats::size_of::SizeOf, AlgebraicValue, ProductValue};
-use spacetimedb_primitives::{ColId, ColSet, IndexId, TableId};
+use spacetimedb_primitives::{ColId, ColSet, IndexId, TableId, ViewDatabaseId};
 use spacetimedb_schema::schema::{IndexSchema, TableSchema};
 use spacetimedb_sql_parser::ast::{BinOp, LogOp};
 use spacetimedb_table::table::RowRef;
@@ -68,6 +72,14 @@ impl DerefMut for ProjectPlan {
     }
 }
 
+impl CollectViews for ProjectPlan {
+    fn collect_views(&self, views: &mut HashSet<ViewDatabaseId>) {
+        match self {
+            Self::None(plan) | Self::Name(plan, ..) => plan.collect_views(views),
+        }
+    }
+}
+
 impl ProjectPlan {
     pub fn optimize(self, auth: &AuthCtx) -> Result<Self> {
         match self {
@@ -240,6 +252,29 @@ pub enum PhysicalPlan {
     Filter(Box<PhysicalPlan>, PhysicalExpr),
 }
 
+impl CollectViews for PhysicalPlan {
+    fn collect_views(&self, views: &mut HashSet<ViewDatabaseId>) {
+        self.visit(&mut |plan| match plan {
+            Self::TableScan(scan, _) => {
+                if let Some(info) = scan.schema.view_info {
+                    views.insert(info.view_id);
+                }
+            }
+            Self::IxScan(scan, _) => {
+                if let Some(info) = scan.schema.view_info {
+                    views.insert(info.view_id);
+                }
+            }
+            Self::IxJoin(join, _) => {
+                if let Some(info) = join.rhs.view_info {
+                    views.insert(info.view_id);
+                }
+            }
+            _ => {}
+        });
+    }
+}
+
 impl PhysicalPlan {
     /// Walks the plan tree and calls `f` on every op
     pub fn visit(&self, f: &mut impl FnMut(&Self)) {

crates/physical-plan/src/plan.rs

@@ -37,7 +37,7 @@ use spacetimedb_lib::db::raw_def::v9::{
     RawUniqueConstraintDataV9, RawViewDefV9, TableAccess, TableType,
 };
 use spacetimedb_lib::{ProductType, RawModuleDef};
-use spacetimedb_primitives::{ColId, ColList, ColOrCols, ColSet, ProcedureId, ReducerId, TableId, ViewId};
+use spacetimedb_primitives::{ColId, ColList, ColOrCols, ColSet, ProcedureId, ReducerId, TableId, ViewDatabaseId};
 use spacetimedb_sats::{AlgebraicType, AlgebraicValue};
 use spacetimedb_sats::{AlgebraicTypeRef, Typespace};
 
@@ -246,7 +246,7 @@ impl ModuleDef {
     }
 
     /// Convenience method to look up a view, possibly by a string, returning its id as well.
-    pub fn view_full<K: ?Sized + Hash + Equivalent<Identifier>>(&self, name: &K) -> Option<(ViewId, &ViewDef)> {
+    pub fn view_full<K: ?Sized + Hash + Equivalent<Identifier>>(&self, name: &K) -> Option<(ViewDatabaseId, &ViewDef)> {
         // If the string IS a valid identifier, we can just look it up.
         self.views.get_full(name).map(|(idx, _, def)| (idx.into(), def))
     }
@@ -296,7 +296,7 @@ impl ModuleDef {
     }
 
     /// Look up a view by its id, panicking if it doesn't exist.
-    pub fn view_by_id(&self, id: ViewId) -> &ViewDef {
+    pub fn view_by_id(&self, id: ViewDatabaseId) -> &ViewDef {
         &self.views[id.idx()]
     }
 

crates/schema/src/def.rs

@@ -6,10 +6,10 @@ use spacetimedb_execution::{
     },
     Datastore, DeltaStore, Row,
 };
-use spacetimedb_expr::check::SchemaView;
+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};
+use spacetimedb_primitives::{ColId, ColList, IndexId, TableId, ViewDatabaseId};
 use spacetimedb_query::compile_subscription;
 use std::sync::Arc;
 use std::{collections::HashSet, ops::RangeBounds};
@@ -363,6 +363,12 @@ pub struct SubscriptionPlan {
     plan_opt: ProjectPlan,
 }
 
+impl CollectViews for SubscriptionPlan {
+    fn collect_views(&self, views: &mut HashSet<ViewDatabaseId>) {
+        self.plan_opt.collect_views(views);
+    }
+}
+
 impl SubscriptionPlan {
     /// Is this a plan for a join?
     pub fn is_join(&self) -> bool {