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/client-api/src/lib.rs

@@ -75,62 +75,49 @@ impl Host {
             .module()
             .await
             .map_err(|_| (StatusCode::NOT_FOUND, "module not found".to_string()))?;
-
-        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),
-                        }],
-                    ))
-                },
-            )
-            .await
-            .map_err(log_and_500)??;
+        println!("Executing SQL: {}", body);
+
+        let (tx_offset, durable_offset, json) = {
+            let db = module_host.module.replica_ctx().relational_db.clone();
+
+            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 _entered = sql_span.enter();
+
+            let result = sql::execute::run(&db, &body, auth, &Some(module_host), &mut header, auth.caller)
+                .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();
+            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();
+
+            Result::<_, axum::Error>::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),
+                }],
+            ))
+        }
+        .unwrap();
 
         if confirmed_read {
             if let Some(mut durable_offset) = durable_offset {
@@ -141,7 +128,6 @@ impl Host {
 
         Ok(json)
     }
-
     pub async fn update(
         &self,
         database: Database,

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,
     ) {