Compute read sets for MutTxId via InstanceEnv (#3520)

# Description of Changes

<!-- Please describe your change, mention any related tickets, and so on
here. -->

The `InstanceEnv` will now compute read sets when executing a view and
store them in the `MutTxId`.

These read sets track table scans as well as singular index key scans.
Index key ranges will be tracked in the future, but for now an index
range scan is treated as a full table scan.

These read sets are maintained as part of the `CommittedState`.

TODO: Check write sets against read sets. Re-evaluate views in the case
of overlap and update the read sets.

# API and ABI breaking changes

<!-- If this is an API or ABI breaking change, please apply the
corresponding GitHub label. -->

None

# Expected complexity level and risk

<!--
How complicated do you think these changes are? Grade on a scale from 1
to 5,
where 1 is a trivial change, and 5 is a deep-reaching and complex
change.

This complexity rating applies not only to the complexity apparent in
the diff,
but also to its interactions with existing and future code.

If you answered more than a 2, explain what is complex about the PR,
and what other components it interacts with in potentially concerning
ways. -->

1

# Testing

<!-- Describe any testing you've done, and any testing you'd like your
reviewers to do,
so that you're confident that all the changes work as expected! -->

- [ ] Unit tests

Author: joshua-spacetime

crates/core/src/db/relational_db.rs

@@ -57,7 +57,7 @@ use std::collections::HashSet;
 use std::fmt;
 use std::fs::File;
 use std::io;
-use std::ops::RangeBounds;
+use std::ops::{Bound, RangeBounds};
 use std::path::Path;
 use std::sync::Arc;
 use tokio::sync::watch;
@@ -1353,7 +1353,15 @@ impl RelationalDB {
         prefix_elems: ColId,
         rstart: &[u8],
         rend: &[u8],
-    ) -> Result<(TableId, impl Iterator<Item = RowRef<'a>>), DBError> {
+    ) -> Result<
+        (
+            TableId,
+            Bound<AlgebraicValue>,
+            Bound<AlgebraicValue>,
+            impl Iterator<Item = RowRef<'a>>,
+        ),
+        DBError,
+    > {
         Ok(tx.index_scan_range(index_id, prefix, prefix_elems, rstart, rend)?)
     }
 

crates/core/src/host/instance_env.rs

@@ -11,7 +11,7 @@ use smallvec::SmallVec;
 use spacetimedb_datastore::locking_tx_datastore::state_view::StateView;
 use spacetimedb_datastore::locking_tx_datastore::MutTxId;
 use spacetimedb_lib::{ConnectionId, Identity, Timestamp};
-use spacetimedb_primitives::{ColId, ColList, IndexId, TableId};
+use spacetimedb_primitives::{ColId, ColList, IndexId, TableId, ViewId};
 use spacetimedb_sats::{
     bsatn::{self, ToBsatn},
     buffer::{CountWriter, TeeWriter},
@@ -31,6 +31,7 @@ pub struct InstanceEnv {
     pub tx: TxSlot,
     /// The timestamp the current reducer began running.
     pub start_time: Timestamp,
+    pub view_id: Option<ViewId>,
 }
 
 #[derive(Clone, Default)]
@@ -172,6 +173,7 @@ impl InstanceEnv {
             scheduler,
             tx: TxSlot::default(),
             start_time: Timestamp::now(),
+            view_id: None,
         }
     }
 
@@ -183,6 +185,12 @@ impl InstanceEnv {
     /// Signal to this `InstanceEnv` that a reducer call is beginning.
     pub fn start_reducer(&mut self, ts: Timestamp) {
         self.start_time = ts;
+        self.view_id = None;
+    }
+
+    /// Signal to this `InstanceEnv` that a we're going to execute a view and compute its read set.
+    pub fn start_view(&mut self, view_id: ViewId) {
+        self.view_id = Some(view_id);
     }
 
     fn get_tx(&self) -> Result<impl DerefMut<Target = MutTxId> + '_, GetTxError> {
@@ -375,7 +383,7 @@ impl InstanceEnv {
         let tx = &mut *self.tx.get()?;
 
         // Find all rows in the table to delete.
-        let (table_id, iter) = stdb.index_scan_range(tx, index_id, prefix, prefix_elems, rstart, rend)?;
+        let (table_id, _, _, iter) = stdb.index_scan_range(tx, index_id, prefix, prefix_elems, rstart, rend)?;
         // Re. `SmallVec`, `delete_by_field` only cares about 1 element, so optimize for that.
         let rows_to_delete = iter.map(|row_ref| row_ref.pointer()).collect::<SmallVec<[_; 1]>>();
 
@@ -460,7 +468,11 @@ impl InstanceEnv {
         let tx = &mut *self.get_tx()?;
 
         // Query the row count for id.
-        stdb.table_row_count_mut(tx, table_id).ok_or(NodesError::TableNotFound)
+        stdb.table_row_count_mut(tx, table_id)
+            .ok_or(NodesError::TableNotFound)
+            .inspect(|_| {
+                tx.record_table_scan(self.view_id, table_id);
+            })
     }
 
     #[tracing::instrument(level = "trace", skip_all)]
@@ -484,6 +496,8 @@ impl InstanceEnv {
             &mut bytes_scanned,
         );
 
+        tx.record_table_scan(self.view_id, table_id);
+
         tx.metrics.rows_scanned += rows_scanned;
         tx.metrics.bytes_scanned += bytes_scanned;
 
@@ -508,11 +522,13 @@ impl InstanceEnv {
         let mut bytes_scanned = 0;
 
         // Open index iterator
-        let (_, iter) = stdb.index_scan_range(tx, index_id, prefix, prefix_elems, rstart, rend)?;
+        let (table_id, lower, upper, iter) = stdb.index_scan_range(tx, index_id, prefix, prefix_elems, rstart, rend)?;
 
         // Scan the index and serialize rows to bsatn
         let chunks = ChunkedWriter::collect_iter(pool, iter, &mut rows_scanned, &mut bytes_scanned);
 
+        tx.record_index_scan(self.view_id, table_id, index_id, lower, upper);
+
         tx.metrics.index_seeks += 1;
         tx.metrics.rows_scanned += rows_scanned;
         tx.metrics.bytes_scanned += bytes_scanned;
@@ -648,6 +664,7 @@ mod test {
                 scheduler,
                 tx: TxSlot::default(),
                 start_time: Timestamp::now(),
+                view_id: None,
             },
             runtime,
         ))

crates/datastore/src/locking_tx_datastore/committed_state.rs

@@ -6,15 +6,11 @@ use super::{
     tx_state::{IndexIdMap, PendingSchemaChange, TxState},
     IterByColEqTx,
 };
-use crate::system_tables::{
-    ST_CONNECTION_CREDENTIALS_ID, ST_CONNECTION_CREDENTIALS_IDX, ST_VIEW_COLUMN_ID, ST_VIEW_COLUMN_IDX, ST_VIEW_ID,
-    ST_VIEW_IDX, ST_VIEW_PARAM_ID, ST_VIEW_PARAM_IDX,
-};
 use crate::{
     db_metrics::DB_METRICS,
     error::{DatastoreError, IndexError, TableError},
     execution_context::ExecutionContext,
-    locking_tx_datastore::state_view::iter_st_column_for_table,
+    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,
@@ -25,12 +21,19 @@ use crate::{
     },
     traits::TxData,
 };
+use crate::{
+    locking_tx_datastore::mut_tx::ReadSet,
+    system_tables::{
+        ST_CONNECTION_CREDENTIALS_ID, ST_CONNECTION_CREDENTIALS_IDX, ST_VIEW_COLUMN_ID, ST_VIEW_COLUMN_IDX, ST_VIEW_ID,
+        ST_VIEW_IDX, ST_VIEW_PARAM_ID, ST_VIEW_PARAM_IDX,
+    },
+};
 use anyhow::anyhow;
 use core::{convert::Infallible, ops::RangeBounds};
-use spacetimedb_data_structures::map::{HashSet, IntMap, IntSet};
+use spacetimedb_data_structures::map::{HashMap, HashSet, IntMap, IntSet};
 use spacetimedb_durability::TxOffset;
 use spacetimedb_lib::{db::auth::StTableType, Identity};
-use spacetimedb_primitives::{ColId, ColList, ColSet, IndexId, TableId};
+use spacetimedb_primitives::{ColId, ColList, ColSet, IndexId, TableId, ViewId};
 use spacetimedb_sats::{algebraic_value::de::ValueDeserializer, memory_usage::MemoryUsage, Deserialize};
 use spacetimedb_sats::{AlgebraicValue, ProductValue};
 use spacetimedb_schema::{
@@ -47,6 +50,40 @@ use std::collections::BTreeMap;
 use std::sync::Arc;
 use thin_vec::ThinVec;
 
+type IndexKeyReadSet = HashMap<AlgebraicValue, IntSet<ViewId>>;
+type IndexColReadSet = HashMap<ColList, IndexKeyReadSet>;
+
+#[derive(Default)]
+struct CommittedReadSets {
+    tables: IntMap<TableId, IntSet<ViewId>>,
+    index_keys: IntMap<TableId, IndexColReadSet>,
+}
+
+impl MemoryUsage for CommittedReadSets {
+    fn heap_usage(&self) -> usize {
+        self.tables.heap_usage() + self.index_keys.heap_usage()
+    }
+}
+
+impl CommittedReadSets {
+    /// Record in the [`CommittedState`] that this view scans this table
+    fn view_scans_table(&mut self, view_id: ViewId, table_id: TableId) {
+        self.tables.entry(table_id).or_default().insert(view_id);
+    }
+
+    /// Record in the [`CommittedState`] that this view reads this index `key` for these table `cols`
+    fn view_reads_index_key(&mut self, view_id: ViewId, table_id: TableId, cols: ColList, key: &AlgebraicValue) {
+        self.index_keys
+            .entry(table_id)
+            .or_default()
+            .entry(cols)
+            .or_default()
+            .entry(key.clone())
+            .or_default()
+            .insert(view_id);
+    }
+}
+
 /// Contains the live, in-memory snapshot of a database. This structure
 /// is exposed in order to support tools wanting to process the commit
 /// logs directly. For normal usage, see the RelationalDB struct instead.
@@ -72,6 +109,11 @@ pub struct CommittedState {
     /// We should split `CommittedState` into two types
     /// where one, e.g., `ReplayCommittedState`, has this field.
     table_dropped: IntSet<TableId>,
+    /// We track the read sets for each view in the committed state.
+    /// We check each reducer's write set against these read sets.
+    /// Any overlap will trigger a re-evaluation of the affected view,
+    /// and its read set will be updated accordingly.
+    read_sets: CommittedReadSets,
 }
 
 impl MemoryUsage for CommittedState {
@@ -83,13 +125,15 @@ impl MemoryUsage for CommittedState {
             index_id_map,
             page_pool: _,
             table_dropped,
+            read_sets,
         } = 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()
             + tables.heap_usage()
             + blob_store.heap_usage()
             + index_id_map.heap_usage()
             + table_dropped.heap_usage()
+            + read_sets.heap_usage()
     }
 }
 
@@ -152,6 +196,7 @@ impl CommittedState {
             blob_store: <_>::default(),
             index_id_map: <_>::default(),
             table_dropped: <_>::default(),
+            read_sets: <_>::default(),
             page_pool,
         }
     }
@@ -614,10 +659,13 @@ impl CommittedState {
         tx_data.has_rows_or_connect_disconnect(ctx.reducer_context())
     }
 
-    pub(super) fn merge(&mut self, tx_state: TxState, ctx: &ExecutionContext) -> TxData {
+    pub(super) fn merge(&mut self, tx_state: TxState, read_sets: ViewReadSets, ctx: &ExecutionContext) -> TxData {
         let mut tx_data = TxData::default();
         let mut truncates = IntSet::default();
 
+        // Merge read sets from the `MutTxId` into the `CommittedState`
+        self.merge_read_sets(read_sets);
+
         // First, apply deletes. This will free up space in the committed tables.
         self.merge_apply_deletes(
             &mut tx_data,
@@ -648,6 +696,26 @@ impl CommittedState {
         tx_data
     }
 
+    fn merge_read_set(&mut self, view_id: ViewId, read_set: ReadSet) {
+        for table_id in read_set.tables_scanned() {
+            self.read_sets.view_scans_table(view_id, *table_id);
+        }
+        for (table_id, index_id, key) in read_set.index_keys_scanned() {
+            if let Some(cols) = self
+                .get_schema(*table_id)
+                .map(|table_schema| table_schema.col_list_for_index_id(*index_id))
+            {
+                self.read_sets.view_reads_index_key(view_id, *table_id, cols, key);
+            }
+        }
+    }
+
+    fn merge_read_sets(&mut self, read_sets: ViewReadSets) {
+        for (view_id, read_set) in read_sets {
+            self.merge_read_set(view_id, read_set);
+        }
+    }
+
     fn merge_apply_deletes(
         &mut self,
         tx_data: &mut TxData,

crates/datastore/src/locking_tx_datastore/datastore.rs

@@ -926,6 +926,7 @@ impl MutTx for Locking {
             sequence_state_lock,
             tx_state: TxState::default(),
             lock_wait_time,
+            read_sets: <_>::default(),
             timer,
             ctx,
             metrics,