Update last_called timestamp in st_view_sub for sql queries

Author: joshua-spacetime

Cargo.lock

@@ -195,16 +195,17 @@ pub 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| {
+    let (mut 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 mut metrics = ExecutionMetrics::default();
 
     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

@@ -33,6 +33,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::{db::raw_def::v9::RawSql, metrics::ExecutionMetrics, Timestamp};
 use spacetimedb_lib::{
     db::{auth::StAccess, raw_def::SEQUENCE_ALLOCATION_STEP},
@@ -62,7 +63,7 @@ use spacetimedb_table::{
     table_index::TableIndex,
 };
 use std::{
-    collections::HashMap,
+    collections::{HashMap, HashSet},
     sync::Arc,
     time::{Duration, Instant},
 };
@@ -1767,6 +1768,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: ViewId, 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, ViewId};
 use spacetimedb_schema::schema::TableOrViewSchema;
 use spacetimedb_sql_parser::ast::{BinOp, LogOp};
 
+pub trait CollectViews {
+    fn collect_views(&self, views: &mut HashSet<ViewId>);
+}
+
+impl<T: CollectViews> CollectViews for Arc<T> {
+    fn collect_views(&self, views: &mut HashSet<ViewId>) {
+        self.as_ref().collect_views(views);
+    }
+}
+
+impl<T: CollectViews> CollectViews for Vec<T> {
+    fn collect_views(&self, views: &mut HashSet<ViewId>) {
+        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<ViewId>) {
+        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<ViewId>) {
+        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<ViewId>) {
+        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, ViewId};
 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<ViewId>) {
+        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<ViewId>) {
+        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

@@ -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, ViewId};
 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<ViewId>) {
+        self.plan_opt.collect_views(views);
+    }
+}
+
 impl SubscriptionPlan {
     /// Is this a plan for a join?
     pub fn is_join(&self) -> bool {