ephermal tables

Shubham8287 · Shubham8287 · commit 5d448901b904 · 2025-11-20T14:47:52.000+05:30
diff --git a/crates/core/src/db/relational_db.rs b/crates/core/src/db/relational_db.rs
@@ -821,9 +821,18 @@ impl RelationalDB {
             Txdata,
         };
 
+        let is_ephemeral_tables = |table_id: &TableId| -> bool {
+            tx_data
+                .ephermal_tables()
+                .map(|etables| etables.contains(table_id))
+                .unwrap_or(false)
+        };
+
         if tx_data.tx_offset().is_some() {
             let inserts: Box<_> = tx_data
                 .inserts()
+                // Skip ephemeral tables
+                .filter(|(table_id, _)| !is_ephemeral_tables(table_id))
                 .map(|(table_id, rowdata)| Ops {
                     table_id: *table_id,
                     rowdata: rowdata.clone(),
@@ -833,7 +842,8 @@ impl RelationalDB {
             let truncates: IntSet<TableId> = tx_data.truncates().collect();
 
             let deletes: Box<_> = tx_data
-                .deletes()
+                .durable_deletes()
+                .filter(|(table_id, _)| !is_ephemeral_tables(table_id))
                 .map(|(table_id, rowdata)| Ops {
                     table_id: *table_id,
                     rowdata: rowdata.clone(),
@@ -842,6 +852,11 @@ impl RelationalDB {
                 .filter(|ops| !truncates.contains(&ops.table_id))
                 .collect();
 
+            let truncates = truncates
+                .into_iter()
+                .filter(|table_id| !is_ephemeral_tables(table_id))
+                .collect();
+
             let inputs = reducer_context.map(|rcx| rcx.into());
 
             let txdata = Txdata {
@@ -850,7 +865,7 @@ impl RelationalDB {
                 mutations: Some(Mutations {
                     inserts,
                     deletes,
-                    truncates: truncates.into_iter().collect(),
+                    truncates,
                 }),
             };
 
diff --git a/crates/datastore/src/error.rs b/crates/datastore/src/error.rs
@@ -1,7 +1,7 @@
 use super::system_tables::SystemTable;
 use enum_as_inner::EnumAsInner;
 use spacetimedb_lib::db::raw_def::{v9::RawSql, RawIndexDefV8};
-use spacetimedb_primitives::{ColId, ColList, IndexId, SequenceId, TableId};
+use spacetimedb_primitives::{ColId, ColList, IndexId, SequenceId, TableId, ViewId};
 use spacetimedb_sats::buffer::DecodeError;
 use spacetimedb_sats::{product_value::InvalidFieldError, satn::Satn};
 use spacetimedb_sats::{AlgebraicType, AlgebraicValue, ProductValue};
@@ -41,6 +41,8 @@ pub enum DatastoreError {
 pub enum ViewError {
     #[error("view '{0}' not found")]
     NotFound(Box<str>),
+    #[error("Table backing View '{0}' not found")]
+    TableNotFound(ViewId),
     #[error("failed to deserialize view arguments from row")]
     DeserializeArgs,
     #[error("failed to deserialize view return value: {0}")]
diff --git a/crates/datastore/src/locking_tx_datastore/committed_state.rs b/crates/datastore/src/locking_tx_datastore/committed_state.rs
@@ -8,18 +8,18 @@ use super::{
 };
 use crate::{
     db_metrics::DB_METRICS,
-    error::{DatastoreError, IndexError, TableError},
+    error::{DatastoreError, IndexError, TableError, ViewError},
     execution_context::ExecutionContext,
     locking_tx_datastore::{mut_tx::ViewReadSets, state_view::iter_st_column_for_table},
     system_tables::{
         system_tables, StColumnRow, StConstraintData, StConstraintRow, StIndexRow, StSequenceRow, StTableFields,
-        StTableRow, SystemTable, ST_CLIENT_ID, ST_CLIENT_IDX, ST_COLUMN_ID, ST_COLUMN_IDX, ST_COLUMN_NAME,
+        StTableRow, StViewRow, SystemTable, ST_CLIENT_ID, ST_CLIENT_IDX, ST_COLUMN_ID, ST_COLUMN_IDX, ST_COLUMN_NAME,
         ST_CONSTRAINT_ID, ST_CONSTRAINT_IDX, ST_CONSTRAINT_NAME, ST_INDEX_ID, ST_INDEX_IDX, ST_INDEX_NAME,
         ST_MODULE_ID, ST_MODULE_IDX, ST_ROW_LEVEL_SECURITY_ID, ST_ROW_LEVEL_SECURITY_IDX, ST_SCHEDULED_ID,
         ST_SCHEDULED_IDX, ST_SEQUENCE_ID, ST_SEQUENCE_IDX, ST_SEQUENCE_NAME, ST_TABLE_ID, ST_TABLE_IDX, ST_VAR_ID,
         ST_VAR_IDX, ST_VIEW_ARG_ID, ST_VIEW_ARG_IDX,
     },
-    traits::TxData,
+    traits::{EphemeralTables, TxData},
 };
 use crate::{
     locking_tx_datastore::ViewCallInfo,
@@ -80,6 +80,12 @@ pub struct CommittedState {
     /// Any overlap will trigger a re-evaluation of the affected view,
     /// and its read set will be updated accordingly.
     read_sets: ViewReadSets,
+
+    /// Tables which do not needs to be made persistent.
+    /// These include:
+    ///     - system tables: `st_view_sub`, `st_view_arg`
+    ///     - Tables which back views.
+    pub(super) ephemeral_tables: EphemeralTables,
 }
 
 impl CommittedState {
@@ -99,6 +105,7 @@ impl MemoryUsage for CommittedState {
             page_pool: _,
             table_dropped,
             read_sets,
+            ephemeral_tables,
         } = self;
         // NOTE(centril): We do not want to include the heap usage of `page_pool` as it's a shared resource.
         next_tx_offset.heap_usage()
@@ -107,6 +114,7 @@ impl MemoryUsage for CommittedState {
             + index_id_map.heap_usage()
             + table_dropped.heap_usage()
             + read_sets.heap_usage()
+            + ephemeral_tables.heap_usage()
     }
 }
 
@@ -171,6 +179,7 @@ impl CommittedState {
             table_dropped: <_>::default(),
             read_sets: <_>::default(),
             page_pool,
+            ephemeral_tables: <_>::default(),
         }
     }
 
@@ -518,6 +527,32 @@ impl CommittedState {
         Ok(())
     }
 
+    pub(super) fn collect_ephemeral_tables(&mut self) -> Result<()> {
+        self.ephemeral_tables = self.ephemeral_tables()?.into_iter().collect();
+        Ok(())
+    }
+
+    fn ephemeral_tables(&self) -> Result<Vec<TableId>> {
+        let mut tables = vec![ST_VIEW_SUB_ID, ST_VIEW_ARG_ID];
+
+        let Some(st_view) = self.tables.get(&ST_VIEW_ID) else {
+            return Ok(tables);
+        };
+        let backing_tables = st_view
+            .scan_rows(&self.blob_store)
+            .map(|row_ref| {
+                let view_row = StViewRow::try_from(row_ref)?;
+                view_row
+                    .table_id
+                    .ok_or_else(|| DatastoreError::View(ViewError::TableNotFound(view_row.view_id)))
+            })
+            .collect::<Result<Vec<_>>>()?;
+
+        tables.extend(backing_tables);
+
+        Ok(tables)
+    }
+
     /// After replaying all old transactions,
     /// inserts and deletes into the system tables
     /// might not be reflected in the schemas of the built tables.
@@ -675,6 +710,8 @@ impl CommittedState {
             self.next_tx_offset += 1;
         }
 
+        tx_data.set_ephemeral_tables(&self.ephemeral_tables);
+
         tx_data
     }
 
@@ -847,17 +884,25 @@ impl CommittedState {
             }
             // A table was removed. Add it back.
             TableRemoved(table_id, table) => {
+                let is_view_table = table.schema.is_view();
                 // We don't need to deal with sub-components.
                 // That is, we don't need to add back indices and such.
                 // Instead, there will be separate pending schema changes like `IndexRemoved`.
                 self.tables.insert(table_id, table);
+
+                // Incase, the table was ephemeral, add it back to that set as well.
+                if is_view_table {
+                    self.ephemeral_tables.insert(table_id);
+                }
             }
             // A table was added. Remove it.
             TableAdded(table_id) => {
                 // We don't need to deal with sub-components.
                 // That is, we don't need to remove indices and such.
                 // Instead, there will be separate pending schema changes like `IndexAdded`.
                 self.tables.remove(&table_id);
+                // Incase, the table was ephemeral, remove it from that set as well.
+                self.ephemeral_tables.remove(&table_id);
             }
             // A table's access was changed. Change back to the old one.
             TableAlterAccess(table_id, access) => {
diff --git a/crates/datastore/src/locking_tx_datastore/datastore.rs b/crates/datastore/src/locking_tx_datastore/datastore.rs
@@ -146,6 +146,8 @@ impl Locking {
         committed_state.build_indexes()?;
         // Figure out where to pick up for each sequence.
         *self.sequence_state.lock() = committed_state.build_sequence_state()?;
+
+        committed_state.collect_ephemeral_tables()?;
         Ok(())
     }
 
diff --git a/crates/datastore/src/locking_tx_datastore/mut_tx.rs b/crates/datastore/src/locking_tx_datastore/mut_tx.rs
@@ -384,6 +384,8 @@ impl MutTxId {
         self.insert_into_st_view_param(view_id, param_columns)?;
         self.insert_into_st_view_column(view_id, return_columns)?;
 
+        self.committed_state_write_lock.ephemeral_tables.insert(table_id);
+
         Ok((view_id, table_id))
     }
 
diff --git a/crates/datastore/src/traits.rs b/crates/datastore/src/traits.rs
@@ -166,6 +166,8 @@ pub enum IsolationLevel {
     Serializable,
 }
 
+pub type EphemeralTables = IntSet<TableId>;
+
 /// A record of all the operations within a transaction.
 ///
 /// Some extra information is embedded here
@@ -191,6 +193,11 @@ pub struct TxData {
     /// `None` implies that `inserts` and `deletes` are both empty,
     /// but `Some` does not necessarily imply that either is non-empty.
     tx_offset: Option<u64>,
+
+    /// Set of ephemeral tables modified in this transaction (only populated when a view is executed).
+    /// These tables do not need to be persisted to disk.
+    /// Every table listed here must appear in either `inserts` or `deletes`.
+    ephermal_tables: Option<EphemeralTables>,
 }
 
 impl TxData {
@@ -226,6 +233,25 @@ impl TxData {
         self.truncates.extend(truncated_tables);
     }
 
+    /// Determines which ephemeral tables were modified in this transaction.
+    ///
+    /// Iterates over the tables updated in this transaction and records those that
+    /// also appear in `all_ephemeral_tables`.  
+    /// `self.ephemeral_tables` remains `None` if no ephemeral tables were modified.
+    pub fn set_ephemeral_tables(&mut self, all_ephemeral_tables: &EphemeralTables) {
+        for tid in self.tables.keys() {
+            if all_ephemeral_tables.contains(tid) {
+                self.ephermal_tables
+                    .get_or_insert_with(EphemeralTables::default)
+                    .insert(*tid);
+            }
+        }
+    }
+
+    pub fn ephermal_tables(&self) -> Option<&EphemeralTables> {
+        self.ephermal_tables.as_ref()
+    }
+
     /// Obtain an iterator over the inserted rows per table.
     pub fn inserts(&self) -> impl Iterator<Item = (&TableId, &Arc<[ProductValue]>)> + '_ {
         self.inserts.iter()
@@ -255,6 +281,17 @@ impl TxData {
         self.deletes.iter()
     }
 
+    /// Obtain an iterator over the deleted rows per table,
+    /// excluding ephemeral tables.
+    pub fn durable_deletes(&self) -> impl Iterator<Item = (&TableId, &Arc<[ProductValue]>)> + '_ {
+        self.deletes.iter().filter(move |(table_id, _)| {
+            !self
+                .ephermal_tables
+                .as_ref()
+                .is_some_and(|etables| etables.contains(*table_id))
+        })
+    }
+
     /// Get the `i`th deleted row for `table_id` if it exists
     pub fn get_ith_delete(&self, table_id: TableId, i: usize) -> Option<&ProductValue> {
         self.deletes.get(&table_id).and_then(|rows| rows.get(i))

Original file line number	Diff line number	Diff line change
`@@ -146,6 +146,8 @@ impl Locking {`
`146`	`146`	`committed_state.build_indexes()?;`
`147`	`147`	`// Figure out where to pick up for each sequence.`
`148`	`148`	`*self.sequence_state.lock() = committed_state.build_sequence_state()?;`
	`149`	`+`
	`150`	`+ committed_state.collect_ephemeral_tables()?;`
`149`	`151`	`Ok(())`
`150`	`152`	`}`
`151`	`153`
Original file line number	Diff line number	Diff line change
`@@ -384,6 +384,8 @@ impl MutTxId {`
`384`	`384`	`self.insert_into_st_view_param(view_id, param_columns)?;`
`385`	`385`	`self.insert_into_st_view_column(view_id, return_columns)?;`
`386`	`386`
	`387`	`+ self.committed_state_write_lock.ephemeral_tables.insert(table_id);`
	`388`	`+`
`387`	`389`	`Ok((view_id, table_id))`
`388`	`390`	`}`
`389`	`391`