Make Table::clone_structure cheaper by:

- Arcing `TableSchema`, and this has benefits elsewhere too.
- Arc<[_]>ing the visitor program instructions.

The data behind the Arcs very rarely change,
which is the perfect case for an Arc.

Author: Centril

crates/bench/benches/special.rs

@@ -90,7 +90,9 @@ fn serialize_benchmarks<T: BenchTable + RandomTable>(c: &mut Criterion) {
     });
 
     let mut table = spacetimedb_table::table::Table::new(
-        TableDef::from_product(name, T::product_type().clone()).into_schema(TableId(0)),
+        TableDef::from_product(name, T::product_type().clone())
+            .into_schema(TableId(0))
+            .into(),
         spacetimedb_table::indexes::SquashedOffset::COMMITTED_STATE,
     );
     let mut blob_store = spacetimedb_table::blob_store::HashMapBlobStore::default();

crates/core/src/db/datastore/locking_tx_datastore/committed_state.rs

@@ -52,7 +52,7 @@ pub(crate) struct CommittedState {
 }
 
 impl StateView for CommittedState {
-    fn get_schema(&self, table_id: &TableId) -> Option<&TableSchema> {
+    fn get_schema(&self, table_id: &TableId) -> Option<&Arc<TableSchema>> {
         self.tables.get(table_id).map(|table| table.get_schema())
     }
     fn iter<'a>(&'a self, ctx: &'a ExecutionContext, table_id: &TableId) -> Result<Iter<'a>> {
@@ -62,11 +62,7 @@ impl StateView for CommittedState {
         Err(TableError::IdNotFound(SystemTable::st_table, table_id.0).into())
     }
     fn table_exists(&self, table_id: &TableId) -> Option<&str> {
-        if let Some(table) = self.tables.get(table_id) {
-            Some(&table.schema.table_name)
-        } else {
-            None
-        }
+        self.tables.get(table_id).map(|t| &*t.schema.table_name)
     }
     /// Returns an iterator,
     /// yielding every row in the table identified by `table_id`,
@@ -118,7 +114,7 @@ impl CommittedState {
         };
 
         // Insert the table row into st_tables, creating st_tables if it's missing
-        let (st_tables, blob_store) = self.get_table_and_blob_store_or_create(ST_TABLES_ID, st_table_schema());
+        let (st_tables, blob_store) = self.get_table_and_blob_store_or_create(ST_TABLES_ID, st_table_schema().into());
         // Insert the table row into `st_tables` for all system tables
         for schema in system_tables() {
             let table_id = schema.table_id;
@@ -140,7 +136,8 @@ impl CommittedState {
         }
 
         // Insert the columns into `st_columns`
-        let (st_columns, blob_store) = self.get_table_and_blob_store_or_create(ST_COLUMNS_ID, st_columns_schema());
+        let (st_columns, blob_store) =
+            self.get_table_and_blob_store_or_create(ST_COLUMNS_ID, st_columns_schema().into());
         for col in system_tables().into_iter().flat_map(|x| x.into_columns()) {
             let row = StColumnRow {
                 table_id: col.table_id,
@@ -160,7 +157,7 @@ impl CommittedState {
 
         // Insert constraints into `st_constraints`
         let (st_constraints, blob_store) =
-            self.get_table_and_blob_store_or_create(ST_CONSTRAINTS_ID, st_constraints_schema());
+            self.get_table_and_blob_store_or_create(ST_CONSTRAINTS_ID, st_constraints_schema().into());
         for (i, constraint) in system_tables()
             .into_iter()
             .flat_map(|x| x.constraints)
@@ -185,7 +182,8 @@ impl CommittedState {
         }
 
         // Insert the indexes into `st_indexes`
-        let (st_indexes, blob_store) = self.get_table_and_blob_store_or_create(ST_INDEXES_ID, st_indexes_schema());
+        let (st_indexes, blob_store) =
+            self.get_table_and_blob_store_or_create(ST_INDEXES_ID, st_indexes_schema().into());
         for (i, index) in system_tables()
             .into_iter()
             .flat_map(|x| x.indexes)
@@ -216,7 +214,7 @@ impl CommittedState {
 
         // Insert the sequences into `st_sequences`
         let (st_sequences, blob_store) =
-            self.get_table_and_blob_store_or_create(ST_SEQUENCES_ID, st_sequences_schema());
+            self.get_table_and_blob_store_or_create(ST_SEQUENCES_ID, st_sequences_schema().into());
         // We create sequences last to get right the starting number
         // so, we don't sort here
         for (i, col) in system_tables().into_iter().flat_map(|x| x.sequences).enumerate() {
@@ -245,8 +243,8 @@ impl CommittedState {
         // Re-read the schema with the correct ids...
         let ctx = ExecutionContext::internal(database_address);
         for schema in system_tables() {
-            *self.tables.get_mut(&schema.table_id).unwrap().schema =
-                self.schema_for_table_raw(&ctx, schema.table_id)?;
+            self.tables.get_mut(&schema.table_id).unwrap().schema =
+                Arc::new(self.schema_for_table_raw(&ctx, schema.table_id)?);
         }
 
         Ok(())
@@ -266,7 +264,7 @@ impl CommittedState {
         Ok(())
     }
 
-    pub fn replay_insert(&mut self, table_id: TableId, schema: &TableSchema, row: &ProductValue) -> Result<()> {
+    pub fn replay_insert(&mut self, table_id: TableId, schema: &Arc<TableSchema>, row: &ProductValue) -> Result<()> {
         let (table, blob_store) = self.get_table_and_blob_store_or_create_ref_schema(table_id, schema);
         table.insert_internal(blob_store, row).map_err(TableError::Insert)?;
         Ok(())
@@ -332,9 +330,7 @@ impl CommittedState {
 
         // Construct their schemas and insert tables for them.
         for table_id in table_ids {
-            let schema = self
-                .schema_for_table(&ExecutionContext::default(), table_id)?
-                .into_owned();
+            let schema = self.schema_for_table(&ExecutionContext::default(), table_id)?;
             self.tables
                 .insert(table_id, Table::new(schema, SquashedOffset::COMMITTED_STATE));
         }
@@ -453,11 +449,8 @@ impl CommittedState {
         //             and the fullness of the page.
 
         for (table_id, mut tx_table) in insert_tables {
-            let (commit_table, commit_blob_store) = self.get_table_and_blob_store_or_create(
-                table_id,
-                // TODO(perf): avoid cloning here.
-                *tx_table.schema.clone(),
-            );
+            let (commit_table, commit_blob_store) =
+                self.get_table_and_blob_store_or_create(table_id, tx_table.schema.clone());
 
             // NOTE: if there is a schema change the table id will not change
             // and that is what is important here so it doesn't matter if we
@@ -527,13 +520,13 @@ impl CommittedState {
 
     fn create_table(&mut self, table_id: TableId, schema: TableSchema) {
         self.tables
-            .insert(table_id, Table::new(schema, SquashedOffset::COMMITTED_STATE));
+            .insert(table_id, Table::new(Arc::new(schema), SquashedOffset::COMMITTED_STATE));
     }
 
     pub fn get_table_and_blob_store_or_create_ref_schema<'this>(
         &'this mut self,
         table_id: TableId,
-        schema: &'_ TableSchema,
+        schema: &Arc<TableSchema>,
     ) -> (&'this mut Table, &'this mut dyn BlobStore) {
         let table = self
             .tables
@@ -546,7 +539,7 @@ impl CommittedState {
     pub fn get_table_and_blob_store_or_create(
         &mut self,
         table_id: TableId,
-        schema: TableSchema,
+        schema: Arc<TableSchema>,
     ) -> (&mut Table, &mut dyn BlobStore) {
         let table = self
             .tables

crates/core/src/db/datastore/locking_tx_datastore/datastore.rs

@@ -14,7 +14,8 @@ use crate::{
                 Epoch, StModuleFields, StModuleRow, StTableFields, ST_MODULE_ID, ST_TABLES_ID, WASM_MODULE,
             },
             traits::{
-                DataRow, IsolationLevel, MutProgrammable, MutTx, MutTxDatastore, Programmable, Tx, TxData, TxDatastore,
+                DataRow, IsolationLevel, MutProgrammable, MutTx, MutTxDatastore, Programmable, RowTypeForTable, Tx,
+                TxData, TxDatastore,
             },
         },
         db_metrics::{DB_METRICS, MAX_TX_CPU_TIME},
@@ -28,7 +29,7 @@ use anyhow::anyhow;
 use parking_lot::{Mutex, RwLock};
 use spacetimedb_primitives::{ColList, ConstraintId, IndexId, SequenceId, TableId};
 use spacetimedb_sats::db::def::{IndexDef, SequenceDef, TableDef, TableSchema};
-use spacetimedb_sats::{hash::Hash, AlgebraicValue, DataKey, ProductType, ProductValue};
+use spacetimedb_sats::{hash::Hash, AlgebraicValue, DataKey, ProductValue};
 use spacetimedb_table::{indexes::RowPointer, table::RowRef};
 use std::ops::RangeBounds;
 use std::sync::Arc;
@@ -150,10 +151,8 @@ impl Locking {
         let mut committed_state = self.committed_state.write_arc();
         for write in &transaction.writes {
             let table_id = TableId(write.set_id);
-            let schema = committed_state
-                .schema_for_table(&ExecutionContext::default(), table_id)?
-                .into_owned();
-            let table_name = schema.table_name.clone();
+            let schema = committed_state.schema_for_table(&ExecutionContext::default(), table_id)?;
+            let table_name = &schema.table_name;
             let row_type = schema.get_row_type();
 
             let decode_row = |mut data: &[u8], source: &str| {
@@ -192,7 +191,7 @@ impl Locking {
                     // and therefore has performance implications and must not be disabled.
                     DB_METRICS
                         .rdb_num_table_rows
-                        .with_label_values(&self.database_address, &table_id.into(), &table_name)
+                        .with_label_values(&self.database_address, &table_id.into(), table_name)
                         .dec();
                 }
                 Operation::Insert => {
@@ -208,7 +207,7 @@ impl Locking {
                     // and therefore has performance implications and must not be disabled.
                     DB_METRICS
                         .rdb_num_table_rows
-                        .with_label_values(&self.database_address, &table_id.into(), &table_name)
+                        .with_label_values(&self.database_address, &table_id.into(), table_name)
                         .inc();
                 }
             }
@@ -290,11 +289,11 @@ impl TxDatastore for Locking {
         Ok(tx.table_exists(&table_id).map(Cow::Borrowed))
     }
 
-    fn schema_for_table_tx<'tx>(&self, tx: &'tx Self::Tx, table_id: TableId) -> Result<Cow<'tx, TableSchema>> {
+    fn schema_for_table_tx(&self, tx: &Self::Tx, table_id: TableId) -> Result<Arc<TableSchema>> {
         tx.schema_for_table(&ExecutionContext::internal(self.database_address), table_id)
     }
 
-    fn get_all_tables_tx<'tx>(&self, ctx: &ExecutionContext, tx: &'tx Self::Tx) -> Result<Vec<Cow<'tx, TableSchema>>> {
+    fn get_all_tables_tx(&self, ctx: &ExecutionContext, tx: &Self::Tx) -> Result<Vec<Arc<TableSchema>>> {
         self.iter_tx(ctx, tx, ST_TABLES_ID)?
             .map(|row_ref| {
                 let table_id = row_ref.read_col(StTableFields::TableId)?;
@@ -322,15 +321,15 @@ impl MutTxDatastore for Locking {
     /// reflect the schema of the table.q
     ///
     /// This function is known to be called quite frequently.
-    fn row_type_for_table_mut_tx<'tx>(&self, tx: &'tx Self::MutTx, table_id: TableId) -> Result<Cow<'tx, ProductType>> {
+    fn row_type_for_table_mut_tx<'tx>(&self, tx: &'tx Self::MutTx, table_id: TableId) -> Result<RowTypeForTable<'tx>> {
         tx.row_type_for_table(table_id, self.database_address)
     }
 
     /// IMPORTANT! This function is relatively expensive, and much more
     /// expensive than `row_type_for_table_mut_tx`.  Prefer
     /// `row_type_for_table_mut_tx` if you only need to access the `ProductType`
     /// of the table.
-    fn schema_for_table_mut_tx<'tx>(&self, tx: &'tx Self::MutTx, table_id: TableId) -> Result<Cow<'tx, TableSchema>> {
+    fn schema_for_table_mut_tx(&self, tx: &Self::MutTx, table_id: TableId) -> Result<Arc<TableSchema>> {
         tx.schema_for_table(&ExecutionContext::internal(self.database_address), table_id)
     }
 
@@ -1220,7 +1219,7 @@ mod tests {
         datastore.commit_mut_tx_for_test(tx)?;
 
         let mut tx = datastore.begin_mut_tx(IsolationLevel::Serializable);
-        let schema = datastore.schema_for_table_mut_tx(&tx, table_id)?.into_owned();
+        let schema = datastore.schema_for_table_mut_tx(&tx, table_id)?;
 
         for index in &*schema.indexes {
             datastore.drop_index_mut_tx(&mut tx, index.index_id)?;