Free pages which become empty during a delete

crates/datastore/src/locking_tx_datastore/committed_state.rs

@@ -613,10 +613,12 @@ impl CommittedState {
         pending_schema_changes: ThinVec<PendingSchemaChange>,
         truncates: &mut IntSet<TableId>,
     ) {
+        #[allow(clippy::too_many_arguments)]
         fn delete_rows(
             tx_data: &mut TxData,
             table_id: TableId,
             table: &mut Table,
+            page_pool: &PagePool,
             blob_store: &mut dyn BlobStore,
             row_ptrs_len: usize,
             row_ptrs: impl Iterator<Item = RowPointer>,
@@ -633,7 +635,7 @@ impl CommittedState {
 
                 // TODO: re-write `TxData` to remove `ProductValue`s
                 let pv = table
-                    .delete(blob_store, row_ptr, |row| row.to_product_value())
+                    .delete(page_pool, blob_store, row_ptr, |row| row.to_product_value())
                     .expect("Delete for non-existent row!");
                 deletes.push(pv);
             }
@@ -650,10 +652,11 @@ impl CommittedState {
 
         for (table_id, row_ptrs) in delete_tables {
             match self.get_table_and_blob_store_mut(table_id) {
-                Ok((table, blob_store, ..)) => delete_rows(
+                Ok((table, blob_store, _index_map, page_pool)) => delete_rows(
                     tx_data,
                     table_id,
                     table,
+                    page_pool,
                     blob_store,
                     row_ptrs.len(),
                     row_ptrs.iter(),
@@ -675,6 +678,7 @@ impl CommittedState {
                     tx_data,
                     table_id,
                     &mut table,
+                    &self.page_pool,
                     &mut self.blob_store,
                     row_ptrs.len(),
                     row_ptrs.into_iter(),

crates/datastore/src/locking_tx_datastore/mut_tx.rs

@@ -3451,11 +3451,12 @@ pub(super) fn insert<'a, const GENERATE: bool>(
         is_scheduler_table: tx_table.is_scheduler(),
     };
     let ok = |row_ref| Ok((gen_cols, row_ref, insert_flags));
+    let page_pool = &committed_state.page_pool;
 
     // `CHECK_SAME_ROW = true`, as there might be an identical row already in the tx state.
     // SAFETY: `tx_table.is_row_present(row)` holds as we still haven't deleted the row,
     // in particular, the `write_gen_val_to_col` call does not remove the row.
-    let res = unsafe { tx_table.confirm_insertion::<true>(tx_blob_store, tx_row_ptr, blob_bytes) };
+    let res = unsafe { tx_table.confirm_insertion::<true>(page_pool, tx_blob_store, tx_row_ptr, blob_bytes) };
 
     match res {
         Ok((tx_row_hash, tx_row_ptr)) => {
@@ -3499,7 +3500,7 @@ pub(super) fn insert<'a, const GENERATE: bool>(
                 // - Insert Row A
                 // This is impossible to recover if `Running 2` elides its insert.
                 tx_table
-                    .delete(tx_blob_store, tx_row_ptr, |_| ())
+                    .delete(page_pool, tx_blob_store, tx_row_ptr, |_| ())
                     .expect("Failed to delete a row we just inserted");
 
                 // It's possible that `row` appears in the committed state,
@@ -3528,7 +3529,7 @@ pub(super) fn insert<'a, const GENERATE: bool>(
             let res = unsafe { commit_table.check_unique_constraints(tx_row_ref, |ixs| ixs, is_deleted) };
             if let Err(e) = res {
                 // There was a constraint violation, so undo the insertion.
-                tx_table.delete(tx_blob_store, tx_row_ptr, |_| {});
+                tx_table.delete(page_pool, tx_blob_store, tx_row_ptr, |_| {});
                 return Err(IndexError::from(e).into());
             }
 
@@ -3601,6 +3602,8 @@ impl MutTxId {
         // SAFETY: `tx_table.is_row_present(tx_row_ptr)` holds as we just inserted it.
         let tx_row_ref = unsafe { tx_table.get_row_ref_unchecked(tx_blob_store, tx_row_ptr) };
 
+        let page_pool = &self.committed_state_write_lock.page_pool;
+
         let err = 'error: {
             // This macros can be thought of as a `throw $e` within `'error`.
             // TODO(centril): Get rid of this once we have stable `try` blocks or polonius.
@@ -3665,7 +3668,7 @@ impl MutTxId {
                 // 3. we just inserted `tx_row_ptr` into `tx_table`, so we know it is valid.
                 if unsafe { Table::eq_row_in_page(commit_table, old_ptr, tx_table, tx_row_ptr) } {
                     // SAFETY: `tx_table.is_row_present(tx_row_ptr)` holds, as noted in 3.
-                    unsafe { tx_table.delete_internal_skip_pointer_map(tx_blob_store, tx_row_ptr) };
+                    unsafe { tx_table.delete_internal_skip_pointer_map(page_pool, tx_blob_store, tx_row_ptr) };
                     // SAFETY: `commit_table.is_row_present(old_ptr)` holds, as noted in 2.
                     let row_ref = unsafe { commit_table.get_row_ref_unchecked(commit_blob_store, old_ptr) };
                     return ok(RowRefInsertion::Existed(row_ref));
@@ -3685,7 +3688,7 @@ impl MutTxId {
                 // in particular, the `write_gen_val_to_col` call does not remove the row.
                 // On error, `tx_row_ptr` has already been removed, so don't do it again.
                 let (_, tx_row_ptr) =
-                    unsafe { tx_table.confirm_insertion::<false>(tx_blob_store, tx_row_ptr, blob_bytes) }?;
+                    unsafe { tx_table.confirm_insertion::<false>(page_pool, tx_blob_store, tx_row_ptr, blob_bytes) }?;
 
                 // Delete the old row.
                 del_table.insert(old_ptr);
@@ -3703,7 +3706,8 @@ impl MutTxId {
                 // SAFETY: `tx_table.is_row_present(tx_row_ptr)` and `tx_table.is_row_present(old_ptr)` both hold
                 // as we've deleted neither.
                 // In particular, the `write_gen_val_to_col` call does not remove the row.
-                let tx_row_ptr = unsafe { tx_table.confirm_update(tx_blob_store, tx_row_ptr, old_ptr, blob_bytes) }?;
+                let tx_row_ptr =
+                    unsafe { tx_table.confirm_update(page_pool, tx_blob_store, tx_row_ptr, old_ptr, blob_bytes) }?;
 
                 if let Some(old_commit_del_ptr) = old_commit_del_ptr {
                     // If we have an identical deleted row in the committed state,
@@ -3718,7 +3722,7 @@ impl MutTxId {
                         // It is important that we `confirm_update` first,
                         // as we must ensure that undeleting the row causes no tx state conflict.
                         tx_table
-                            .delete(tx_blob_store, tx_row_ptr, |_| ())
+                            .delete(page_pool, tx_blob_store, tx_row_ptr, |_| ())
                             .expect("Failed to delete a row we just inserted");
 
                         // Undelete.
@@ -3748,7 +3752,7 @@ impl MutTxId {
         // When we reach here, we had an error and we need to revert the insertion of `tx_row_ref`.
         // SAFETY: `tx_table.is_row_present(tx_row_ptr)` holds,
         // as we still haven't deleted the row physically.
-        unsafe { tx_table.delete_internal_skip_pointer_map(tx_blob_store, tx_row_ptr) };
+        unsafe { tx_table.delete_internal_skip_pointer_map(page_pool, tx_blob_store, tx_row_ptr) };
         Err(err)
     }
 
@@ -3770,7 +3774,7 @@ impl MutTxId {
         let (tx_table, tx_blob_store, delete_table) = self
             .tx_state
             .get_table_and_blob_store_or_create_from(table_id, commit_table);
-        let mut rows_removed = tx_table.clear(tx_blob_store);
+        let mut rows_removed = tx_table.clear(&self.committed_state_write_lock.page_pool, tx_blob_store);
 
         // Mark every row in the committed state as deleted.
         for row in commit_table.scan_rows(commit_bs) {
@@ -3796,7 +3800,9 @@ pub(super) fn delete(
             let (table, blob_store) = tx_state
                 .get_table_and_blob_store(table_id)
                 .ok_or(TableError::IdNotFoundState(table_id))?;
-            Ok(table.delete(blob_store, row_pointer, |_| ()).is_some())
+            Ok(table
+                .delete(&committed_state.page_pool, blob_store, row_pointer, |_| ())
+                .is_some())
         }
         SquashedOffset::COMMITTED_STATE => {
             let commit_table = committed_state
@@ -3855,7 +3861,7 @@ impl MutTxId {
         // Do this before actually deleting to drop the borrows on the table.
         // SAFETY: `temp_ptr` is valid because we just inserted it and haven't deleted it since.
         unsafe {
-            tx_table.delete_internal_skip_pointer_map(tx_blob_store, temp_ptr);
+            tx_table.delete_internal_skip_pointer_map(page_pool, tx_blob_store, temp_ptr);
         }
 
         // Delete the found row either by marking (commit table)

crates/datastore/src/locking_tx_datastore/replay.rs

@@ -570,7 +570,7 @@ impl<'cs> ReplayCommittedState<'cs> {
             })
             .collect::<Vec<_>>();
 
-        let (st_sequence, blob_store, ..) = self
+        let (st_sequence, blob_store, _index_map, page_pool) = self
             .get_table_and_blob_store_mut(ST_SEQUENCE_ID)
             .expect("`st_sequence` should exist");
 
@@ -602,7 +602,7 @@ impl<'cs> ReplayCommittedState<'cs> {
                     prev_row_pointer
                 };
 
-                st_sequence.delete(blob_store, row_pointer_to_delete, |_| ())
+                st_sequence.delete(page_pool, blob_store, row_pointer_to_delete, |_| ())
                     .expect("Duplicated `st_sequence` row at `row_pointer_to_delete` should be present in `st_sequence` during fixup");
             }
         }
@@ -646,13 +646,13 @@ impl<'cs> ReplayCommittedState<'cs> {
             .map(|row_ref| row_ref.pointer())
             .collect();
 
-        let (st_event_table, blob_store, ..) = self
+        let (st_event_table, blob_store, _index_map, page_pool) = self
             .get_table_and_blob_store_mut(ST_EVENT_TABLE_ID)
             .expect("`st_event_table` was found above");
 
         for ptr in orphaned_rows {
             st_event_table
-                .delete(blob_store, ptr, |_| ())
+                .delete(page_pool, blob_store, ptr, |_| ())
                 .expect("Orphaned `st_event_table` row at `ptr` should be present in `st_event_table` during fixup");
         }
     }
@@ -1039,12 +1039,12 @@ impl<'cs> ReplayCommittedState<'cs> {
         }
 
         // Get the table for mutation.
-        let (table, blob_store, ..) = self.get_table_and_blob_store_mut(table_id)?;
+        let (table, blob_store, _index_map, page_pool) = self.get_table_and_blob_store_mut(table_id)?;
 
         // We do not need to consider a truncation of `st_table` itself,
         // as if that happens, the database is bricked.
 
-        table.clear(blob_store);
+        table.clear(page_pool, blob_store);
 
         Ok(())
     }

crates/table/benches/page_manager.rs

@@ -300,7 +300,7 @@ fn table_insert_one_row(c: &mut Criterion) {
         let mut ctx = (table, NullBlobStore);
         let ptr = ctx.0.insert(&pool, &mut ctx.1, &val).unwrap().1.pointer();
         let pre = |_, (table, bs): &mut (Table, NullBlobStore)| {
-            table.delete(bs, ptr, |_| ()).unwrap();
+            table.delete(&pool, bs, ptr, |_| ()).unwrap();
         };
         group.bench_function(name, |b| {
             iter_time_with(b, &mut ctx, pre, |_, _, (table, bs)| {
@@ -353,8 +353,8 @@ fn table_delete_one_row(c: &mut Criterion) {
         };
 
         group.bench_function(name, |b| {
-            iter_time_with(b, &mut ctx, insert, |row, _, (table, bs, _)| {
-                table.delete(bs, row, |_| ())
+            iter_time_with(b, &mut ctx, insert, |row, _, (table, bs, pool)| {
+                table.delete(pool, bs, row, |_| ())
             });
         });
     }
@@ -555,13 +555,13 @@ fn insert_num_same<R: IndexedRow>(
         .flatten()
 }
 
-fn clear_all_same<R: IndexedRow>(tbl: &mut Table, index_id: IndexId, val_same: u64) {
+fn clear_all_same<R: IndexedRow>(pool: &PagePool, tbl: &mut Table, index_id: IndexId, val_same: u64) {
     let index = tbl.get_index_by_id(index_id).unwrap();
     let key = R::column_value_from_u64(val_same);
     let key = index.key_from_algebraic_value(&key);
     let ptrs = index.seek_point(&key).collect::<Vec<_>>();
     for ptr in ptrs {
-        tbl.delete(&mut NullBlobStore, ptr, |_| ()).unwrap();
+        tbl.delete(pool, &mut NullBlobStore, ptr, |_| ()).unwrap();
     }
 }
 
@@ -615,7 +615,7 @@ fn index_insert(c: &mut Criterion) {
             &num_rows,
             |b, &num_rows| {
                 let pre = |_, (tbl, _, pool): &mut (Table, NullBlobStore, PagePool)| {
-                    clear_all_same::<R>(tbl, index_id, num_rows);
+                    clear_all_same::<R>(pool, tbl, index_id, num_rows);
                     insert_num_same(pool, tbl, || make_row(num_rows), num_same - 1);
                     make_row(num_rows).to_product()
                 };
@@ -738,11 +738,11 @@ fn index_delete(c: &mut Criterion) {
             &num_rows,
             |b, &num_rows| {
                 let pre = |_, tbl: &mut Table| {
-                    clear_all_same::<R>(tbl, index_id, num_rows);
+                    clear_all_same::<R>(&pool, tbl, index_id, num_rows);
                     insert_num_same(&pool, tbl, || make_row(num_rows), num_same).unwrap()
                 };
                 iter_time_with(b, &mut tbl, pre, |ptr, _, tbl| {
-                    tbl.delete(&mut NullBlobStore, ptr, |_| ())
+                    tbl.delete(&pool, &mut NullBlobStore, ptr, |_| ())
                 });
             },
         );

crates/table/src/eq.rs

@@ -250,14 +250,14 @@ mod test {
         let a0 = product![AlgebraicValue::sum(0, u64::MAX.into())];
         let (_, a0_rr) = table_a.insert(&pool, bs, &a0).unwrap();
         let a0_ptr = a0_rr.pointer();
-        assert!(table_a.delete(bs, a0_ptr, |_| {}).is_some());
+        assert!(table_a.delete(&pool, bs, a0_ptr, |_| {}).is_some());
 
         // Insert u64::ALTERNATING_BIT_PATTERN with tag 0 and then delete it.
         let b0 = 0b01010101_01010101_01010101_01010101_01010101_01010101_01010101_01010101u64;
         let b0 = product![AlgebraicValue::sum(0, b0.into())];
         let (_, b0_rr) = table_b.insert(&pool, bs, &b0).unwrap();
         let b0_ptr = b0_rr.pointer();
-        assert!(table_b.delete(bs, b0_ptr, |_| {}).is_some());
+        assert!(table_b.delete(&pool, bs, b0_ptr, |_| {}).is_some());
 
         // Insert two identical rows `a1` and `b2` into the tables.
         // They should occupy the spaces of the previous rows.

crates/table/src/page.rs

@@ -1185,6 +1185,13 @@ impl Page {
         self.header.fixed.num_rows as usize
     }
 
+    /// Is this page empty, that is, does it contain zero rows?
+    ///
+    /// This method runs in constant time.
+    pub fn is_empty(&self) -> bool {
+        self.num_rows() == 0
+    }
+
     #[cfg(test)]
     /// Use this page's present rows bitvec to compute the number of present rows.
     ///