From 89712af9bff4fd8914066350252f7bd96cb4eb9a Mon Sep 17 00:00:00 2001 From: Aljoscha Krettek Date: Tue, 28 Jul 2026 19:55:38 +0000 Subject: [PATCH] adapter: let the group committer write at a caller-chosen timestamp The group committer picks a timestamp from the oracle, which is right for a write whose diffs are valid whenever they land. A write computed against a snapshot is not like that: its diffs are only correct at the frontier they were read at, so the timestamp has to come from the caller. `TableWriteCmd::TimestampedWrite` carries that timestamp and gets exactly one attempt. An `InvalidUppers` means another writer holds the upper at or past it, so the snapshot the diffs were computed from is stale and the answer is `TimestampPassed`, never a retry at a fresh timestamp. Retrying stale diffs later is precisely the lost update this exists to prevent. The write also names the `GlobalId` it was computed against, and the committer refuses it if the table's latest generation has moved on, which is how a concurrent `ALTER TABLE` is caught rather than applied to rows of the wrong arity. Both txns-shard write paths now share `attempt_write_to_txns`, so the catalog upper advance, the metric, the `InvalidUppers` classification and the oracle apply exist once. The two differ only in what they do with a conflict, which is the difference worth seeing. The result travels back through a responder whose `Drop` reports `Indeterminate`. Without that, a dropped sender on shutdown would panic the waiting session task instead of failing its statement. Callers arrive later in the stack. Until they do, a `dead_code` allowance on the unreached items keeps the crate warning-free, and it goes away with the commit that adds them. --- src/adapter/src/coord.rs | 2 + src/adapter/src/coord/appends.rs | 482 +++++++++++++++++++---- src/adapter/src/coord/message_handler.rs | 6 + src/adapter/src/session.rs | 5 + 4 files changed, 418 insertions(+), 77 deletions(-) diff --git a/src/adapter/src/coord.rs b/src/adapter/src/coord.rs index 6356d9d3ed0b9..66650756c64ef 100644 --- a/src/adapter/src/coord.rs +++ b/src/adapter/src/coord.rs @@ -362,6 +362,8 @@ pub enum Message { responses: Vec, /// Statement executions associated with this commit. statement_logging_ids: Vec, + /// Frontend-sequenced writes to complete after local timestamp bookkeeping. + internal_results: Vec, /// The applied write timestamp. write_ts: Timestamp, }, diff --git a/src/adapter/src/coord/appends.rs b/src/adapter/src/coord/appends.rs index a0c482c179938..732e50fd1ff3d 100644 --- a/src/adapter/src/coord/appends.rs +++ b/src/adapter/src/coord/appends.rs @@ -164,12 +164,98 @@ pub(crate) enum BuiltinTableUpdateSource { Background(oneshot::Sender<()>), } +/// Result of a write submitted by frontend sequencing. +// The read-then-write path that submits these writes lands later in this +// stack, this attribute goes away with it. +#[allow(dead_code)] +#[derive(Debug, Clone)] +pub enum WriteResult { + /// The write committed at this timestamp. + Success { timestamp: Timestamp }, + /// The requested timestamp was no longer eligible. + TimestampPassed { + target_timestamp: Timestamp, + next_eligible_timestamp: Timestamp, + }, + /// The write was canceled before it entered the committer. + Canceled, + /// The coordinator cannot accept writes. + ReadOnly, + /// The target table was dropped or changed after planning. + TargetChanged, + /// The committer shut down with the write's outcome unknown. + Indeterminate, +} + +/// Delivers an internal write result, including on task shutdown. +/// +/// The `Drop` impl is load-bearing. The session task waiting on the other end +/// `expect`s a reply, so a `oneshot::Sender` that is dropped silently would +/// panic that session on coordinator shutdown or on a dead group-committer +/// task. Reporting [`WriteResult::Indeterminate`] instead lets the session +/// report an error. +#[derive(Debug)] +pub struct InternalWriteResponder { + tx: Option>, +} + +impl InternalWriteResponder { + // The read-then-write path that uses this lands later in this stack, this + // attribute goes away with it. + #[allow(dead_code)] + pub(crate) fn new(tx: oneshot::Sender) -> Self { + Self { tx: Some(tx) } + } + + pub(crate) fn send(mut self, result: WriteResult) { + if let Some(tx) = self.tx.take() { + let _ = tx.send(result); + } + } +} + +impl Drop for InternalWriteResponder { + fn drop(&mut self) { + if let Some(tx) = self.tx.take() { + let _ = tx.send(WriteResult::Indeterminate); + } + } +} + /// Where to deliver the result of a [`PendingWriteTxn::User`] write. #[derive(Debug)] pub(crate) enum UserWriteResponder { /// Session-bound write. The coordinator retires the session's /// `ExecuteContext` once the write commits. Session(PendingTxn), + /// Frontend-sequenced blind write. + // The read-then-write path that uses this lands later in this stack, this + // attribute goes away with it. + #[allow(dead_code)] + Internal { + conn_id: ConnectionId, + /// The table the diffs were computed against, item id and the generation + /// current at that time. Group commit refuses the write if the table's + /// latest generation has moved on. + target: WriteTarget, + result: InternalWriteResponder, + }, +} + +/// A write's target table, pinned to one generation of it. +#[derive(Debug, Clone, Copy)] +pub(crate) struct WriteTarget { + pub(crate) item_id: CatalogItemId, + pub(crate) global_id: GlobalId, +} + +impl UserWriteResponder { + pub(crate) fn conn_id(&self) -> &ConnectionId { + match self { + UserWriteResponder::Session(pending) => pending.ctx.session().conn_id(), + UserWriteResponder::Internal { conn_id, .. } => conn_id, + } + } } /// A pending write transaction that will be committing during the next group commit. @@ -209,6 +295,10 @@ impl PendingWriteTxn { pub(crate) enum TableWriteCmd { GroupCommit(GroupCommitRequest), + // The read-then-write path that uses this lands later in this stack, this + // attribute goes away with it. + #[allow(dead_code)] + TimestampedWrite(TimestampedWriteRequest), Register { tables: Vec, result: oneshot::Sender, @@ -219,6 +309,21 @@ pub(crate) enum TableWriteCmd { }, } +/// An OCC write whose diffs are valid only at `target_timestamp`. +/// +/// The `GlobalId`s reach the table write worker unvalidated, so a submitter +/// must resolve them against the current catalog on the coordinator loop and +/// send with no await in between. Channel order then keeps the append ahead of +/// any `Forget` for the same table. A non-empty append for a table whose write +/// handle is already gone trips an assert in the storage controller and takes +/// the process down. +pub(crate) struct TimestampedWriteRequest { + pub(crate) appends: Vec<(GlobalId, Vec)>, + pub(crate) target_timestamp: Timestamp, + pub(crate) result: InternalWriteResponder, + pub(crate) span: Span, +} + /// A group commit staged on the coordinator loop for the [`GroupCommitter`]. pub(crate) struct GroupCommitRequest { /// Appends resolved to their latest [`GlobalId`]. Empty for a keepalive. @@ -226,6 +331,7 @@ pub(crate) struct GroupCommitRequest { responses: Vec, statement_logging_ids: Vec, notifies: Vec>, + internal_results: Vec, write_locks: GroupCommitWriteLocks, /// In-progress permits held until the commit is applied. permits: Vec, @@ -240,6 +346,7 @@ impl GroupCommitRequest { responses, statement_logging_ids, notifies, + internal_results, write_locks, permits, contains_internal_system_write, @@ -249,6 +356,7 @@ impl GroupCommitRequest { self.responses.extend(responses); self.statement_logging_ids.extend(statement_logging_ids); self.notifies.extend(notifies); + self.internal_results.extend(internal_results); self.write_locks.extend(write_locks); self.permits.extend(permits); self.contains_internal_system_write |= contains_internal_system_write; @@ -270,6 +378,17 @@ pub(crate) struct GroupCommitter { max_attempts: ConfigValHandle, } +/// Outcome of one txns-shard write attempt. +enum TxnsWriteAttempt { + /// The write landed and the oracle has applied its timestamp. + Applied, + /// Another writer holds the upper at or past the attempted timestamp. The + /// write did not land. + UpperConflict, + /// The table write worker is gone, so the outcome is unknown. + WorkerGone, +} + impl GroupCommitter { async fn run(mut self) { while let Some(cmd) = self.rx.recv().await { @@ -285,6 +404,17 @@ impl GroupCommitter { ControlFlow::Break(()) => return, } } + TableWriteCmd::TimestampedWrite(request) => { + let span = request.span.clone(); + if self + .commit_timestamped(request) + .instrument(span) + .await + .is_break() + { + return; + } + } TableWriteCmd::Register { tables, result } => { let Some(write_ts) = self .write_to_txns(None, |ts, _advance_to| { @@ -312,7 +442,85 @@ impl GroupCommitter { } } - /// Writes at a fresh oracle timestamp and applies a successful write to the oracle. + /// Attempts an OCC write exactly once at its requested timestamp. + /// + /// A conflict is reported as `TimestampPassed` and is the caller's to + /// resolve with a new snapshot. Retrying the same diffs at a fresh + /// timestamp would apply a mutation to state it was not computed from. + /// + /// What [`Self::commit`] does that this skips, and why that is safe: + /// + /// * The wall-clock throttle. `target_timestamp` is the caller's to choose, + /// and it must not run the write timeline ahead of the clock. + /// * A [`GroupCommitPermit`]. The caller bounds how many of these are in + /// flight, and that is the backpressure for this path. + /// * Merging queued commits. There is nothing to merge into: these diffs + /// are valid at this one timestamp, so they cannot share a timestamp with + /// another write. + /// * Write locks. The point of OCC is to detect a conflicting write after + /// the fact, through the timestamp, rather than to exclude it. + /// + /// `Break` means the table worker shut down. + async fn commit_timestamped(&self, request: TimestampedWriteRequest) -> ControlFlow<(), ()> { + let TimestampedWriteRequest { + appends, + target_timestamp, + result, + span: _, + } = request; + + let oracle_write_ts = self.oracle.peek_write_ts().await; + if target_timestamp <= oracle_write_ts { + result.send(WriteResult::TimestampPassed { + target_timestamp, + next_eligible_timestamp: oracle_write_ts.step_forward(), + }); + return ControlFlow::Continue(()); + } + + let write_ts = WriteTimestamp { + timestamp: target_timestamp, + advance_to: target_timestamp.step_forward(), + }; + match self + .attempt_write_to_txns( + &write_ts, + Some(&self.metrics.append_table_duration_seconds), + |ts, advance_to| self.table_write_handle.append(ts, advance_to, appends), + ) + .await + { + TxnsWriteAttempt::Applied => {} + TxnsWriteAttempt::UpperConflict => { + result.send(WriteResult::TimestampPassed { + target_timestamp, + next_eligible_timestamp: write_ts.advance_to, + }); + return ControlFlow::Continue(()); + } + TxnsWriteAttempt::WorkerGone => { + warn!("table write worker gone with a timestamped write outstanding"); + return ControlFlow::Break(()); + } + } + + if self + .internal_cmd_tx + .send(Message::GroupCommitApplied { + responses: Vec::new(), + statement_logging_ids: Vec::new(), + internal_results: vec![result], + write_ts: target_timestamp, + }) + .is_err() + { + warn!("coordinator shut down before a timestamped write could be finalized"); + } + ControlFlow::Continue(()) + } + + /// Writes at a fresh oracle timestamp, retrying an upper conflict at a new + /// timestamp, and applies a successful write to the oracle. /// /// Returns `None` when the table write worker shuts down. async fn write_to_txns( @@ -322,7 +530,7 @@ impl GroupCommitter { ) -> Option { // Persistent conflicts indicate an unexpected writer. Halt instead of spinning forever. let mut attempt = 0; - let write_ts = loop { + loop { let max_attempts = self.max_attempts.get().max(1); if attempt >= max_attempts { halt!( @@ -332,26 +540,16 @@ impl GroupCommitter { attempt += 1; let write_ts = self.oracle.write_ts().await; - // A post-fence retry has an advance frontier above this handle's stale upper, so this - // reaches Persist and observes the fence. - let catalog_upper_start = Instant::now(); - self.catalog_upper - .advance_upper(write_ts.advance_to) + // A post-fence retry has an advance frontier above this handle's stale upper, so the + // advance inside reaches Persist and observes the fence. + match self + .attempt_write_to_txns(&write_ts, op_duration_metric, |ts, advance_to| { + op(ts, advance_to) + }) .await - .unwrap_or_terminate("unable to advance catalog upper"); - self.metrics - .group_commit_catalog_upper_seconds - .observe(catalog_upper_start.elapsed().as_secs_f64()); - - let op_start = Instant::now(); - let op_res = op(write_ts.timestamp, write_ts.advance_to).await; - if let Some(metric) = op_duration_metric { - metric.observe(op_start.elapsed().as_secs_f64()); - } - - match op_res { - Ok(Ok(())) => break write_ts, - Ok(Err(StorageError::InvalidUppers(_))) => { + { + TxnsWriteAttempt::Applied => return Some(write_ts), + TxnsWriteAttempt::UpperConflict => { warn!( write_ts = %write_ts.timestamp, attempt, @@ -359,24 +557,57 @@ impl GroupCommitter { ); continue; } - Ok(Err(other)) => { - Err::<(), _>(other).unwrap_or_terminate("cannot fail to write to txns shard"); - unreachable!("unwrap_or_terminate does not return on Err"); - } - Err(_recv) => { + TxnsWriteAttempt::WorkerGone => { // The outcome is indeterminate. Stop before processing more writes. warn!("table write worker gone (process shutting down), winding down"); return None; } } - }; + } + } + + /// Runs `op` against the txns shard once, at `write_ts`. + /// + /// Advancing the catalog upper first keeps the catalog readable at the + /// oracle read timestamp. A write that lands is applied to the oracle + /// before this returns. + async fn attempt_write_to_txns( + &self, + write_ts: &WriteTimestamp, + op_duration_metric: Option<&prometheus::Histogram>, + op: impl FnOnce(Timestamp, Timestamp) -> oneshot::Receiver>, + ) -> TxnsWriteAttempt { + let catalog_upper_start = Instant::now(); + self.catalog_upper + .advance_upper(write_ts.advance_to) + .await + .unwrap_or_terminate("unable to advance catalog upper"); + self.metrics + .group_commit_catalog_upper_seconds + .observe(catalog_upper_start.elapsed().as_secs_f64()); + + let op_start = Instant::now(); + let op_res = op(write_ts.timestamp, write_ts.advance_to).await; + if let Some(metric) = op_duration_metric { + metric.observe(op_start.elapsed().as_secs_f64()); + } + + match op_res { + Ok(Ok(())) => {} + Ok(Err(StorageError::InvalidUppers(_))) => return TxnsWriteAttempt::UpperConflict, + Ok(Err(other)) => { + Err::<(), _>(other).unwrap_or_terminate("cannot fail to write to txns shard"); + unreachable!("unwrap_or_terminate does not return on Err"); + } + Err(_recv) => return TxnsWriteAttempt::WorkerGone, + } let now: Timestamp = (self.now)().into(); crate::coord::timeline::check_runaway_write_ts(&now, write_ts.timestamp); self.oracle.apply_write(write_ts.timestamp).await; - Some(write_ts) + TxnsWriteAttempt::Applied } /// Applies a staged group commit. @@ -439,6 +670,7 @@ impl GroupCommitter { responses, statement_logging_ids, notifies, + internal_results, write_locks, permits, contains_internal_system_write: _, @@ -487,6 +719,7 @@ impl GroupCommitter { .send(Message::GroupCommitApplied { responses, statement_logging_ids, + internal_results, write_ts: timestamp, }) .is_err() @@ -640,90 +873,151 @@ impl Coordinator { // Validate, merge, and possibly acquire write locks for as many pending writes as possible. for pending_write in pending_writes { match pending_write { - // We always allow system writes to proceed. PendingWriteTxn::System { .. } => validated_writes.push(pending_write), - // We have a set of locks! Validate they're correct (expected). PendingWriteTxn::User { span, write_locks: Some(write_locks), writes, - responder: UserWriteResponder::Session(pending_txn), + responder, } => match write_locks.validate(writes.keys().copied()) { Ok(validated_locks) => { - // Merge all of our write locks together since we can allow concurrent - // writes at the same timestamp. + // Locks from different sessions can be merged into one + // group because every write in the group commits at the + // same timestamp. group_write_locks.merge(validated_locks); - - let validated_write = PendingWriteTxn::User { + validated_writes.push(PendingWriteTxn::User { span, writes, write_locks: None, - responder: UserWriteResponder::Session(pending_txn), - }; - validated_writes.push(validated_write); + responder, + }); } - // This is very unexpected since callers of this method should be validating. - // - // We cannot allow these write to occur since if the correct set of locks was - // not taken we could violate serializability. + // Callers validate before they get here, so a partial set is + // a bug. We must not let the write proceed: without the + // right locks it can violate serializability. Err(missing) => { let writes: Vec<_> = writes.keys().collect(); panic!( - "got to group commit with partial set of locks!\nmissing: {:?}, writes: {:?}, txn: {:?}", - missing, writes, pending_txn, + "got to group commit with partial set of locks!\nmissing: {:?}, writes: {:?}, conn_id: {}", + missing, + writes, + responder.conn_id(), ); } }, - // If we don't have any locks, try to acquire them, otherwise defer the write. + // Without handed-off locks, acquire just in time. On a miss a + // session write defers, an internal write re-queues. PendingWriteTxn::User { span, writes, write_locks: None, - responder: UserWriteResponder::Session(pending_txn), + responder, } => { let missing = group_write_locks.missing_locks(writes.keys().copied()); - if missing.is_empty() { - // We have all the locks! Queue the pending write. - let validated_write = PendingWriteTxn::User { + validated_writes.push(PendingWriteTxn::User { span, writes, write_locks: None, - responder: UserWriteResponder::Session(pending_txn), - }; - validated_writes.push(validated_write); - } else { - // Try to acquire the locks we're missing. - let mut just_in_time_locks = WriteLocks::builder(missing.clone()); - for collection in missing { - if let Some(lock) = self.try_grant_object_write_lock(collection) { - just_in_time_locks.insert_lock(collection, lock); + responder, + }); + continue; + } + + match responder { + UserWriteResponder::Session(pending_txn) => { + let mut just_in_time_locks = WriteLocks::builder(missing.clone()); + for collection in missing { + if let Some(lock) = self.try_grant_object_write_lock(collection) { + just_in_time_locks.insert_lock(collection, lock); + } + } + match just_in_time_locks + .all_or_nothing(pending_txn.ctx.session().conn_id()) + { + Ok(locks) => { + group_write_locks.merge(locks); + validated_writes.push(PendingWriteTxn::User { + span, + writes, + write_locks: None, + responder: UserWriteResponder::Session(pending_txn), + }); + } + Err(missing) => { + let acquire_future = + self.grant_object_write_lock(missing).map(Option::Some); + deferred_writes.push(( + acquire_future, + DeferredWrite { + span, + writes, + pending_txn, + }, + )); + } } } - - match just_in_time_locks.all_or_nothing(pending_txn.ctx.session().conn_id()) - { - // We acquired all of the locks! Proceed with the write. - Ok(locks) => { - group_write_locks.merge(locks); - let validated_write = PendingWriteTxn::User { + UserWriteResponder::Internal { + conn_id, + target, + result, + } => { + // All-or-nothing, like `WriteLocks::all_or_nothing` + // for session writes: `collect` into an `Option` + // drops every lock it did acquire as soon as one is + // unavailable. Holding a partial set across the + // re-queue below could deadlock against another + // writer holding the complement. + let acquired = missing + .into_iter() + .map(|id| { + self.try_grant_object_write_lock(id).map(|lock| (id, lock)) + }) + .collect::>>(); + if let Some(acquired) = acquired { + for (id, lock) in acquired { + group_write_locks.insert_lock(id, lock); + } + validated_writes.push(PendingWriteTxn::User { span, writes, write_locks: None, - responder: UserWriteResponder::Session(pending_txn), - }; - validated_writes.push(validated_write); - } - // Darn. We couldn't acquire the locks, defer the write. - Err(missing) => { - let acquire_future = - self.grant_object_write_lock(missing).map(Option::Some); - let write = DeferredWrite { + responder: UserWriteResponder::Internal { + conn_id, + target, + result, + }, + }); + } else { + // Retry by riding the next group commit + // initiate, at the latest the periodic + // timeline advancement tick. Internal writes + // have no `ExecuteContext`, so they can't use + // `defer_op` like session writes. + // + // Deliberately without `trigger_group_commit`. + // The lock is held by a writer that is not + // waiting on us, so an immediate retry would + // find it held, re-queue, and trigger again, + // spinning for as long as the holder keeps it. + // Waiting for a trigger someone else raises + // costs at most one tick and no CPU. + // + // Lock hold times are short while frontend OCC + // sequencing is enabled because the + // coordinator's lock-based read-then-write + // path is disabled. + self.pending_writes.push(PendingWriteTxn::User { span, writes, - pending_txn, - }; - deferred_writes.push((acquire_future, write)); + write_locks: None, + responder: UserWriteResponder::Internal { + conn_id, + target, + result, + }, + }); } } } @@ -744,6 +1038,7 @@ impl Coordinator { let mut responses = Vec::with_capacity(validated_writes.len()); let mut statement_logging_ids = Vec::new(); let mut notifies = Vec::new(); + let mut internal_results = Vec::new(); for validated_write_txn in validated_writes { match validated_write_txn { @@ -775,6 +1070,38 @@ impl Coordinator { responses.push(CompletedClientTransmitter::new(ctx, response, action)); } + PendingWriteTxn::User { + span: _, + writes, + write_locks, + responder: UserWriteResponder::Internal { target, result, .. }, + } => { + assert_none!(write_locks, "should have merged together all locks above"); + let current_global_id = self + .catalog() + .try_get_entry(&target.item_id) + .map(|entry| entry.latest_global_id()); + if current_global_id != Some(target.global_id) { + result.send(WriteResult::TargetChanged); + continue; + } + // A frontend write's data all belongs to `target`, which + // `handle_attempt_write` enforces by building `writes` with + // that single key. Folding it under `target.item_id` + // regardless would append to the wrong table, so check + // rather than trust the submitter. + assert!( + writes.keys().all(|id| *id == target.item_id), + "frontend write for {:?} carries other tables: {:?}", + target.item_id, + writes.keys().collect::>(), + ); + appends + .entry(target.item_id) + .or_default() + .extend(writes.into_values().flatten()); + internal_results.push(result); + } PendingWriteTxn::System { updates, source } => { for update in updates { appends.entry(update.id).or_default().push(update.data); @@ -821,6 +1148,7 @@ impl Coordinator { responses, statement_logging_ids, notifies, + internal_results, write_locks: group_write_locks, permits: permit.into_iter().collect(), contains_internal_system_write, diff --git a/src/adapter/src/coord/message_handler.rs b/src/adapter/src/coord/message_handler.rs index c9eb4e0e17c95..988000c1abfb4 100644 --- a/src/adapter/src/coord/message_handler.rs +++ b/src/adapter/src/coord/message_handler.rs @@ -114,6 +114,7 @@ impl Coordinator { Message::GroupCommitApplied { responses, statement_logging_ids, + internal_results, write_ts, } => { // Record statement timestamps before retiring, since retiring ends the statement @@ -130,6 +131,11 @@ impl Coordinator { // that and we can downgrade the local read holds without an oracle round trip. self.downgrade_local_read_holds(write_ts); self.advance_custom_timelines().boxed_local().await; + for result in internal_results { + result.send(crate::coord::appends::WriteResult::Success { + timestamp: write_ts, + }); + } } Message::AdvanceTimelines => { // Only sent by the periodic tick in read-only mode, where group commits (which diff --git a/src/adapter/src/session.rs b/src/adapter/src/session.rs index 09cf80b7ce5fc..428e82a1086be 100644 --- a/src/adapter/src/session.rs +++ b/src/adapter/src/session.rs @@ -1863,6 +1863,11 @@ impl GroupCommitWriteLocks { self.locks.extend(std::mem::take(&mut other.locks)); } + /// Inserts a single lock, keyed by the collection it guards. + pub fn insert_lock(&mut self, id: CatalogItemId, lock: tokio::sync::OwnedMutexGuard<()>) { + self.locks.insert(id, lock); + } + /// Returns the collections we're missing locks for, if any. pub fn missing_locks( &self,