From 7288bb11e00388e0d956d7579917ef802f6919c7 Mon Sep 17 00:00:00 2001 From: Pere Diaz Bou Date: Mon, 7 Sep 2026 17:25:51 +0200 Subject: [PATCH] core/mvcc: collect checkpoint rows from a pending-key set MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Passive collect walked every live SkipMap key each tick even when almost nothing was still above durable_txid_max. Under load Rule 3 keeps currents in the map, so that walk grew with the working set. Enqueue write-set keys while the commit is still Preparing, collect from those maps, and retire tags only after a durable watermark publish. Recovery seeds the same maps above the replay cutoff. Tests: checkpoint_state_machine collect_* unit tests Measure: EC2 NVMe c=4 A/B tip-gc (45s): TPS 1781ā†’2000, ckpt_max 7.0sā†’3.2s --- .../mvcc/database/checkpoint_state_machine.rs | 181 ++++++++++++------ core/mvcc/database/mod.rs | 64 +++++++ 2 files changed, 189 insertions(+), 56 deletions(-) diff --git a/core/mvcc/database/checkpoint_state_machine.rs b/core/mvcc/database/checkpoint_state_machine.rs index 632e40532ec..d474af36e99 100644 --- a/core/mvcc/database/checkpoint_state_machine.rs +++ b/core/mvcc/database/checkpoint_state_machine.rs @@ -4,9 +4,9 @@ use crate::alloc::{ }; use crate::mvcc::clock::LogicalClock; use crate::mvcc::database::{ - DeleteRowStateMachine, MVTableId, MvStore, Row, RowID, RowKey, RowVersion, SortableIndexKey, - TxTimestampOrID, WalPos, WriteRowStateMachine, MVCC_META_KEY_PERSISTENT_TX_TS_MAX, - MVCC_META_TABLE_NAME, SQLITE_SCHEMA_MVCC_TABLE_ID, + DeleteRowStateMachine, MVTableId, MvStore, Row, RowID, RowKey, RowVersion, TxTimestampOrID, + WalPos, WriteRowStateMachine, MVCC_META_KEY_PERSISTENT_TX_TS_MAX, MVCC_META_TABLE_NAME, + SQLITE_SCHEMA_MVCC_TABLE_ID, }; #[cfg(any(test, injected_yields))] use crate::mvcc::yield_hooks::{ProvidesYieldContext, YieldContext, YieldPointMarker}; @@ -238,8 +238,7 @@ pub struct CheckpointStateMachine, collect_table_cursor: Option, - collect_index_tableid_cursor: Option, - collect_index_key_cursor: Option>, + collect_index_cursor: Option, /// Async driver for `CheckpointState::CompactSequences`. Lazily set /// on first entry to that state; cleared when the driver completes. seq_compact: Option>, @@ -835,8 +834,7 @@ impl CheckpointStateMachine CheckpointStateMachine Result> { - // Invariant: RowID ordering is (table_id, row_id) with table_id ascending. - // Since MV table IDs are negative and sqlite_schema is table_id=-1, iterating - // in reverse visits sqlite_schema first so CREATE/DROP metadata is applied - // before user-table rows in this checkpoint pass. + // Pending keys only; chains still live in `mvstore.rows`. Rev order keeps + // sqlite_schema ahead of user tables. + if self.mvstore.checkpoint_pending_table_rows.is_empty() + && self.collect_table_cursor.is_none() + { + return Ok(None); + } let bounds: (Bound, Bound) = match self.collect_table_cursor.clone() { None => (Bound::Unbounded, Bound::Unbounded), Some(last) => (Bound::Unbounded, Bound::Excluded(last)), }; let mut processed = 0; - for entry in self.mvstore.rows.range(bounds).rev() { + for entry in self + .mvstore + .checkpoint_pending_table_rows + .range(bounds) + .rev() + { let key = entry.key(); tracing::trace!("collecting {key:?}"); self.collect_table_cursor = Some(key.clone()); @@ -1158,7 +1164,14 @@ impl CheckpointStateMachine= COLLECT_PREEMPTION_THRESHOLD { + return Ok(Some(IOCompletions(Completion::new_yield()))); + } + continue; + }; + let row_versions = row_entry.value().read(); for version in self.maybe_get_checkpointable_versions(&row_versions, key.table_id) { let is_delete = version.end().is_some(); @@ -1327,56 +1340,59 @@ impl CheckpointStateMachine Result> { - let outer_bounds: (Bound, Bound) = - match self.collect_index_tableid_cursor { - None => (Bound::Unbounded, Bound::Unbounded), - Some(last) if self.collect_index_key_cursor.is_none() => { - (Bound::Excluded(last), Bound::Unbounded) - } - Some(last) => (Bound::Included(last), Bound::Unbounded), - }; + if self.mvstore.checkpoint_pending_index_rows.is_empty() + && self.collect_index_cursor.is_none() + { + return Ok(None); + } + let bounds: (Bound, Bound) = match self.collect_index_cursor.clone() { + None => (Bound::Unbounded, Bound::Unbounded), + Some(last) => (Bound::Excluded(last), Bound::Unbounded), + }; let mut processed = 0; - for entry in self.mvstore.index_rows.range(outer_bounds) { - let index_id = *entry.key(); + for entry in self.mvstore.checkpoint_pending_index_rows.range(bounds) { + let key = entry.key(); + self.collect_index_cursor = Some(key.clone()); + let index_id = key.table_id; - // Skip destroyed indexes - we won't checkpoint rows for indexes that will be destroyed if self.destroyed_indexes.contains(&index_id) { - self.collect_index_tableid_cursor = Some(index_id); - self.collect_index_key_cursor = None; continue; } - let index_rows_map = entry.value(); - let inner_bounds: (Bound>, Bound>) = - match self.collect_index_key_cursor.clone() { - None => (Bound::Unbounded, Bound::Unbounded), - Some(last) => (Bound::Excluded(last), Bound::Unbounded), - }; - for entry in index_rows_map.range(inner_bounds) { - let versions = entry.value().read(); - self.collect_index_tableid_cursor = Some(index_id); - self.collect_index_key_cursor = Some(entry.key().clone()); - - for version in self.maybe_get_checkpointable_versions(&versions, index_id) { - let is_delete = version.end().is_some(); - if is_delete && !self.table_exists_for_snapshot(index_id) { - continue; - } - - // Only write the row to the B-tree if it is not a delete, or if it is a delete and it exists in - // the database file. - with_mvcc_checkpoint_allocation_site!(CheckpointIndexWriteSet, { - self.index_write_set - .try_push((index_id, version, is_delete))?; - }); + let RowKey::Record(sortable_key) = &key.row_id else { + continue; + }; + let Some(index_map) = self.mvstore.index_rows.get(&index_id) else { + processed += 1; + if processed >= COLLECT_PREEMPTION_THRESHOLD { + return Ok(Some(IOCompletions(Completion::new_yield()))); } + continue; + }; + let Some(versions_entry) = index_map.value().get(sortable_key.as_ref()) else { processed += 1; if processed >= COLLECT_PREEMPTION_THRESHOLD { return Ok(Some(IOCompletions(Completion::new_yield()))); } + continue; + }; + let versions = versions_entry.value().read(); + + for version in self.maybe_get_checkpointable_versions(&versions, index_id) { + let is_delete = version.end().is_some(); + if is_delete && !self.table_exists_for_snapshot(index_id) { + continue; + } + + with_mvcc_checkpoint_allocation_site!(CheckpointIndexWriteSet, { + self.index_write_set + .try_push((index_id, version, is_delete))?; + }); + } + processed += 1; + if processed >= COLLECT_PREEMPTION_THRESHOLD { + return Ok(Some(IOCompletions(Completion::new_yield()))); } - self.collect_index_tableid_cursor = Some(index_id); - self.collect_index_key_cursor = None; } Ok(None) } @@ -1717,6 +1733,8 @@ impl CheckpointStateMachine CheckpointStateMachine>::new_in(crate::alloc::DynAllocator::default()); versions.push(version); - mvstore.rows.insert( - RowID::new(table_id, RowKey::Int(i)), - Arc::new(RwLock::new(versions)), - ); + let row_id = RowID::new(table_id, RowKey::Int(i)); + mvstore + .rows + .insert(row_id.clone(), Arc::new(RwLock::new(versions))); + mvstore + .note_checkpoint_pending_row(&row_id, 5) + .expect("note pending row"); } // The first chunk fills up before the scan finishes, so it must yield. @@ -3665,9 +3688,13 @@ mod tests { let row_count = COLLECT_PREEMPTION_THRESHOLD + 10; for i in 0..row_count as i64 { let (key, version) = index_row_version(index_id, "k", i, 1, Some(5), None, false); - mvstore + let (canonical_key, _) = mvstore .insert_index_version(index_id, key, version) .unwrap(); + let row_id = RowID::new(index_id, RowKey::Record(canonical_key)); + mvstore + .note_checkpoint_pending_row(&row_id, 5) + .expect("note pending index row"); } let first = checkpoint.collect_index_rows().unwrap(); @@ -3680,6 +3707,48 @@ mod tests { assert_eq!(checkpoint.index_write_set.len(), row_count); } + #[test] + fn collect_table_rows_skips_rows_not_in_pending_set() { + let db = MvccTestDbNoConn::new(); + let conn = db.connect(); + let mvstore = db.get_mvcc_store(); + let pager = conn.pager.load().clone(); + let mut checkpoint = CheckpointStateMachine::new( + pager, + mvstore.clone(), + conn.clone(), + true, + conn.get_sync_mode(), + crate::MAIN_DB_ID, + CheckpointMode::Truncate { + upper_bound_inclusive: None, + }, + ); + + let table_id = MVTableId::from(-2); + for i in 0..50 { + let version = committed_table_row_version(table_id, i); + let mut versions = + as crate::alloc::TursoVecInExt< + RowVersion, + crate::alloc::DynAllocator, + >>::new_in(crate::alloc::DynAllocator::default()); + versions.push(version); + mvstore.rows.insert( + RowID::new(table_id, RowKey::Int(i)), + Arc::new(RwLock::new(versions)), + ); + } + let pending = RowID::new(table_id, RowKey::Int(7)); + mvstore + .note_checkpoint_pending_row(&pending, 5) + .expect("note pending row"); + + while checkpoint.collect_table_rows().unwrap().is_some() {} + assert_eq!(checkpoint.write_set.len(), 1); + assert_eq!(checkpoint.write_set[0].0.row.id.row_id, RowKey::Int(7)); + } + #[test] fn gc_checkpointed_table_versions_preempts_on_large_scan() { let db = MvccTestDbNoConn::new(); diff --git a/core/mvcc/database/mod.rs b/core/mvcc/database/mod.rs index 375bfbec628..72ed97ed1f9 100644 --- a/core/mvcc/database/mod.rs +++ b/core/mvcc/database/mod.rs @@ -3124,6 +3124,13 @@ impl StateTransition for CommitStat return Err(LimboError::SchemaConflict); } tracing::trace!("prepare_tx(tx_id={}, end_ts={})", self.tx_id, end_ts); + // Note while still Preparing: once Committed, a younger finalize can + // raise last_committed above end_ts and a checkpoint could publish a + // watermark that covers keys it never collected. + mvcc_store.note_checkpoint_pending( + end_ts, + tx.write_set.lock().iter().map(|(id, _)| id), + )?; /* In order to implement serializability, we need the following steps: ** ** 1. Validate if all read versions are still visible by inspecting the read_set @@ -4352,6 +4359,14 @@ pub struct MvStore /// a mismatch, since a key inserted at or behind an already-positioned /// finger would otherwise be skipped (#7578). index_rows_epoch: AtomicU64, + /// Keys still owed to Passive/Truncate collect, tagged with the highest + /// commit timestamp that touched them. Collect walks this map instead of + /// all of `rows`. Enqueue while Preparing; retire only after a durable + /// watermark at or above the tag is published. + checkpoint_pending_table_rows: SkipMap, + /// Index counterpart of `checkpoint_pending_table_rows` (index id in + /// `RowID::table_id`, index key in `RowID::row_id`). + checkpoint_pending_index_rows: SkipMap, txs: SkipMap, BasicComparator, A>, /// Final state for removed transactions. Readers may still race with stale TxID /// references in row versions after a transaction is removed from `txs`. @@ -4592,6 +4607,8 @@ impl MvStore { table_id_to_rootpage, index_rows: SkipMap::new_in(alloc.clone()), index_rows_epoch: AtomicU64::new(0), + checkpoint_pending_table_rows: SkipMap::new_in(alloc.clone()), + checkpoint_pending_index_rows: SkipMap::new_in(alloc.clone()), txs: SkipMap::new_in(alloc.clone()), finalized_tx_states: SkipMap::new_in(alloc.clone()), alloc, @@ -7726,6 +7743,46 @@ impl MvStore { snapshot_ts } + /// Enqueue `keys` for collect, tagged with `end_ts`. Call only while the + /// writer is still `Preparing` (see call site at prepare_tx). + pub(crate) fn note_checkpoint_pending<'a>( + &self, + end_ts: u64, + keys: impl Iterator, + ) -> Result<(), TryReserveError> { + for key in keys { + self.note_checkpoint_pending_row(key, end_ts)?; + } + Ok(()) + } + + pub(crate) fn note_checkpoint_pending_row( + &self, + key: &RowID, + commit_ts: u64, + ) -> Result<(), TryReserveError> { + let pending = match key.row_id { + RowKey::Int(_) => &self.checkpoint_pending_table_rows, + RowKey::Record(_) => &self.checkpoint_pending_index_rows, + }; + pending.try_compare_insert(key.clone(), commit_ts, |queued| *queued < commit_ts)?; + Ok(()) + } + + /// Drop pending keys tagged at or below a published durable watermark. + pub(crate) fn retire_checkpoint_pending_through(&self, durable_through: u64) { + for pending in [ + &self.checkpoint_pending_table_rows, + &self.checkpoint_pending_index_rows, + ] { + for entry in pending.iter() { + if *entry.value() <= durable_through { + entry.remove(); + } + } + } + } + pub(crate) fn uses_passive_checkpoint(&self) -> bool { self.experimental_mvcc_passive_checkpoint } @@ -8733,6 +8790,9 @@ impl MvStore { /// Passive sequence compaction: record end-stamped deletes instead of inline B-tree purge. pub fn seqcompact_commit_delete(&self, rowid: RowID, num_cols: usize, end_ts: u64) { + if self.note_checkpoint_pending_row(&rowid, end_ts).is_err() { + return; + } loop { let Ok(row_versions) = self.get_or_create_table_row_versions(rowid.clone()) else { return; @@ -9805,6 +9865,7 @@ impl MvStore { if commit_ts <= replay_cutoff_ts { continue; } + self.note_checkpoint_pending_row(&rowid, commit_ts)?; let is_schema_row = rowid.table_id == SQLITE_SCHEMA_MVCC_TABLE_ID; if is_schema_row { let record = ImmutableRecordRef::from_bin_record(row.payload()); @@ -9930,6 +9991,7 @@ impl MvStore { if commit_ts <= replay_cutoff_ts { continue; } + self.note_checkpoint_pending_row(&rowid, commit_ts)?; if self.table_id_to_rootpage.get(&rowid.table_id).is_none() { // See comment in UpsertTableRow: old logs may have data rows // serialized before the schema INSERT that registers the table_id. @@ -10054,6 +10116,7 @@ impl MvStore { if commit_ts <= replay_cutoff_ts { continue; } + self.note_checkpoint_pending_row(&rowid, commit_ts)?; let version_id = self.get_version_id(); let row_version = RowVersion { id: version_id, @@ -10080,6 +10143,7 @@ impl MvStore { if commit_ts <= replay_cutoff_ts { continue; } + self.note_checkpoint_pending_row(&rowid, commit_ts)?; let RowKey::Record(sortable_key) = rowid.row_id.clone() else { panic!("Index writes must be to a record"); };