Fix boostrating ids for relating objects like indexes, caused by cach…

…e invalidation

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

@@ -259,6 +259,13 @@ impl CommittedState {
                 .inc();
         }
 
+        // 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)?;
+        }
+
         Ok(())
     }
 

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

@@ -626,8 +626,8 @@ impl MutProgrammable for Locking {
 mod tests {
     use super::*;
     use crate::db::datastore::system_tables::{
-        StColumnRow, StConstraintRow, StIndexRow, StSequenceRow, StTableRow, ST_COLUMNS_ID, ST_CONSTRAINTS_ID,
-        ST_INDEXES_ID, ST_SEQUENCES_ID,
+        system_tables, StColumnRow, StConstraintRow, StIndexRow, StSequenceRow, StTableRow, ST_COLUMNS_ID,
+        ST_CONSTRAINTS_ID, ST_INDEXES_ID, ST_SEQUENCES_ID,
     };
     use crate::db::datastore::traits::MutTx;
     use crate::db::datastore::Result;
@@ -1068,6 +1068,60 @@ mod tests {
             ConstraintRow { constraint_id: 4, table_id: 3, columns: col(0), constraints: Constraints::primary_key_auto(), constraint_name: "ct_st_indexes_index_id_primary_key_auto" },
             ConstraintRow { constraint_id: 5, table_id: 4, columns: col(0), constraints: Constraints::primary_key_auto(), constraint_name: "ct_st_constraints_constraint_id_primary_key_auto" },
         ]));
+
+        // Verify we get back the tables correctly with the proper ids...
+        let cols = query.scan_st_columns()?;
+        let idx = query.scan_st_indexes()?;
+        let seq = query.scan_st_sequences()?;
+        let ct = query.scan_st_constraints()?;
+
+        for st in system_tables() {
+            let schema = datastore.schema_for_table_mut_tx(&tx, st.table_id).unwrap();
+            assert_eq!(
+                schema.columns().to_vec(),
+                cols.iter()
+                    .filter(|x| x.table_id == st.table_id)
+                    .cloned()
+                    .map(Into::into)
+                    .collect::<Vec<_>>(),
+                "Columns for {}",
+                schema.table_name
+            );
+
+            assert_eq!(
+                schema.indexes,
+                idx.iter()
+                    .filter(|x| x.table_id == st.table_id)
+                    .cloned()
+                    .map(Into::into)
+                    .collect::<Vec<_>>(),
+                "Indexes for {}",
+                schema.table_name
+            );
+
+            assert_eq!(
+                schema.sequences,
+                seq.iter()
+                    .filter(|x| x.table_id == st.table_id)
+                    .cloned()
+                    .map(Into::into)
+                    .collect::<Vec<_>>(),
+                "Sequences for {}",
+                schema.table_name
+            );
+
+            assert_eq!(
+                schema.constraints,
+                ct.iter()
+                    .filter(|x| x.table_id == st.table_id)
+                    .cloned()
+                    .map(Into::into)
+                    .collect::<Vec<_>>(),
+                "Constraints for {}",
+                schema.table_name
+            );
+        }
+
         datastore.rollback_mut_tx_for_test(tx);
         Ok(())
     }

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

@@ -62,11 +62,8 @@ pub(crate) trait StateView {
         self.iter_by_col_range(ctx, table_id, cols, value)
     }
 
-    fn schema_for_table(&self, ctx: &ExecutionContext, table_id: TableId) -> Result<Cow<'_, TableSchema>> {
-        if let Some(schema) = self.get_schema(&table_id) {
-            return Ok(Cow::Borrowed(schema));
-        }
-
+    /// Reads the schema information for the specified `table_id` directly from the database.
+    fn schema_for_table_raw(&self, ctx: &ExecutionContext, table_id: TableId) -> Result<TableSchema> {
         // Look up the table_name for the table in question.
         let st_table_table_id_col = StTableFields::TableId.col_id().into();
         let value: AlgebraicValue = table_id.into();
@@ -149,7 +146,7 @@ pub(crate) trait StateView {
             })
             .collect::<Result<Vec<_>>>()?;
 
-        Ok(Cow::Owned(TableSchema::new(
+        Ok(TableSchema::new(
             table_id,
             table_name,
             columns,
@@ -158,7 +155,20 @@ pub(crate) trait StateView {
             sequences,
             table_type,
             table_access,
-        )))
+        ))
+    }
+
+    /// Reads the schema information for the specified `table_id`, consulting the `cache` first.
+    ///
+    /// If the schema is not found in the cache, the method calls [Self::schema_for_table_raw].
+    ///
+    /// Note: The responsibility of populating the cache is left to the caller.
+    fn schema_for_table(&self, ctx: &ExecutionContext, table_id: TableId) -> Result<Cow<'_, TableSchema>> {
+        if let Some(schema) = self.get_schema(&table_id) {
+            return Ok(Cow::Borrowed(schema));
+        }
+
+        self.schema_for_table_raw(ctx, table_id).map(Cow::Owned)
     }
 }
 

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

@@ -361,7 +361,7 @@ impl From<StTableRow<String>> for ProductValue {
     }
 }
 
-#[derive(Debug, PartialEq, Eq)]
+#[derive(Debug, Clone, PartialEq, Eq)]
 pub struct StColumnRow<Name: AsRef<str>> {
     pub(crate) table_id: TableId,
     pub(crate) col_pos: ColId,
@@ -395,7 +395,18 @@ impl From<StColumnRow<String>> for ProductValue {
     }
 }
 
-#[derive(Debug, PartialEq, Eq)]
+impl From<StColumnRow<String>> for ColumnSchema {
+    fn from(column: StColumnRow<String>) -> Self {
+        Self {
+            table_id: column.table_id,
+            col_pos: column.col_pos,
+            col_name: column.col_name,
+            col_type: column.col_type,
+        }
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
 pub struct StIndexRow<Name: AsRef<str>> {
     pub(crate) index_id: IndexId,
     pub(crate) table_id: TableId,
@@ -452,7 +463,20 @@ impl From<StIndexRow<String>> for ProductValue {
     }
 }
 
-#[derive(Debug, PartialEq, Eq, Clone)]
+impl From<StIndexRow<String>> for IndexSchema {
+    fn from(x: StIndexRow<String>) -> Self {
+        Self {
+            index_id: x.index_id,
+            table_id: x.table_id,
+            index_type: x.index_type,
+            index_name: x.index_name,
+            is_unique: x.is_unique,
+            columns: x.columns,
+        }
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
 pub struct StSequenceRow<Name: AsRef<str>> {
     pub(crate) sequence_id: SequenceId,
     pub(crate) sequence_name: Name,
@@ -514,7 +538,7 @@ impl From<StSequenceRow<String>> for SequenceSchema {
     }
 }
 
-#[derive(Debug, PartialEq, Eq)]
+#[derive(Debug, Clone, PartialEq, Eq)]
 pub struct StConstraintRow<Name: AsRef<str>> {
     pub(crate) constraint_id: ConstraintId,
     pub(crate) constraint_name: Name,
@@ -555,6 +579,18 @@ impl From<StConstraintRow<String>> for ProductValue {
     }
 }
 
+impl From<StConstraintRow<String>> for ConstraintSchema {
+    fn from(x: StConstraintRow<String>) -> Self {
+        Self {
+            constraint_id: x.constraint_id,
+            constraint_name: x.constraint_name,
+            constraints: x.constraints,
+            table_id: x.table_id,
+            columns: x.columns,
+        }
+    }
+}
+
 /// Indicates the kind of module the `program_hash` of a [`StModuleRow`]
 /// describes.
 ///