@@ -91,6 +91,22 @@ pub struct CommittedState {
9191 /// - system tables: `st_view_sub`, `st_view_arg`
9292 /// - Tables which back views.
9393 pub ( super ) ephemeral_tables : EphemeralTables ,
94+
95+ /// Rows within `st_column` which should be ignored during replay
96+ /// due to having been superseded by a new row representing the same column.
97+ ///
98+ /// During replay, we visit all of the inserts table-by-table, followed by all of the deletes table-by-table.
99+ /// This means that, when multiple columns of a table change type within the same transaction,
100+ /// we see all of the newly-inserted `st_column` rows first, and then later, all of the deleted rows.
101+ /// We may even see inserts into the altered table before seeing the `st_column` deletes!
102+ ///
103+ /// In order to maintain a proper view of the schema of tables during replay,
104+ /// we must remember the old versions of the `st_column` rows when we insert the new ones,
105+ /// so that we can respect only the new versions.
106+ ///
107+ /// We insert into this set during [`Self::replay_insert`] of `st_column` rows
108+ /// and delete from it during [`Self::replay_delete`] of `st_column` rows.
109+ replay_columns_to_ignore : HashSet < RowPointer > ,
94110}
95111
96112impl CommittedState {
@@ -120,6 +136,7 @@ impl MemoryUsage for CommittedState {
120136 table_dropped,
121137 read_sets,
122138 ephemeral_tables,
139+ replay_columns_to_ignore,
123140 } = self ;
124141 // NOTE(centril): We do not want to include the heap usage of `page_pool` as it's a shared resource.
125142 next_tx_offset. heap_usage ( )
@@ -129,6 +146,7 @@ impl MemoryUsage for CommittedState {
129146 + table_dropped. heap_usage ( )
130147 + read_sets. heap_usage ( )
131148 + ephemeral_tables. heap_usage ( )
149+ + replay_columns_to_ignore. heap_usage ( )
132150 }
133151}
134152
@@ -199,6 +217,7 @@ impl CommittedState {
199217 read_sets : <_ >:: default ( ) ,
200218 page_pool,
201219 ephemeral_tables : <_ >:: default ( ) ,
220+ replay_columns_to_ignore : <_ >:: default ( ) ,
202221 }
203222 }
204223
@@ -483,7 +502,7 @@ impl CommittedState {
483502 let ( table, blob_store, _, page_pool) = self . get_table_and_blob_store_mut ( table_id) ?;
484503
485504 // Delete the row.
486- table
505+ let row_ptr = table
487506 . delete_equal_row ( page_pool, blob_store, row)
488507 . map_err ( TableError :: Bflatn ) ?
489508 . ok_or_else ( || anyhow ! ( "Delete for non-existent row when replaying transaction" ) ) ?;
@@ -492,15 +511,30 @@ impl CommittedState {
492511 // A row was removed from `st_table`, so a table was dropped.
493512 // Remove that table from the in-memory structures.
494513 let dropped_table_id = Self :: read_table_id ( row) ;
495- self . tables
496- . remove ( & dropped_table_id)
497- . ok_or_else ( || anyhow ! ( "table {} to remove should exist" , dropped_table_id) ) ?;
514+ // It's safe to ignore the case where we don't have an in-memory structure for the deleted table.
515+ // This can happen if a table is initially empty at the snapshot or its creation,
516+ // and never has any rows inserted into or deleted from it.
517+ self . tables . remove ( & dropped_table_id) ;
518+
498519 // Mark the table as dropped so that when
499520 // processing row deletions for that table later,
500521 // they are simply ignored in (1).
501522 self . table_dropped . insert ( dropped_table_id) ;
502523 }
503524
525+ if table_id == ST_COLUMN_ID {
526+ // We may have reached the corresponding delete to an insert in `st_column`
527+ // as the result of a column-type-altering migration.
528+ // Now that the outdated `st_column` row isn't present any more,
529+ // we can stop ignoring it.
530+ //
531+ // It's also possible that we're deleting this column as the result of a deleted table,
532+ // and that there wasn't any corresponding insert at all.
533+ // If that's the case, `row_ptr` won't be in `self.replay_columns_to_ignore`,
534+ // which is fine.
535+ self . replay_columns_to_ignore . remove ( & row_ptr) ;
536+ }
537+
504538 Ok ( ( ) )
505539 }
506540
@@ -530,39 +564,63 @@ impl CommittedState {
530564 Err ( InsertError :: IndexError ( e) ) => return Err ( IndexError :: UniqueConstraintViolation ( e) . into ( ) ) ,
531565 } ;
532566
533- let row_ptr = row_ref. pointer ( ) ;
534-
535567 if table_id == ST_COLUMN_ID {
536568 // We've made a modification to `st_column`.
537569 // The type of a table has changed, so figure out which.
538570 // The first column in `StColumnRow` is `table_id`.
539- self . st_column_changed ( row, row_ptr) ?;
571+ let row_ptr = row_ref. pointer ( ) ;
572+ let table_id = self . ignore_previous_versions_of_column ( row, row_ptr) ?;
573+ self . st_column_changed ( table_id) ?;
540574 }
541575
542576 Ok ( ( ) )
543577 }
544578
579+ /// Mark all `st_column` rows which refer to the same column as `st_column_row`
580+ /// other than the one at `row_pointer` as outdated
581+ /// by storing them in [`Self::replay_columns_to_ignore`].
582+ ///
583+ /// Returns the ID of the table to which `st_column_row` belongs.
584+ fn ignore_previous_versions_of_column (
585+ & mut self ,
586+ st_column_row : & ProductValue ,
587+ row_ptr : RowPointer ,
588+ ) -> Result < TableId > {
589+ let target_table_id = Self :: read_table_id ( st_column_row) ;
590+ let target_col_id = ColId :: deserialize ( ValueDeserializer :: from_ref ( & st_column_row. elements [ 1 ] ) )
591+ . expect ( "second field in `st_column` should decode to a `ColId`" ) ;
592+
593+ let outdated_st_column_rows = iter_st_column_for_table ( self , & target_table_id. into ( ) ) ?
594+ . filter_map ( |row_ref| {
595+ StColumnRow :: try_from ( row_ref)
596+ . map ( |c| ( c. col_pos == target_col_id && row_ref. pointer ( ) != row_ptr) . then ( || row_ref. pointer ( ) ) )
597+ . transpose ( )
598+ } )
599+ . collect :: < Result < Vec < RowPointer > > > ( ) ?;
600+
601+ for row in outdated_st_column_rows {
602+ self . replay_columns_to_ignore . insert ( row) ;
603+ }
604+
605+ Ok ( target_table_id)
606+ }
607+
545608 /// Refreshes the columns and layout of a table
546609 /// when a `row` has been inserted from `st_column`.
547610 ///
548611 /// The `row_ptr` is a pointer to `row`.
549- fn st_column_changed ( & mut self , row : & ProductValue , row_ptr : RowPointer ) -> Result < ( ) > {
550- let target_table_id = Self :: read_table_id ( row) ;
551- let target_col_id = ColId :: deserialize ( ValueDeserializer :: from_ref ( & row. elements [ 1 ] ) )
552- . expect ( "second field in `st_column` should decode to a `ColId`" ) ;
553-
612+ fn st_column_changed ( & mut self , table_id : TableId ) -> Result < ( ) > {
554613 // We're replaying and we don't have unique constraints yet.
555614 // Due to replay handling all inserts first and deletes after,
556615 // when processing `st_column` insert/deletes,
557616 // we may end up with two definitions for the same `col_pos`.
558617 // Of those two, we're interested in the one we just inserted
559618 // and not the other one, as it is being replaced.
560- let mut columns = iter_st_column_for_table ( self , & target_table_id. into ( ) ) ?
561- . filter_map ( |row_ref| {
562- StColumnRow :: try_from ( row_ref)
563- . map ( |c| ( c. col_pos != target_col_id || row_ref. pointer ( ) == row_ptr) . then ( || c. into ( ) ) )
564- . transpose ( )
565- } )
619+ // `Self::ignore_previous_version_of_column` has marked the old version as ignored,
620+ // so filter only the non-ignored columns.
621+ let mut columns = iter_st_column_for_table ( self , & table_id. into ( ) ) ?
622+ . filter ( |row_ref| self . replay_columns_to_ignore . contains ( & row_ref. pointer ( ) ) )
623+ . map ( |row_ref| StColumnRow :: try_from ( row_ref) . map ( Into :: into) )
566624 . collect :: < Result < Vec < _ > > > ( ) ?;
567625
568626 // Columns in `st_column` are not in general sorted by their `col_pos`,
@@ -571,13 +629,23 @@ impl CommittedState {
571629 columns. sort_by_key ( |col : & ColumnSchema | col. col_pos ) ;
572630
573631 // Update the columns and layout of the the in-memory table.
574- if let Some ( table) = self . tables . get_mut ( & target_table_id ) {
632+ if let Some ( table) = self . tables . get_mut ( & table_id ) {
575633 table. change_columns_to ( columns) . map_err ( TableError :: from) ?;
576634 }
577635
578636 Ok ( ( ) )
579637 }
580638
639+ pub ( super ) fn replay_end_tx ( & mut self ) -> Result < ( ) > {
640+ self . next_tx_offset += 1 ;
641+
642+ if !self . replay_columns_to_ignore . is_empty ( ) {
643+ Err ( anyhow:: anyhow!( "`CommittedState::replay_columns_to_ignore` should be empty at the end of a commit, but found {} entries" , self . replay_columns_to_ignore. len( ) ) . into ( ) )
644+ } else {
645+ Ok ( ( ) )
646+ }
647+ }
648+
581649 /// Assuming that a `TableId` is stored as the first field in `row`, read it.
582650 fn read_table_id ( row : & ProductValue ) -> TableId {
583651 TableId :: deserialize ( ValueDeserializer :: from_ref ( & row. elements [ 0 ] ) )
0 commit comments