Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
19 commits
Select commit Hold shift + click to select a range
e394bbc
Represent absent pages in `Table` and snapshots
gefjon Aug 20, 2026
da456e7
Free pages which become empty during a delete
gefjon Aug 20, 2026
592ea3b
clippy, and fix (remove) broken benchmarks
gefjon Aug 21, 2026
81fca19
Merge branch 'phoebe/free-empty-pages/snapshot-format-sentinel' into …
gefjon Aug 21, 2026
159dd5d
fix compliation errors in datastore crate
gefjon Aug 21, 2026
f0cc92a
Don't attempt to fsync or count non-pages in snapshot
gefjon Aug 21, 2026
a37754c
Merge branch 'phoebe/free-empty-pages/snapshot-format-sentinel' into …
gefjon Aug 21, 2026
f21f158
Merge remote-tracking branch 'origin/master' into phoebe/free-empty-p…
gefjon Aug 21, 2026
225560b
Merge branch 'phoebe/free-empty-pages/snapshot-format-sentinel' into …
gefjon Aug 21, 2026
78dfcfc
Clippy
gefjon Aug 21, 2026
aef3283
Merge branch 'master' into phoebe/free-empty-pages/snapshot-format-se…
bfops Aug 21, 2026
bda0156
Remove methods that will be removed from next patch
gefjon Aug 21, 2026
adbf8a7
Merge branch 'phoebe/free-empty-pages/snapshot-format-sentinel' into …
gefjon Aug 21, 2026
524f690
Count absent pages during snapshot replay towards a different metric
gefjon Aug 22, 2026
9390295
Count absent pages during snapshot replay towards a different metric
gefjon Aug 22, 2026
f9b2368
Merge remote-tracking branch 'origin/master' into phoebe/free-empty-p…
gefjon Aug 22, 2026
57034d1
Merge branch 'phoebe/free-empty-pages/snapshot-format-sentinel' into …
gefjon Aug 22, 2026
c5c62e1
Merge remote-tracking branch 'origin/master' into phoebe/free-empty-p…
gefjon Sep 2, 2026
1d758fb
Delete unused method
gefjon Sep 2, 2026
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 6 additions & 2 deletions crates/datastore/src/locking_tx_datastore/committed_state.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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>,
Expand All @@ -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);
}
Expand All @@ -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(),
Expand All @@ -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(),
Expand Down
28 changes: 17 additions & 11 deletions crates/datastore/src/locking_tx_datastore/mut_tx.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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)) => {
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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());
}

Expand Down Expand Up @@ -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.
Expand Down Expand Up @@ -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));
Expand All @@ -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);
Expand All @@ -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,
Expand All @@ -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.
Expand Down Expand Up @@ -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)
}

Expand All @@ -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) {
Expand All @@ -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
Expand Down Expand Up @@ -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)
Expand Down
12 changes: 6 additions & 6 deletions crates/datastore/src/locking_tx_datastore/replay.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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");

Expand Down Expand Up @@ -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");
}
}
Expand Down Expand Up @@ -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");
}
}
Expand Down Expand Up @@ -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(())
}
Expand Down
16 changes: 8 additions & 8 deletions crates/table/benches/page_manager.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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)| {
Expand Down Expand Up @@ -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, |_| ())
});
});
}
Expand Down Expand Up @@ -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();
}
}

Expand Down Expand Up @@ -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()
};
Expand Down Expand Up @@ -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, |_| ())
});
},
);
Expand Down
4 changes: 2 additions & 2 deletions crates/table/src/eq.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
7 changes: 7 additions & 0 deletions crates/table/src/page.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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.
///
Expand Down
Loading
Loading