VM: Wrap Header in Arc to avoid cloning (#897)

crates/core/src/client/messages.rs

@@ -205,7 +205,7 @@ impl ServerMessage for OneOffQueryResponseMessage {
                 .results
                 .into_iter()
                 .map(|table| OneOffTableJson {
-                    table_name: table.head.table_name,
+                    table_name: table.head.table_name.clone(),
                     rows: table.data,
                 })
                 .collect(),
@@ -221,7 +221,7 @@ impl ServerMessage for OneOffQueryResponseMessage {
                     .results
                     .into_iter()
                     .map(|table| OneOffTable {
-                        table_name: table.head.table_name,
+                        table_name: table.head.table_name.clone(),
                         row: table
                             .data
                             .into_iter()

crates/core/src/sql/compiler.rs

@@ -10,6 +10,7 @@ use spacetimedb_vm::dsl::{db_table, db_table_raw, query};
 use spacetimedb_vm::expr::{ColumnOp, CrudExpr, DbType, Expr, QueryExpr, SourceExpr};
 use spacetimedb_vm::operator::OpCmp;
 use std::collections::HashMap;
+use std::sync::Arc;
 
 use super::ast::TableSchemaView;
 
@@ -182,7 +183,7 @@ fn compile_columns(table: &TableSchema, columns: Vec<FieldName>) -> DbTable {
         }
     }
     DbTable::new(
-        Header::new(table.table_name.clone(), new, table.get_constraints()),
+        Arc::new(Header::new(table.table_name.clone(), new, table.get_constraints())),
         table.table_id,
         table.table_type,
         table.table_access,

crates/core/src/sql/execute.rs

@@ -133,7 +133,7 @@ pub(crate) mod tests {
     use spacetimedb_sats::relation::Header;
     use spacetimedb_sats::{product, AlgebraicType, ProductType};
     use spacetimedb_vm::dsl::{mem_table, scalar};
-    use spacetimedb_vm::eval::create_game_data;
+    use spacetimedb_vm::eval::test_data::create_game_data;
     use tempfile::TempDir;
 
     /// Short-cut for simplify test execution
@@ -369,9 +369,27 @@ pub(crate) mod tests {
         let (db, _tmp_dir) = make_test_db()?;
 
         let mut tx = db.begin_mut_tx(IsolationLevel::Serializable);
-        create_table_with_rows(&db, &mut tx, "Inventory", data.inv.head.into(), &data.inv.data)?;
-        create_table_with_rows(&db, &mut tx, "Player", data.player.head.into(), &data.player.data)?;
-        create_table_with_rows(&db, &mut tx, "Location", data.location.head.into(), &data.location.data)?;
+        create_table_with_rows(
+            &db,
+            &mut tx,
+            "Inventory",
+            data.inv.head.to_product_type(),
+            &data.inv.data,
+        )?;
+        create_table_with_rows(
+            &db,
+            &mut tx,
+            "Player",
+            data.player.head.to_product_type(),
+            &data.player.data,
+        )?;
+        create_table_with_rows(
+            &db,
+            &mut tx,
+            "Location",
+            data.location.head.to_product_type(),
+            &data.location.data,
+        )?;
         db.commit_tx(&ExecutionContext::default(), tx)?;
 
         let result = &run_for_testing(

crates/core/src/subscription/query.rs

@@ -1,3 +1,4 @@
+use std::sync::Arc;
 use std::time::Instant;
 
 use crate::db::db_metrics::{DB_METRICS, MAX_QUERY_COMPILE_TIME};
@@ -32,7 +33,7 @@ pub const OP_TYPE_FIELD_NAME: &str = "__op_type";
 /// Create a virtual table from a sequence of table updates.
 /// Add a special column __op_type to distinguish inserts and deletes.
 #[tracing::instrument(skip_all)]
-pub fn to_mem_table_with_op_type(head: Header, table_access: StAccess, data: &DatabaseTableUpdate) -> MemTable {
+pub fn to_mem_table_with_op_type(head: Arc<Header>, table_access: StAccess, data: &DatabaseTableUpdate) -> MemTable {
     let mut t = MemTable::new(head, table_access, vec![]);
 
     if let Some(pos) = t.head.find_pos_by_name(OP_TYPE_FIELD_NAME) {
@@ -42,11 +43,14 @@ pub fn to_mem_table_with_op_type(head: Header, table_access: StAccess, data: &Da
             new
         }));
     } else {
-        t.head.fields.push(Column::new(
+        // TODO(perf): Eliminate this `clone_for_error` call, as we're not in an error path.
+        let mut head = t.head.clone_for_error();
+        head.fields.push(Column::new(
             FieldName::named(&t.head.table_name, OP_TYPE_FIELD_NAME),
             AlgebraicType::U8,
             t.head.fields.len().into(),
         ));
+        t.head = Arc::new(head);
         for row in &data.ops {
             let mut new = row.row.clone();
             new.elements.push(row.op_type.into());

crates/core/src/subscription/subscription.rs

@@ -27,6 +27,7 @@ use anyhow::Context;
 use derive_more::{Deref, DerefMut, From, IntoIterator};
 use std::collections::{btree_set, BTreeSet, HashMap, HashSet};
 use std::ops::Deref;
+use std::sync::Arc;
 use std::time::Instant;
 
 use crate::db::db_metrics::{DB_METRICS, MAX_QUERY_CPU_TIME};
@@ -681,7 +682,7 @@ impl<'a> IncrementalJoin<'a> {
 /// Replace an [IndexJoin]'s scan or fetch operation with a delta table.
 /// A delta table consists purely of updates or changes to the base table.
 fn with_delta_table(mut join: IndexJoin, index_side: bool, delta: DatabaseTableUpdate) -> IndexJoin {
-    fn to_mem_table(head: Header, table_access: StAccess, delta: DatabaseTableUpdate) -> MemTable {
+    fn to_mem_table(head: Arc<Header>, table_access: StAccess, delta: DatabaseTableUpdate) -> MemTable {
         MemTable::new(
             head,
             table_access,

crates/core/src/vm.rs

@@ -19,6 +19,7 @@ use spacetimedb_vm::iterators::RelIter;
 use spacetimedb_vm::program::ProgramVm;
 use spacetimedb_vm::rel_ops::RelOps;
 use spacetimedb_vm::relation::{MemTable, RelValue, Table};
+use std::sync::Arc;
 
 pub enum TxMode<'a> {
     MutTx(&'a mut MutTx),
@@ -155,7 +156,7 @@ fn join_inner<'a>(
     let header = if semi {
         col_lhs_header.clone()
     } else {
-        col_lhs_header.extend(&col_rhs_header)
+        Arc::new(col_lhs_header.extend(&col_rhs_header))
     };
 
     lhs.join_inner(
@@ -221,7 +222,7 @@ pub struct IndexSemiJoin<'a, Rhs: RelOps<'a>> {
     // The field whose value will be used to probe the index.
     pub probe_field: FieldName,
     // The header for the index side of the join.
-    pub index_header: Header,
+    pub index_header: Arc<Header>,
     // An optional predicate to evaluate over the matching rows of the index.
     pub index_select: Option<ColumnOp>,
     // The table id on which the index is defined.
@@ -261,7 +262,7 @@ impl<'a, Rhs: RelOps<'a>> IndexSemiJoin<'a, Rhs> {
 }
 
 impl<'a, Rhs: RelOps<'a>> RelOps<'a> for IndexSemiJoin<'a, Rhs> {
-    fn head(&self) -> &Header {
+    fn head(&self) -> &Arc<Header> {
         if self.return_index_rows {
             &self.index_header
         } else {
@@ -502,7 +503,7 @@ impl ProgramVm for DbProgram<'_, '_> {
 }
 
 impl<'a> RelOps<'a> for TableCursor<'a> {
-    fn head(&self) -> &Header {
+    fn head(&self) -> &Arc<Header> {
         &self.table.head
     }
 
@@ -516,7 +517,7 @@ impl<'a> RelOps<'a> for TableCursor<'a> {
 }
 
 impl<'a, R: RangeBounds<AlgebraicValue>> RelOps<'a> for IndexCursor<'a, R> {
-    fn head(&self) -> &Header {
+    fn head(&self) -> &Arc<Header> {
         &self.table.head
     }
 
@@ -533,7 +534,7 @@ impl<'a, I> RelOps<'a> for CatalogCursor<I>
 where
     I: Iterator<Item = ProductValue>,
 {
-    fn head(&self) -> &Header {
+    fn head(&self) -> &Arc<Header> {
         &self.table.head
     }
 
@@ -663,7 +664,7 @@ pub(crate) mod tests {
         let result = run_ast(p, q.into());
 
         //The expected result
-        let input = mem_table(schema.head, vec![row]);
+        let input = mem_table(schema.head.clone_for_error(), vec![row]);
 
         assert_eq!(result, Code::Table(input), "{}", name);
     }

crates/sats/src/db/def.rs

@@ -1,6 +1,7 @@
 use derive_more::Display;
 use itertools::Itertools;
 use std::collections::{HashMap, HashSet};
+use std::sync::Arc;
 
 use crate::db::auth::{StAccess, StTableType};
 use crate::db::error::{DefType, SchemaError};
@@ -938,7 +939,12 @@ impl From<&TableSchema> for ProductType {
 
 impl From<&TableSchema> for DbTable {
     fn from(value: &TableSchema) -> Self {
-        DbTable::new(value.into(), value.table_id, value.table_type, value.table_access)
+        DbTable::new(
+            Arc::new(value.into()),
+            value.table_id,
+            value.table_type,
+            value.table_access,
+        )
     }
 }
 

crates/sats/src/relation.rs

@@ -8,14 +8,15 @@ use spacetimedb_primitives::{ColId, ColList, ColListBuilder, Constraints, TableI
 use std::collections::hash_map::DefaultHasher;
 use std::fmt;
 use std::hash::{Hash, Hasher};
+use std::sync::Arc;
 
 pub fn calculate_hash<T: Hash>(t: &T) -> u64 {
     let mut s = DefaultHasher::new();
     t.hash(&mut s);
     s.finish()
 }
 
-#[derive(Debug, Clone, PartialEq, PartialOrd, Eq, Ord, Hash)]
+#[derive(Debug, Clone, Copy, PartialEq, PartialOrd, Eq, Ord, Hash)]
 pub struct TableField<'a> {
     pub table: Option<&'a str>,
     pub field: &'a str,
@@ -34,7 +35,7 @@ pub fn extract_table_field(ident: &str) -> Result<TableField, RelationError> {
     }
 }
 
-#[derive(Debug, Clone, PartialEq, PartialOrd, Eq, Ord, Hash)]
+#[derive(Debug, Clone, Copy, PartialEq, PartialOrd, Eq, Ord, Hash)]
 pub enum FieldOnly<'a> {
     Name(&'a str),
     Pos(usize),
@@ -53,6 +54,7 @@ impl fmt::Display for FieldOnly<'_> {
     }
 }
 
+// TODO(perf): Remove `Clone` derivation.
 #[derive(Debug, Clone, PartialEq, PartialOrd, Eq, Ord, Hash)]
 pub enum FieldName {
     Name { table: String, field: String },
@@ -101,6 +103,7 @@ impl FieldName {
     }
 }
 
+// TODO(perf): Remove `Clone` derivation.
 #[derive(Debug, Clone, PartialEq, PartialOrd, Eq, Ord, Hash, From)]
 pub enum FieldExpr {
     Name(FieldName),
@@ -135,12 +138,13 @@ impl fmt::Display for FieldExpr {
     }
 }
 
-#[derive(Debug, Clone, Eq, PartialEq, Ord, PartialOrd, Hash)]
+#[derive(Debug, Clone, Copy, Eq, PartialEq, Ord, PartialOrd, Hash)]
 pub struct ColumnOnlyField<'a> {
     pub field: FieldOnly<'a>,
     pub algebraic_type: &'a AlgebraicType,
 }
 
+// TODO(perf): Remove `Clone` derivation.
 #[derive(Debug, Clone, Eq, PartialEq, Ord, PartialOrd, Hash)]
 pub struct Column {
     pub field: FieldName,
@@ -165,12 +169,13 @@ impl Column {
     }
 }
 
+// TODO(perf): Remove `Clone` impl.
 #[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
 pub struct HeaderOnlyField<'a> {
     pub fields: Vec<ColumnOnlyField<'a>>,
 }
 
-#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
+#[derive(Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
 pub struct Header {
     pub table_name: String,
     pub fields: Vec<Column>,
@@ -186,6 +191,20 @@ impl Header {
         }
     }
 
+    /// Equivalent to what [`Clone::clone`] would do.
+    ///
+    /// `Header` intentionally does not implement `Clone`,
+    /// as we can't afford to clone it in normal execution paths.
+    /// However, we don't care about performance in error paths,
+    /// and we need to embed owned `Header`s in error objects to report useful messages.
+    pub fn clone_for_error(&self) -> Self {
+        Header {
+            table_name: self.table_name.clone(),
+            fields: self.fields.clone(),
+            constraints: self.constraints.clone(),
+        }
+    }
+
     pub fn from_product_type(table_name: String, fields: ProductType) -> Self {
         let cols = fields
             .elements
@@ -209,6 +228,14 @@ impl Header {
         Self::new(table_name, cols, Default::default())
     }
 
+    pub fn to_product_type(&self) -> ProductType {
+        ProductType::from_iter(
+            self.fields.iter().map(|x| {
+                ProductTypeElement::new(x.algebraic_type.clone(), x.field.field_name().map(ToString::to_string))
+            }),
+        )
+    }
+
     pub fn for_mem_table(fields: ProductType) -> Self {
         let table_name = format!("mem#{:x}", calculate_hash(&fields));
         Self::from_product_type(table_name, fields)
@@ -246,7 +273,7 @@ impl Header {
 
     pub fn column_pos_or_err<'a>(&'a self, col: &'a FieldName) -> Result<ColId, RelationError> {
         self.column_pos(col)
-            .ok_or_else(|| RelationError::FieldNotFound(self.clone(), col.clone()))
+            .ok_or_else(|| RelationError::FieldNotFound(self.clone_for_error(), col.clone()))
     }
 
     /// Finds the position of a field with `name`.
@@ -419,7 +446,7 @@ impl RowCount {
 /// A [Relation] is anything that could be represented as a [Header] of `[ColumnName:ColumnType]` that
 /// generates rows/tuples of [AlgebraicValue] that exactly match that [Header].
 pub trait Relation {
-    fn head(&self) -> &Header;
+    fn head(&self) -> &Arc<Header>;
     /// Specify the size in rows of the [Relation].
     ///
     /// Warning: It should at least be precise in the lower-bound estimate.
@@ -429,14 +456,14 @@ pub trait Relation {
 /// A stored table from [RelationalDB]
 #[derive(Debug, Clone, Eq, PartialEq, PartialOrd, Ord, Hash)]
 pub struct DbTable {
-    pub head: Header,
+    pub head: Arc<Header>,
     pub table_id: TableId,
     pub table_type: StTableType,
     pub table_access: StAccess,
 }
 
 impl DbTable {
-    pub fn new(head: Header, table_id: TableId, table_type: StTableType, table_access: StAccess) -> Self {
+    pub fn new(head: Arc<Header>, table_id: TableId, table_type: StTableType, table_access: StAccess) -> Self {
         Self {
             head,
             table_id,
@@ -447,7 +474,7 @@ impl DbTable {
 }
 
 impl Relation for DbTable {
-    fn head(&self) -> &Header {
+    fn head(&self) -> &Arc<Header> {
         &self.head
     }
 
@@ -489,28 +516,28 @@ mod tests {
         let head = head("t1", ("a", "b"), 0);
         let new = head.project(&[] as &[FieldName]).unwrap();
 
-        let mut empty = head.clone();
+        let mut empty = head.clone_for_error();
         empty.fields.clear();
         empty.constraints.clear();
 
         assert_eq!(empty, new);
 
-        let all = head.clone();
+        let all = head.clone_for_error();
         let new = head
             .project(&[FieldName::named("t1", "a"), FieldName::named("t1", "b")])
             .unwrap();
 
         assert_eq!(all, new);
 
-        let mut first = head.clone();
+        let mut first = head.clone_for_error();
         first.fields.pop();
         first.constraints = first.retain_constraints(&0.into());
 
         let new = head.project(&[FieldName::named("t1", "a")]).unwrap();
 
         assert_eq!(first, new);
 
-        let mut second = head.clone();
+        let mut second = head.clone_for_error();
         second.fields.remove(0);
         second.constraints = second.retain_constraints(&1.into());