Add system table for tracking view subscribers (#3482)

# Description of Changes

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

Adds two new system tables for views. One tracks view arguments, the
other tracks view subscribers.

`st_view_arg` generates a unique id for each unique argument
instantiation. This id will act as a foreign key that both read sets and
backing tables can reference.

`st_view_client` is needed so that we can drop views when clients
unsubscribe or disconnect. Note this disconnect logic is not implemented
in this patch.

Eventually both of these tables should be ephemeral. There's no reason
they need to write to the commitlog.

Note also that the schema of a view's backing table has been updated. It
no longer stores a view's argument values but rather the `id` from
`st_view_arg`.

# 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! -->

Tests will be added in the patch that computes read sets

---------

Signed-off-by: joshua-spacetime <[email protected]>
Co-authored-by: Shubham Mishra <[email protected]>

Author: joshua-spacetime

crates/datastore/src/locking_tx_datastore/committed_state.rs

@@ -17,15 +17,15 @@ use crate::{
         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_VAR_IDX, ST_VIEW_ARG_ID, ST_VIEW_ARG_IDX,
     },
     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,
+        ST_CONNECTION_CREDENTIALS_ID, ST_CONNECTION_CREDENTIALS_IDX, ST_VIEW_CLIENT_ID, ST_VIEW_CLIENT_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;
@@ -304,6 +304,8 @@ impl CommittedState {
         self.create_table(ST_VIEW_ID, schemas[ST_VIEW_IDX].clone());
         self.create_table(ST_VIEW_PARAM_ID, schemas[ST_VIEW_PARAM_IDX].clone());
         self.create_table(ST_VIEW_COLUMN_ID, schemas[ST_VIEW_COLUMN_IDX].clone());
+        self.create_table(ST_VIEW_CLIENT_ID, schemas[ST_VIEW_CLIENT_IDX].clone());
+        self.create_table(ST_VIEW_ARG_ID, schemas[ST_VIEW_ARG_IDX].clone());
 
         // Insert the sequences into `st_sequences`
         let (st_sequences, blob_store, pool) =

crates/datastore/src/locking_tx_datastore/datastore.rs

@@ -1252,12 +1252,13 @@ mod tests {
     use crate::system_tables::{
         system_tables, StColumnRow, StConnectionCredentialsFields, StConstraintData, StConstraintFields,
         StConstraintRow, StIndexAlgorithm, StIndexFields, StIndexRow, StRowLevelSecurityFields, StScheduledFields,
-        StSequenceFields, StSequenceRow, StTableRow, StVarFields, StViewFields, ST_CLIENT_NAME, ST_COLUMN_ID,
-        ST_COLUMN_NAME, ST_CONNECTION_CREDENTIALS_ID, ST_CONNECTION_CREDENTIALS_NAME, ST_CONSTRAINT_ID,
+        StSequenceFields, StSequenceRow, StTableRow, StVarFields, StViewArgFields, StViewFields, ST_CLIENT_NAME,
+        ST_COLUMN_ID, ST_COLUMN_NAME, ST_CONNECTION_CREDENTIALS_ID, ST_CONNECTION_CREDENTIALS_NAME, ST_CONSTRAINT_ID,
         ST_CONSTRAINT_NAME, ST_INDEX_ID, ST_INDEX_NAME, ST_MODULE_NAME, ST_RESERVED_SEQUENCE_RANGE,
         ST_ROW_LEVEL_SECURITY_ID, ST_ROW_LEVEL_SECURITY_NAME, ST_SCHEDULED_ID, ST_SCHEDULED_NAME, ST_SEQUENCE_ID,
-        ST_SEQUENCE_NAME, ST_TABLE_NAME, ST_VAR_ID, ST_VAR_NAME, ST_VIEW_COLUMN_ID, ST_VIEW_COLUMN_NAME, ST_VIEW_ID,
-        ST_VIEW_NAME, ST_VIEW_PARAM_ID, ST_VIEW_PARAM_NAME,
+        ST_SEQUENCE_NAME, ST_TABLE_NAME, ST_VAR_ID, ST_VAR_NAME, ST_VIEW_ARG_ID, ST_VIEW_ARG_NAME, ST_VIEW_CLIENT_ID,
+        ST_VIEW_CLIENT_NAME, ST_VIEW_COLUMN_ID, ST_VIEW_COLUMN_NAME, ST_VIEW_ID, ST_VIEW_NAME, ST_VIEW_PARAM_ID,
+        ST_VIEW_PARAM_NAME,
     };
     use crate::traits::{IsolationLevel, MutTx};
     use crate::Result;
@@ -1713,6 +1714,8 @@ mod tests {
             TableRow { id: ST_VIEW_ID.into(), name: ST_VIEW_NAME, ty: StTableType::System, access: StAccess::Public, primary_key: Some(StViewFields::ViewId.into()) },
             TableRow { id: ST_VIEW_PARAM_ID.into(), name: ST_VIEW_PARAM_NAME, ty: StTableType::System, access: StAccess::Public, primary_key: None },
             TableRow { id: ST_VIEW_COLUMN_ID.into(), name: ST_VIEW_COLUMN_NAME, ty: StTableType::System, access: StAccess::Public, primary_key: None },
+            TableRow { id: ST_VIEW_CLIENT_ID.into(), name: ST_VIEW_CLIENT_NAME, ty: StTableType::System, access: StAccess::Public, primary_key: None },
+            TableRow { id: ST_VIEW_ARG_ID.into(), name: ST_VIEW_ARG_NAME, ty: StTableType::System, access: StAccess::Public, primary_key: Some(StViewArgFields::Id.into()) },
 
         ]));
         #[rustfmt::skip]
@@ -1788,6 +1791,14 @@ mod tests {
             ColRow { table: ST_VIEW_COLUMN_ID.into(), pos: 1, name: "col_pos", ty: ColId::get_type() },
             ColRow { table: ST_VIEW_COLUMN_ID.into(), pos: 2, name: "col_name", ty: AlgebraicType::String },
             ColRow { table: ST_VIEW_COLUMN_ID.into(), pos: 3, name: "col_type", ty: AlgebraicType::bytes() },
+
+            ColRow { table: ST_VIEW_CLIENT_ID.into(), pos: 0, name: "view_id", ty: ViewId::get_type() },
+            ColRow { table: ST_VIEW_CLIENT_ID.into(), pos: 1, name: "arg_id", ty: AlgebraicType::U64 },
+            ColRow { table: ST_VIEW_CLIENT_ID.into(), pos: 2, name: "identity", ty: AlgebraicType::U256 },
+            ColRow { table: ST_VIEW_CLIENT_ID.into(), pos: 3, name: "connection_id", ty: AlgebraicType::U128 },
+
+            ColRow { table: ST_VIEW_ARG_ID.into(), pos: 0, name: "id", ty: AlgebraicType::U64 },
+            ColRow { table: ST_VIEW_ARG_ID.into(), pos: 1, name: "bytes", ty: AlgebraicType::bytes() },
         ]));
         #[rustfmt::skip]
         assert_eq!(query.scan_st_indexes()?, map_array([
@@ -1808,6 +1819,10 @@ mod tests {
             IndexRow { id: 15, table: ST_VIEW_ID.into(), col: col(1), name: "st_view_view_name_idx_btree", },
             IndexRow { id: 16, table: ST_VIEW_PARAM_ID.into(), col: col_list![0, 1], name: "st_view_param_view_id_param_pos_idx_btree", },
             IndexRow { id: 17, table: ST_VIEW_COLUMN_ID.into(), col: col_list![0, 1], name: "st_view_column_view_id_col_pos_idx_btree", },
+            IndexRow { id: 18, table: ST_VIEW_CLIENT_ID.into(), col: col_list![0, 1], name: "st_view_client_view_id_arg_id_idx_btree", },
+            IndexRow { id: 19, table: ST_VIEW_CLIENT_ID.into(), col: col_list![2, 3], name: "st_view_client_identity_connection_id_idx_btree", },
+            IndexRow { id: 20, table: ST_VIEW_ARG_ID.into(), col: col(0), name: "st_view_arg_id_idx_btree", },
+            IndexRow { id: 21, table: ST_VIEW_ARG_ID.into(), col: col(1), name: "st_view_arg_bytes_idx_btree", },
         ]));
         let start = ST_RESERVED_SEQUENCE_RANGE as i128 + 1;
         #[rustfmt::skip]
@@ -1819,6 +1834,7 @@ mod tests {
                 SequenceRow { id: 3, table: ST_CONSTRAINT_ID.into(), col_pos: 0, name: "st_constraint_constraint_id_seq", start },
                 SequenceRow { id: 4, table: ST_SCHEDULED_ID.into(), col_pos: 0, name: "st_scheduled_schedule_id_seq", start },
                 SequenceRow { id: 6, table: ST_VIEW_ID.into(), col_pos: 0, name: "st_view_view_id_seq", start },
+                SequenceRow { id: 7, table: ST_VIEW_ARG_ID.into(), col_pos: 0, name: "st_view_arg_id_seq", start },
             ],
             |row| StSequenceRow {
                 allocated: start - 1,
@@ -1843,6 +1859,8 @@ mod tests {
             ConstraintRow { constraint_id: 14, table_id: ST_VIEW_ID.into(), unique_columns: col(1), constraint_name: "st_view_view_name_key", },
             ConstraintRow { constraint_id: 15, table_id: ST_VIEW_PARAM_ID.into(), unique_columns: col_list![0, 1], constraint_name: "st_view_param_view_id_param_pos_key", },
             ConstraintRow { constraint_id: 16, table_id: ST_VIEW_COLUMN_ID.into(), unique_columns: col_list![0, 1], constraint_name: "st_view_column_view_id_col_pos_key", },
+            ConstraintRow { constraint_id: 17, table_id: ST_VIEW_ARG_ID.into(), unique_columns: col(0), constraint_name: "st_view_arg_id_key", },
+            ConstraintRow { constraint_id: 18, table_id: ST_VIEW_ARG_ID.into(), unique_columns: col(1), constraint_name: "st_view_arg_bytes_key", },
             ]));
 
         // Verify we get back the tables correctly with the proper ids...
@@ -2263,6 +2281,10 @@ mod tests {
             IndexRow { id: 15, table: ST_VIEW_ID.into(), col: col(1), name: "st_view_view_name_idx_btree", },
             IndexRow { id: 16, table: ST_VIEW_PARAM_ID.into(), col: col_list![0, 1], name: "st_view_param_view_id_param_pos_idx_btree", },
             IndexRow { id: 17, table: ST_VIEW_COLUMN_ID.into(), col: col_list![0, 1], name: "st_view_column_view_id_col_pos_idx_btree", },
+            IndexRow { id: 18, table: ST_VIEW_CLIENT_ID.into(), col: col_list![0, 1], name: "st_view_client_view_id_arg_id_idx_btree", },
+            IndexRow { id: 19, table: ST_VIEW_CLIENT_ID.into(), col: col_list![2, 3], name: "st_view_client_identity_connection_id_idx_btree", },
+            IndexRow { id: 20, table: ST_VIEW_ARG_ID.into(), col: col(0), name: "st_view_arg_id_idx_btree", },
+            IndexRow { id: 21, table: ST_VIEW_ARG_ID.into(), col: col(1), name: "st_view_arg_bytes_idx_btree", },
             IndexRow { id: seq_start,     table: FIRST_NON_SYSTEM_ID, col: col(0), name: "Foo_id_idx_btree",  },
             IndexRow { id: seq_start + 1, table: FIRST_NON_SYSTEM_ID, col: col(1), name: "Foo_name_idx_btree",  },
             IndexRow { id: seq_start + 2, table: FIRST_NON_SYSTEM_ID, col: col(2), name: "Foo_age_idx_btree",  },

crates/datastore/src/system_tables.rs

@@ -71,6 +71,10 @@ pub const ST_VIEW_ID: TableId = TableId(12);
 pub const ST_VIEW_PARAM_ID: TableId = TableId(13);
 /// The static ID of the table that tracks view columns
 pub const ST_VIEW_COLUMN_ID: TableId = TableId(14);
+/// The static ID of the table that tracks the clients subscribed to each view
+pub const ST_VIEW_CLIENT_ID: TableId = TableId(15);
+/// The static ID of the table that tracks view arguments
+pub const ST_VIEW_ARG_ID: TableId = TableId(16);
 
 pub(crate) const ST_CONNECTION_CREDENTIALS_NAME: &str = "st_connection_credentials";
 pub const ST_TABLE_NAME: &str = "st_table";
@@ -86,6 +90,8 @@ pub(crate) const ST_ROW_LEVEL_SECURITY_NAME: &str = "st_row_level_security";
 pub(crate) const ST_VIEW_NAME: &str = "st_view";
 pub(crate) const ST_VIEW_PARAM_NAME: &str = "st_view_param";
 pub(crate) const ST_VIEW_COLUMN_NAME: &str = "st_view_column";
+pub(crate) const ST_VIEW_CLIENT_NAME: &str = "st_view_client";
+pub(crate) const ST_VIEW_ARG_NAME: &str = "st_view_arg";
 /// Reserved range of sequence values used for system tables.
 ///
 /// Ids for user-created tables will start at `ST_RESERVED_SEQUENCE_RANGE`.
@@ -115,7 +121,7 @@ pub enum SystemTable {
     st_row_level_security,
 }
 
-pub fn system_tables() -> [TableSchema; 14] {
+pub fn system_tables() -> [TableSchema; 16] {
     [
         // The order should match the `id` of the system table, that start with [ST_TABLE_IDX].
         st_table_schema(),
@@ -132,6 +138,8 @@ pub fn system_tables() -> [TableSchema; 14] {
         st_view_schema(),
         st_view_param_schema(),
         st_view_column_schema(),
+        st_view_client_schema(),
+        st_view_arg_schema(),
     ]
 }
 
@@ -174,6 +182,8 @@ pub(crate) const ST_CONNECTION_CREDENTIALS_IDX: usize = 10;
 pub(crate) const ST_VIEW_IDX: usize = 11;
 pub(crate) const ST_VIEW_PARAM_IDX: usize = 12;
 pub(crate) const ST_VIEW_COLUMN_IDX: usize = 13;
+pub(crate) const ST_VIEW_CLIENT_IDX: usize = 14;
+pub(crate) const ST_VIEW_ARG_IDX: usize = 15;
 
 macro_rules! st_fields_enum {
     ($(#[$attr:meta])* enum $ty_name:ident { $($name:expr, $var:ident = $discr:expr,)* }) => {
@@ -240,6 +250,18 @@ st_fields_enum!(enum StViewColumnFields {
     "col_type", ColType = 3,
 });
 // WARNING: For a stable schema, don't change the field names and discriminants.
+st_fields_enum!(enum StViewClientFields {
+    "view_id", ViewId = 0,
+    "arg_id", ArgId = 1,
+    "identity", Identity = 2,
+    "connection_id", ConnectionId = 3,
+});
+// WARNING: For a stable schema, don't change the field names and discriminants.
+st_fields_enum!(enum StViewArgFields {
+    "id", Id = 0,
+    "bytes", Bytes = 1,
+});
+// WARNING: For a stable schema, don't change the field names and discriminants.
 st_fields_enum!(enum StViewParamFields {
     "view_id", ViewId = 0,
     "param_pos", ParamPos = 1,
@@ -376,6 +398,25 @@ fn system_module_def() -> ModuleDef {
         .with_unique_constraint(st_view_param_unique_cols)
         .with_index_no_accessor_name(btree(st_view_param_unique_cols));
 
+    let st_view_client_type = builder.add_type::<StViewClientRow>();
+    builder
+        .build_table(
+            ST_VIEW_CLIENT_NAME,
+            *st_view_client_type.as_ref().expect("should be ref"),
+        )
+        .with_type(TableType::System)
+        .with_index_no_accessor_name(btree([StViewClientFields::ViewId, StViewClientFields::ArgId]))
+        .with_index_no_accessor_name(btree([StViewClientFields::Identity, StViewClientFields::ConnectionId]));
+
+    let st_view_arg_type = builder.add_type::<StViewArgRow>();
+    builder
+        .build_table(ST_VIEW_ARG_NAME, *st_view_arg_type.as_ref().expect("should be ref"))
+        .with_type(TableType::System)
+        .with_auto_inc_primary_key(StViewArgFields::Id)
+        .with_index_no_accessor_name(btree(StViewArgFields::Id))
+        .with_unique_constraint(StViewArgFields::Bytes)
+        .with_index_no_accessor_name(btree(StViewArgFields::Bytes));
+
     let st_index_type = builder.add_type::<StIndexRow>();
     builder
         .build_table(ST_INDEX_NAME, *st_index_type.as_ref().expect("should be ref"))
@@ -475,6 +516,8 @@ fn system_module_def() -> ModuleDef {
     validate_system_table::<StViewFields>(&result, ST_VIEW_NAME);
     validate_system_table::<StViewParamFields>(&result, ST_VIEW_PARAM_NAME);
     validate_system_table::<StViewColumnFields>(&result, ST_VIEW_COLUMN_NAME);
+    validate_system_table::<StViewClientFields>(&result, ST_VIEW_CLIENT_NAME);
+    validate_system_table::<StViewArgFields>(&result, ST_VIEW_ARG_NAME);
 
     result
 }
@@ -514,6 +557,8 @@ lazy_static::lazy_static! {
         m.insert("st_view_view_name_key", ConstraintId(14));
         m.insert("st_view_param_view_id_param_pos_key", ConstraintId(15));
         m.insert("st_view_column_view_id_col_pos_key", ConstraintId(16));
+        m.insert("st_view_arg_id_key", ConstraintId(17));
+        m.insert("st_view_arg_bytes_key", ConstraintId(18));
         m
     };
 }
@@ -540,6 +585,10 @@ lazy_static::lazy_static! {
         m.insert("st_view_view_name_idx_btree", IndexId(15));
         m.insert("st_view_param_view_id_param_pos_idx_btree", IndexId(16));
         m.insert("st_view_column_view_id_col_pos_idx_btree", IndexId(17));
+        m.insert("st_view_client_view_id_arg_id_idx_btree", IndexId(18));
+        m.insert("st_view_client_identity_connection_id_idx_btree", IndexId(19));
+        m.insert("st_view_arg_id_idx_btree", IndexId(20));
+        m.insert("st_view_arg_bytes_idx_btree", IndexId(21));
         m
     };
 }
@@ -555,6 +604,7 @@ lazy_static::lazy_static! {
         m.insert("st_scheduled_schedule_id_seq", SequenceId(4));
         m.insert("st_sequence_sequence_id_seq", SequenceId(5));
         m.insert("st_view_view_id_seq", SequenceId(6));
+        m.insert("st_view_arg_id_seq", SequenceId(7));
         m
     };
 }
@@ -667,6 +717,14 @@ pub fn st_view_column_schema() -> TableSchema {
     st_schema(ST_VIEW_COLUMN_NAME, ST_VIEW_COLUMN_ID)
 }
 
+pub fn st_view_client_schema() -> TableSchema {
+    st_schema(ST_VIEW_CLIENT_NAME, ST_VIEW_CLIENT_ID)
+}
+
+pub fn st_view_arg_schema() -> TableSchema {
+    st_schema(ST_VIEW_ARG_NAME, ST_VIEW_ARG_ID)
+}
+
 /// If `table_id` refers to a known system table, return its schema.
 ///
 /// Used when restoring from a snapshot; system tables are reinstantiated with this schema,
@@ -689,6 +747,8 @@ pub(crate) fn system_table_schema(table_id: TableId) -> Option<TableSchema> {
         ST_VIEW_ID => Some(st_view_schema()),
         ST_VIEW_PARAM_ID => Some(st_view_param_schema()),
         ST_VIEW_COLUMN_ID => Some(st_view_column_schema()),
+        ST_VIEW_CLIENT_ID => Some(st_view_client_schema()),
+        ST_VIEW_ARG_ID => Some(st_view_arg_schema()),
         _ => None,
     }
 }
@@ -863,6 +923,32 @@ pub struct StViewParamRow {
     pub param_type: AlgebraicTypeViaBytes,
 }
 
+/// System table [ST_VIEW_CLIENT_NAME]
+///
+/// | view_id | arg_id | identity | connection_id |
+/// |---------|--------|----------|---------------|
+/// | 1       | 2      | 0x...    | 0x...         |
+#[derive(Debug, Clone, Eq, PartialEq, SpacetimeType)]
+#[sats(crate = spacetimedb_lib)]
+pub struct StViewClientRow {
+    pub view_id: ViewId,
+    pub arg_id: u64,
+    pub identity: IdentityViaU256,
+    pub connection_id: ConnectionIdViaU128,
+}
+
+/// System table [ST_VIEW_ARG_NAME]
+///
+/// | id | bytes   |
+/// |----|---------|
+/// | 1  | <bytes> |
+#[derive(Debug, Clone, Eq, PartialEq, SpacetimeType)]
+#[sats(crate = spacetimedb_lib)]
+pub struct StViewArgRow {
+    pub id: u64,
+    pub bytes: Box<[u8]>,
+}
+
 /// System Table [ST_INDEX_NAME]
 ///
 /// | index_id | table_id | index_name  | index_algorithm            |