alter_table_row_type: write changes to st_column (#3196)

# Description of Changes

`fn alter_table_row_type` now writes the new row type to `st_column`.
This is not a fix for the adding-variants problem, but it does get us
closer.

# API and ABI breaking changes

None

# Expected complexity level and risk

2, system tables stuff is always risky.

# Testing

`test_alter_table_row_type_is_transactional` is amended with assertions.

Author: Centril

crates/datastore/src/locking_tx_datastore/datastore.rs

@@ -3284,7 +3284,13 @@ mod tests {
                 AlgebraicType::sum([("ba", AlgebraicType::U16), ("bb", AlgebraicType::U8)])
             ]
         );
-        commit(&datastore, tx)?;
+        let tx_data = commit(&datastore, tx)?;
+        // Ensure the change has been persisted in the commitlog.
+        let to_product = |col: &ColumnSchema| value_serialize(&StColumnRow::from(col.clone())).into_product().unwrap();
+        let (_, inserts) = tx_data.inserts().find(|(id, _)| **id == ST_COLUMN_ID).unwrap();
+        assert_eq!(&**inserts, [to_product(&columns[1])].as_slice());
+        let (_, deletes) = tx_data.deletes().find(|(id, _)| **id == ST_COLUMN_ID).unwrap();
+        assert_eq!(&**deletes, [to_product(&columns_original[1])].as_slice());
 
         // Check that we can successfully scan using the new schema type post commit.
         let tx = begin_tx(&datastore);

crates/datastore/src/locking_tx_datastore/mut_tx.rs

@@ -216,16 +216,8 @@ impl MutTxId {
 
         // Generate the full definition of the table, with the generated indexes, constraints, sequences...
 
-        // Insert the columns into `st_columns`
-        for col in table_schema.columns() {
-            let row = StColumnRow {
-                table_id: col.table_id,
-                col_pos: col.col_pos,
-                col_name: col.col_name.clone(),
-                col_type: col.col_type.clone().into(),
-            };
-            self.insert_via_serialize_bsatn(ST_COLUMN_ID, &row)?;
-        }
+        // Insert the columns into `st_column`.
+        self.insert_st_column(table_schema.columns())?;
 
         let schedule = table_schema.schedule.clone();
         let mut schema_internal = table_schema;
@@ -279,6 +271,15 @@ impl MutTxId {
         Ok(table_id)
     }
 
+    /// Insert `columns` into `st_column`.
+    fn insert_st_column(&mut self, columns: &[ColumnSchema]) -> Result<()> {
+        columns.iter().try_for_each(|col| {
+            let row: StColumnRow = col.clone().into();
+            self.insert_via_serialize_bsatn(ST_COLUMN_ID, &row)?;
+            Ok(())
+        })
+    }
+
     fn create_table_internal(&mut self, schema: Arc<TableSchema>) {
         // Construct the in memory tables.
         let table_id = schema.table_id;
@@ -319,6 +320,11 @@ impl MutTxId {
         Ok(RowTypeForTable::Arc(self.schema_for_table(table_id)?))
     }
 
+    /// Drops all the columns of `table_id` in `st_column`.
+    fn drop_st_column(&mut self, table_id: TableId) -> Result<()> {
+        self.delete_col_eq(ST_COLUMN_ID, StColumnFields::TableId.col_id(), &table_id.into())
+    }
+
     pub fn drop_table(&mut self, table_id: TableId) -> Result<()> {
         let schema = &*self.schema_for_table(table_id)?;
 
@@ -336,7 +342,7 @@ impl MutTxId {
 
         // Drop the table and their columns
         self.delete_col_eq(ST_TABLE_ID, StTableFields::TableId.col_id(), &table_id.into())?;
-        self.delete_col_eq(ST_COLUMN_ID, StColumnFields::TableId.col_id(), &table_id.into())?;
+        self.drop_st_column(table_id)?;
 
         if let Some(schedule) = &schema.schedule {
             self.delete_col_eq(
@@ -464,11 +470,19 @@ impl MutTxId {
         column_schemas.iter_mut().for_each(|c| c.table_id = table_id);
 
         // Try to change the tables into what we want.
-        let old_column_schemas = tx_table.change_columns_to(column_schemas).map_err(TableError::from)?;
+        let old_column_schemas = tx_table
+            .change_columns_to(column_schemas.clone())
+            .map_err(TableError::from)?;
         // SAFETY: `commit_table` should have a schema identical to that of `tx_table`
         // prior to changing it just now.
         unsafe { commit_table.set_layout_and_schema_to(tx_table) };
 
+        // Update system tables.
+        // We'll simply remove all rows in `st_columns` and then add the new ones.
+        // The datastore takes care of not persisting any no-op delete/inserts to the commitlog.
+        self.drop_st_column(table_id)?;
+        self.insert_st_column(&column_schemas)?;
+
         // Remember the pending change so we can undo if necessary.
         self.push_schema_change(PendingSchemaChange::TableAlterRowType(table_id, old_column_schemas));
 

crates/datastore/src/system_tables.rs

@@ -568,6 +568,17 @@ impl From<StColumnRow> for ColumnSchema {
     }
 }
 
+impl From<ColumnSchema> for StColumnRow {
+    fn from(column: ColumnSchema) -> Self {
+        Self {
+            table_id: column.table_id,
+            col_pos: column.col_pos,
+            col_name: column.col_name,
+            col_type: column.col_type.into(),
+        }
+    }
+}
+
 /// System Table [ST_INDEX_NAME]
 ///
 /// | index_id | table_id | index_name  | index_algorithm            |