invoke view

Author: Shubham8287

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

@@ -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/core/src/db/relational_db.rs

@@ -1,12 +1,15 @@
 use crate::db::MetricsRecorderQueue;
-use crate::error::{DBError, DatabaseError, RestoreSnapshotError};
+use crate::error::{DBError, DatabaseError, RestoreSnapshotError, ViewError};
+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;
@@ -20,6 +23,7 @@ use spacetimedb_datastore::locking_tx_datastore::state_view::{
 use spacetimedb_datastore::locking_tx_datastore::{MutTxId, TxId};
 use spacetimedb_datastore::system_tables::{system_tables, StModuleRow, StViewRow, ST_VIEW_ID};
 use spacetimedb_datastore::system_tables::{StFields, StVarFields, StVarName, StVarRow, ST_MODULE_ID, ST_VAR_ID};
+use spacetimedb_datastore::system_tables::{StViewArgRow, ST_VIEW_ARG_ID};
 use spacetimedb_datastore::traits::{
     InsertFlags, IsolationLevel, Metadata, MutTx as _, MutTxDatastore, Program, RowTypeForTable, Tx as _, TxDatastore,
     UpdateFlags,
@@ -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,
@@ -1512,6 +1518,117 @@ impl RelationalDB {
             .into()
         })
     }
+
+    /// Get or insert view argument into `ST_VIEW_ARG_ID`.
+    pub fn get_or_insert_st_view_arg(&self, tx: &mut MutTxId, args: &Bytes) -> Result<u64, DBError> {
+        let bytes_av = AlgebraicValue::Bytes(args.to_vec().into());
+        let mut rows = self.iter_by_col_eq_mut(tx, ST_VIEW_ARG_ID, col_list![1], &bytes_av)?;
+
+        // Extract the first matching `arg_id`, if any.
+        if let Some(res) = rows.next() {
+            let row = StViewArgRow::try_from(res).expect("valid StViewArgRow");
+            return Ok(row.id);
+        }
+
+        let view_arg_bytes = product![0u64, bytes_av]
+            .to_bsatn_vec()
+            .map_err(|_| ViewError::SerializeArgs)?;
+
+        let (_, view_arg_row, _) = self.insert(tx, ST_VIEW_ARG_ID, &view_arg_bytes)?;
+        let StViewArgRow { id: arg_id, .. } = view_arg_row.try_into().expect("valid StViewArgRow");
+
+        Ok(arg_id)
+    }
+
+    /// Evaluate and update View.
+    /// This involves:
+    /// 1. Serializing the view arguments into `ST_VIEW_ARG_ID`
+    /// 2. Deleting all rows in the view table matching the view arguments
+    /// 3. Deserializing the return value from the view execution
+    /// 4. Inserting all rows from the return value into the view table, with the arg_id
+    ///    set to the inserted view argument's id.
+    /// The `typespace` is needed for deserializing the return value.
+    pub fn evaluate_view(
+        &self,
+        tx: &mut MutTxId,
+        // Name of the view to update
+        view: &str,
+        // Arguments passed to the view call
+        args: ArgsTuple,
+        // Return type of the view call
+        return_type: AlgebraicType,
+        // Serialized bytes of the return value from the view call
+        //TODO: pass arg_id; do the insertion during starting of invoking view
+        bytes: Bytes,
+        typespace: &Typespace,
+        // Identity of the caller (for non-anonymous views)
+        caller_identity: Identity,
+    ) -> Result<(), DBError> {
+        let st_view_row = tx.lookup_st_view_by_name(view)?;
+
+        let (table_id, is_anonymous) = (
+            st_view_row
+                .table_id
+                .ok_or_else(|| ViewError::ViewNotFound(view.to_string()))?,
+            st_view_row.is_anonymous,
+        );
+
+        // Insert the view arguments into ST_VIEW_ARG_ID
+        let arg_id = self.get_or_insert_st_view_arg(tx, &args.get_bsatn())?;
+
+        // Delete all existing rows in the view table matching the view arguments
+        let av: AlgebraicValue = args.tuple.into();
+        let rows_to_delete: Vec<_> = self
+            .iter_by_col_eq_mut(tx, table_id, col_list![0], &av)?
+            .map(|res| res.pointer())
+            .collect();
+        let count = self.delete(tx, table_id, rows_to_delete);
+        trace!("Deleted {count} 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| ViewError::DeserializeReturn(e.to_string()))?;
+        let products = Self::extract_products(return_val, &return_type)?;
+
+        // Insert all rows from the return value into the view table
+        for product in products {
+            let row = {
+                let mut elements = Vec::with_capacity(2 + product.elements.len());
+                elements.push(if is_anonymous {
+                    AlgebraicValue::OptionNone()
+                } else {
+                    AlgebraicValue::OptionSome(caller_identity.into())
+                });
+                elements.push(AlgebraicValue::U64(arg_id));
+                elements.extend_from_slice(&product.elements);
+
+                ProductValue {
+                    elements: elements.into_boxed_slice(),
+                }
+            };
+            let row_bytes = row.to_bsatn_vec().map_err(|_| ViewError::SerializeRow)?;
+            self.insert(tx, table_id, &row_bytes)?;
+        }
+
+        Ok(())
+    }
+
+    fn extract_products(
+        return_val: AlgebraicValue,
+        return_type: &AlgebraicType,
+    ) -> Result<Vec<ProductValue>, ViewError> {
+        if return_type.is_array() {
+            let arr = return_val.into_array().expect("return type is array");
+            Ok(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");
+            Ok(opt.into_iter().map(|v| v.into_product().unwrap()).collect())
+        } else {
+            Err(ViewError::InvalidReturnType(return_type.clone()))
+        }
+    }
 }
 
 #[allow(unused)]

crates/core/src/energy.rs

@@ -5,18 +5,18 @@ use spacetimedb_lib::{Hash, Identity};
 use crate::messages::control_db::Database;
 
 pub use spacetimedb_client_api_messages::energy::*;
-pub struct ReducerFingerprint<'a> {
+pub struct FunctionFingerprint<'a> {
     pub module_hash: Hash,
     pub module_identity: Identity,
     pub caller_identity: Identity,
-    pub reducer_name: &'a str,
+    pub function_name: &'a str,
 }
 
 pub trait EnergyMonitor: Send + Sync + 'static {
-    fn reducer_budget(&self, fingerprint: &ReducerFingerprint<'_>) -> ReducerBudget;
+    fn reducer_budget(&self, fingerprint: &FunctionFingerprint<'_>) -> FunctionBudget;
     fn record_reducer(
         &self,
-        fingerprint: &ReducerFingerprint<'_>,
+        fingerprint: &FunctionFingerprint<'_>,
         energy_used: EnergyQuanta,
         execution_duration: Duration,
     );
@@ -29,13 +29,13 @@ pub trait EnergyMonitor: Send + Sync + 'static {
 pub struct NullEnergyMonitor;
 
 impl EnergyMonitor for NullEnergyMonitor {
-    fn reducer_budget(&self, _fingerprint: &ReducerFingerprint<'_>) -> ReducerBudget {
-        ReducerBudget::DEFAULT_BUDGET
+    fn reducer_budget(&self, _fingerprint: &FunctionFingerprint<'_>) -> FunctionBudget {
+        FunctionBudget::DEFAULT_BUDGET
     }
 
     fn record_reducer(
         &self,
-        _fingerprint: &ReducerFingerprint<'_>,
+        _fingerprint: &FunctionFingerprint<'_>,
         _energy_used: EnergyQuanta,
         _execution_duration: Duration,
     ) {

crates/core/src/error.rs

@@ -8,7 +8,7 @@ use hex::FromHexError;
 use spacetimedb_commitlog::repo::TxOffset;
 use spacetimedb_durability::DurabilityExited;
 use spacetimedb_expr::errors::TypingError;
-use spacetimedb_lib::Identity;
+use spacetimedb_lib::{AlgebraicType, Identity};
 use spacetimedb_schema::error::ValidationErrors;
 use spacetimedb_snapshot::SnapshotError;
 use spacetimedb_table::table::ReadViaBsatnError;
@@ -147,6 +147,8 @@ pub enum DBError {
     RestoreSnapshot(#[from] RestoreSnapshotError),
     #[error(transparent)]
     DurabilityGone(#[from] DurabilityExited),
+    #[error("ViewError: {0}")]
+    View(#[from] ViewError),
 }
 
 impl DBError {
@@ -190,6 +192,28 @@ impl<'a, T: ?Sized + 'a> From<PoisonError<std::sync::MutexGuard<'a, T>>> for DBE
     }
 }
 
+#[derive(Error, Debug)]
+pub enum ViewError {
+    #[error("view '{0}' not found")]
+    ViewNotFound(String),
+    #[error("failed to deserialize view arguments from row")]
+    DeserializeArgs,
+    #[error("failed to deserialize view return value: {0}")]
+    DeserializeReturn(String),
+    #[error("failed to serialize row to BSATN")]
+    SerializeRow,
+    #[error("invalid return type: expected Array or Option, got {0:?}")]
+    InvalidReturnType(AlgebraicType),
+    #[error("return type is Array but deserialized value is not Array")]
+    TypeMismatchArray,
+    #[error("return type is Option but deserialized value is not Option")]
+    TypeMismatchOption,
+    #[error("expected ProductValue in view result")]
+    ExpectedProduct,
+    #[error("failed to serialize view arguments")]
+    SerializeArgs,
+}
+
 #[derive(Debug, Error)]
 pub enum LogReplayError {
     #[error(

crates/core/src/host/host_controller.rs

@@ -27,6 +27,7 @@ use spacetimedb_data_structures::error_stream::ErrorStream;
 use spacetimedb_data_structures::map::IntMap;
 use spacetimedb_datastore::db_metrics::data_size::DATA_SIZE_METRICS;
 use spacetimedb_datastore::db_metrics::DB_METRICS;
+use spacetimedb_datastore::locking_tx_datastore::MutTxId;
 use spacetimedb_datastore::traits::Program;
 use spacetimedb_durability::{self as durability};
 use spacetimedb_lib::{hash_bytes, AlgebraicValue, Identity, Timestamp};
@@ -170,6 +171,29 @@ impl From<&EventStatus> for ReducerOutcome {
     }
 }
 
+pub enum ViewOutcome {
+    Success,
+    Failed(String),
+    BudgetExceeded,
+}
+
+impl From<EventStatus> for ViewOutcome {
+    fn from(status: EventStatus) -> Self {
+        match status {
+            EventStatus::Committed(_) => ViewOutcome::Success,
+            EventStatus::Failed(e) => ViewOutcome::Failed(e),
+            EventStatus::OutOfEnergy => ViewOutcome::BudgetExceeded,
+        }
+    }
+}
+
+pub struct ViewCallResult {
+    pub outcome: ViewOutcome,
+    pub tx: MutTxId,
+    pub energy_used: EnergyQuanta,
+    pub execution_duration: Duration,
+}
+
 #[derive(Clone, Debug)]
 pub struct ProcedureCallResult {
     pub return_val: AlgebraicValue,