Merge remote-tracking branch 'origin/shub/view-from-host' into noa/ts-views

Author: coolreader18

Cargo.lock

@@ -121,21 +121,21 @@ impl fmt::Debug for EnergyBalance {
     }
 }
 
-/// A measure of energy representing the energy budget for a reducer.
+/// A measure of energy representing the energy budget for a reducer or any callable function.
 ///
 /// In contrast to [`EnergyQuanta`], this is represented by a 64-bit integer. This makes energy handling
 /// for reducers easier, while still providing a unlikely-to-ever-be-reached maximum value (e.g. for wasmtime:
 /// `(u64::MAX eV / 1000 eV/instruction) * 3 ns/instruction = 640 days`)
 #[derive(Copy, Clone, From, Add, Sub)]
-pub struct ReducerBudget(u64);
+pub struct FunctionBudget(u64);
 
-impl ReducerBudget {
+impl FunctionBudget {
     // 1 second of wasm runtime is roughly 2 TeV, so this is
     // roughly 1 minute of wasm runtime
-    pub const DEFAULT_BUDGET: Self = ReducerBudget(120_000_000_000_000);
+    pub const DEFAULT_BUDGET: Self = FunctionBudget(120_000_000_000_000);
 
-    pub const ZERO: Self = ReducerBudget(0);
-    pub const MAX: Self = ReducerBudget(u64::MAX);
+    pub const ZERO: Self = FunctionBudget(0);
+    pub const MAX: Self = FunctionBudget(u64::MAX);
 
     pub fn new(v: u64) -> Self {
         Self(v)
@@ -151,13 +151,13 @@ impl ReducerBudget {
     }
 }
 
-impl From<ReducerBudget> for EnergyQuanta {
-    fn from(value: ReducerBudget) -> Self {
+impl From<FunctionBudget> for EnergyQuanta {
+    fn from(value: FunctionBudget) -> Self {
         EnergyQuanta::new(value.0.into())
     }
 }
 
-impl fmt::Debug for ReducerBudget {
+impl fmt::Debug for FunctionBudget {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         f.debug_tuple("ReducerBudget")
             .field(&EnergyQuanta::from(*self))

crates/client-api-messages/src/energy.rs

@@ -79,57 +79,52 @@ impl Host {
 
         let (tx_offset, durable_offset, json) = self
             .host_controller
-            .using_database(
-                database,
-                self.replica_id,
-                move |db| -> axum::response::Result<_, (StatusCode, String)> {
-                    tracing::info!(sql = body);
-
-                    // We need a header for query results
-                    let mut header = vec![];
-
-                    let sql_start = std::time::Instant::now();
-                    let sql_span =
-                        tracing::trace_span!("execute_sql", total_duration = tracing::field::Empty,).entered();
-
-                    let result = sql::execute::run(
-                        // Returns an empty result set for mutations
-                        db,
-                        &body,
-                        auth,
-                        Some(&module_host.info().subscriptions),
-                        &mut header,
-                    )
-                    .map_err(|e| {
-                        log::warn!("{e}");
-                        if let Some(auth_err) = e.get_auth_error() {
-                            (StatusCode::UNAUTHORIZED, auth_err.to_string())
-                        } else {
-                            (StatusCode::BAD_REQUEST, e.to_string())
-                        }
-                    })?;
-
-                    let total_duration = sql_start.elapsed();
-                    sql_span.record("total_duration", tracing::field::debug(total_duration));
-
-                    // Turn the header into a `ProductType`
-                    let schema = header
-                        .into_iter()
-                        .map(|(col_name, col_type)| ProductTypeElement::new(col_type, Some(col_name)))
-                        .collect();
-
-                    Ok((
-                        result.tx_offset,
-                        db.durable_tx_offset(),
-                        vec![SqlStmtResult {
-                            schema,
-                            rows: result.rows,
-                            total_duration_micros: total_duration.as_micros() as u64,
-                            stats: SqlStmtStats::from_metrics(&result.metrics),
-                        }],
-                    ))
-                },
-            )
+            .using_database(database, self.replica_id, move |db| async move {
+                tracing::info!(sql = body);
+                let mut header = vec![];
+                let sql_start = std::time::Instant::now();
+                let sql_span = tracing::trace_span!("execute_sql", total_duration = tracing::field::Empty,);
+                let _guard = sql_span.enter();
+
+                let result = sql::execute::run(
+                    &db,
+                    &body,
+                    auth,
+                    Some(&module_host.info.subscriptions),
+                    Some(&module_host),
+                    auth.caller,
+                    &mut header,
+                )
+                .await
+                .map_err(|e| {
+                    log::warn!("{e}");
+                    if let Some(auth_err) = e.get_auth_error() {
+                        (StatusCode::UNAUTHORIZED, auth_err.to_string())
+                    } else {
+                        (StatusCode::BAD_REQUEST, e.to_string())
+                    }
+                })?;
+
+                let total_duration = sql_start.elapsed();
+                drop(_guard);
+                sql_span.record("total_duration", tracing::field::debug(total_duration));
+
+                let schema = header
+                    .into_iter()
+                    .map(|(col_name, col_type)| ProductTypeElement::new(col_type, Some(col_name)))
+                    .collect();
+
+                Ok::<_, (StatusCode, String)>((
+                    result.tx_offset,
+                    db.durable_tx_offset(),
+                    vec![SqlStmtResult {
+                        schema,
+                        rows: result.rows,
+                        total_duration_micros: total_duration.as_micros() as u64,
+                        stats: SqlStmtStats::from_metrics(&result.metrics),
+                    }],
+                ))
+            })
             .await
             .map_err(log_and_500)??;
 
@@ -155,7 +150,6 @@ impl Host {
             .await
     }
 }
-
 /// Parameters for publishing a database.
 ///
 /// See [`ControlStateDelegate::publish_database`].

crates/client-api/src/lib.rs

@@ -126,7 +126,8 @@ pub async fn call<S: ControlStateDelegate + NodeDelegate>(
     };
 
     module
-        .call_identity_disconnected(caller_identity, connection_id)
+        // We don't clear views after reducer calls
+        .call_identity_disconnected(caller_identity, connection_id, false)
         .await
         .map_err(client_disconnected_error_to_response)?;
 
@@ -299,7 +300,8 @@ async fn procedure<S: ControlStateDelegate + NodeDelegate>(
     };
 
     module
-        .call_identity_disconnected(caller_identity, connection_id)
+        // We don't clear views after procedure calls
+        .call_identity_disconnected(caller_identity, connection_id, false)
         .await
         .map_err(client_disconnected_error_to_response)?;
 

crates/client-api/src/routes/database.rs

@@ -1,24 +1,28 @@
 use crate::db::MetricsRecorderQueue;
 use crate::error::{DBError, DatabaseError, RestoreSnapshotError};
+use crate::host::ArgsTuple;
 use crate::messages::control_db::HostType;
 use crate::subscription::ExecutionCounters;
 use crate::util::{asyncify, spawn_rayon};
 use crate::worker_metrics::WORKER_METRICS;
 use anyhow::{anyhow, Context};
+use bytes::Bytes;
 use enum_map::EnumMap;
 use fs2::FileExt;
+use log::trace;
 use spacetimedb_commitlog as commitlog;
 use spacetimedb_commitlog::repo::OnNewSegmentFn;
 use spacetimedb_data_structures::map::IntSet;
 use spacetimedb_datastore::db_metrics::DB_METRICS;
-use spacetimedb_datastore::error::{DatastoreError, TableError};
+use spacetimedb_datastore::error::{DatastoreError, TableError, ViewError};
 use spacetimedb_datastore::execution_context::{ReducerContext, Workload, WorkloadType};
 use spacetimedb_datastore::locking_tx_datastore::datastore::TxMetrics;
 use spacetimedb_datastore::locking_tx_datastore::state_view::{
     IterByColEqMutTx, IterByColRangeMutTx, IterMutTx, IterTx, StateView,
 };
 use spacetimedb_datastore::locking_tx_datastore::{MutTxId, TxId};
-use spacetimedb_datastore::system_tables::{system_tables, StModuleRow};
+use spacetimedb_datastore::system_tables::ST_VIEW_ID;
+use spacetimedb_datastore::system_tables::{system_tables, StModuleRow, StViewRow};
 use spacetimedb_datastore::system_tables::{StFields, StVarFields, StVarName, StVarRow, ST_MODULE_ID, ST_VAR_ID};
 use spacetimedb_datastore::traits::{
     InsertFlags, IsolationLevel, Metadata, MutTx as _, MutTxDatastore, Program, RowTypeForTable, Tx as _, TxDatastore,
@@ -32,16 +36,18 @@ use spacetimedb_datastore::{
     traits::TxData,
 };
 use spacetimedb_durability as durability;
+use spacetimedb_lib::bsatn::ToBsatn;
 use spacetimedb_lib::db::auth::StAccess;
 use spacetimedb_lib::db::raw_def::v9::{btree, RawModuleDefV9Builder, RawSql};
+use spacetimedb_lib::de::DeserializeSeed as _;
 use spacetimedb_lib::st_var::StVarValue;
-use spacetimedb_lib::ConnectionId;
 use spacetimedb_lib::Identity;
+use spacetimedb_lib::{bsatn, ConnectionId};
 use spacetimedb_paths::server::{CommitLogDir, ReplicaDir, SnapshotsPath};
 use spacetimedb_primitives::*;
 use spacetimedb_sats::algebraic_type::fmt::fmt_algebraic_type;
 use spacetimedb_sats::memory_usage::MemoryUsage;
-use spacetimedb_sats::{AlgebraicType, AlgebraicValue, ProductType, ProductValue};
+use spacetimedb_sats::{product, AlgebraicType, AlgebraicValue, ProductType, ProductValue, Typespace};
 use spacetimedb_schema::def::{ModuleDef, TableDef, ViewDef};
 use spacetimedb_schema::schema::{
     ColumnSchema, IndexSchema, RowLevelSecuritySchema, Schema, SequenceSchema, TableSchema,
@@ -1403,6 +1409,20 @@ impl RelationalDB {
         Ok(rows_deleted)
     }
 
+    /// Clear all rows from all view tables without dropping them.
+    pub fn clear_all_views(&self, tx: &mut MutTx) -> Result<(), DBError> {
+        for table_id in tx
+            .iter(ST_VIEW_ID)?
+            .map(StViewRow::try_from)
+            .collect::<Result<Vec<_>, _>>()?
+            .into_iter()
+            .filter_map(|row| row.table_id)
+        {
+            tx.clear_table(table_id)?;
+        }
+        Ok(())
+    }
+
     pub fn create_sequence(&self, tx: &mut MutTx, sequence_schema: SequenceSchema) -> Result<SequenceId, DBError> {
         Ok(self.inner.create_sequence_mut_tx(tx, sequence_schema)?)
     }
@@ -1498,8 +1518,106 @@ impl RelationalDB {
             .into()
         })
     }
-}
 
+    /// Materialize View backing table.
+    ///
+    /// # Process
+    /// 1. Serializes view arguments into `ST_VIEW_ARG_ID`
+    /// 2. Deletes stale rows matching the view arguments
+    /// 3. Deserializes the new view execution results
+    /// 4. Inserts fresh rows with the corresponding arg_id
+    ///
+    /// # Arguments
+    /// * `tx` - Mutable transaction context
+    /// * `view` - Name of the view to update
+    /// * `args` - Arguments passed to the view call
+    /// * `return_type` - Expected return type of the view
+    /// * `bytes` - Serialized (bsatn encoded) return value from view execution
+    /// * `typespace` - Type information for deserialization
+    /// * `caller_identity` - Identity of the caller (for non-anonymous views)
+    #[allow(clippy::too_many_arguments)]
+    pub fn materialize_view(
+        &self,
+        tx: &mut MutTxId,
+        view: &str,
+        args: ArgsTuple,
+        return_type: AlgebraicType,
+        bytes: Bytes,
+        typespace: &Typespace,
+        caller_identity: Identity,
+    ) -> Result<(), DBError> {
+        // Fetch view metadata
+        let st_view_row = tx.lookup_st_view_by_name(view)?;
+        let table_id = st_view_row.table_id.expect("View table must exist for materialization");
+        let is_anonymous = st_view_row.is_anonymous;
+
+        let arg_id = tx.get_or_insert_st_view_arg(args.get_bsatn())?;
+
+        // Build the filter key for identifying rows to update
+        let input_args = product![
+            if is_anonymous {
+                AlgebraicValue::OptionNone()
+            } else {
+                AlgebraicValue::OptionSome(caller_identity.into())
+            },
+            AlgebraicValue::U64(arg_id)
+        ];
+
+        // Remove stale View entries
+        let rows_to_delete: Vec<_> = self
+            .iter_by_col_eq_mut(tx, table_id, [0, 1], &input_args.clone().into())?
+            .map(|res| res.pointer())
+            .collect();
+
+        let deleted_count = self.delete(tx, table_id, rows_to_delete);
+        trace!("Deleted {deleted_count} stale rows from view table {table_id} for arg_id {arg_id}");
+
+        // Deserialize the return value
+        let seed = spacetimedb_sats::WithTypespace::new(typespace, &return_type);
+        let return_val = seed
+            .deserialize(bsatn::Deserializer::new(&mut &bytes[..]))
+            .map_err(|e| DatastoreError::from(ViewError::DeserializeReturn(e.to_string())))?;
+
+        // Extract products from return value (must be array or option)
+        let products: Vec<ProductValue> = if return_type.is_array() {
+            let arr = return_val
+                .into_array()
+                .expect("return_type.is_array() ensures this is an array");
+
+            arr.into_iter().map(|v| v.into_product().unwrap()).collect()
+        } else if return_type.is_option() {
+            let opt = return_val
+                .into_option()
+                .expect("return_type.is_option() ensures this is an option");
+            opt.into_iter().map(|v| v.into_product().unwrap()).collect()
+        } else {
+            return Err(DatastoreError::from(ViewError::InvalidReturnType(return_type)).into());
+        };
+
+        // Insert fresh results into the view table
+        let mut elements: Vec<AlgebraicValue> =
+            Vec::with_capacity(input_args.elements.len() + products.first().map_or(0, |p| p.elements.len()));
+        for product in products {
+            elements.clear();
+            // Build complete row by prepending filter key to product data
+            let mut elements = Vec::with_capacity(input_args.elements.len() + product.elements.len());
+            elements.extend_from_slice(&input_args.elements);
+            elements.extend_from_slice(&product.elements);
+
+            let row = ProductValue {
+                elements: elements.into_boxed_slice(),
+            };
+
+            let row_bytes = row
+                .to_bsatn_vec()
+                .map_err(|_| DatastoreError::from(ViewError::SerializeRow))?;
+
+            self.insert(tx, table_id, &row_bytes)?;
+        }
+
+        Ok(())
+    }
+}
 #[allow(unused)]
 #[derive(Clone)]
 struct LockFile {