mz_compute/compute_state/error_scan.rs
1// Copyright Materialize, Inc. and contributors. All rights reserved.
2//
3// Use of this software is governed by the Business Source License
4// included in the LICENSE file.
5
6//! An index peek's walk over its error trace, the phase that runs before
7//! [`PeekResultIterator`](super::peek_result_iterator::PeekResultIterator) walks the ok trace.
8
9use std::time::{Duration, Instant};
10
11use differential_dataflow::trace::{Cursor, TraceReader};
12use mz_compute_client::protocol::response::PeekError;
13use mz_repr::{Diff, GlobalId, Timestamp};
14use timely::order::PartialOrder;
15use tracing::error;
16
17use crate::arrangement::manager::PaddedTrace;
18use crate::compute_state::{PeekRowIterationTracker, peek_result_iterator};
19use crate::typedefs::ErrAgent;
20
21/// The error trace of an index, as
22/// [`TraceBundle::errs_mut`](crate::arrangement::manager::TraceBundle::errs_mut) hands it out.
23pub(super) type ErrsHandle = PaddedTrace<ErrAgent<Timestamp, Diff>>;
24
25/// A walk over an index peek's error trace, suspendable between cursor positions.
26///
27/// Holds nothing of the ok trace or of the rows a peek returns. A peek reaches those only once
28/// this walk reports [`ErrorScanStep::Finished`] with an `Ok`.
29pub(super) struct ErrorScan {
30 cursor: peek_result_iterator::TraceCursor<ErrsHandle>,
31 storage: peek_result_iterator::TraceStorage<ErrsHandle>,
32 /// The limit spans this walk and the ok scan after it, so the count accrued here is handed
33 /// on with [`ErrorScanStep::Finished`].
34 row_iteration_tracker: PeekRowIterationTracker,
35 /// Worker time spent walking, summed over the calls the walk was sliced into.
36 pub(super) scan_time: Duration,
37}
38
39/// The outcome of a fueled [`ErrorScan::step`].
40#[derive(Debug, PartialEq)]
41pub(super) enum ErrorScanStep {
42 /// The walk reached its end. `Ok` carries the rows it examined over a trace holding no error
43 /// at the peek's timestamp, which the ok scan continues from; `Err` is the peek's answer,
44 /// either an error the trace holds or a failure of the walk itself.
45 Finished(Result<usize, PeekError>),
46 /// The fuel ran out first. The walk resumes at the cursor position it stopped on, and that
47 /// position has not been examined yet.
48 OutOfFuel,
49}
50
51impl ErrorScan {
52 /// Opens a walk over `errs`.
53 ///
54 /// The walk starts without a row-iteration limit. The limit in effect is the caller's to
55 /// supply through [`ErrorScan::set_row_iteration_limit`] before each step.
56 pub(super) fn new(errs: &mut ErrsHandle) -> Self {
57 let scan_start = Instant::now();
58 let (cursor, storage) = errs.cursor();
59 let mut scan = Self::from_cursor(cursor, storage);
60 scan.scan_time = scan_start.elapsed();
61 scan
62 }
63
64 /// Opens a walk over an already-opened cursor.
65 pub(super) fn from_cursor(
66 cursor: peek_result_iterator::TraceCursor<ErrsHandle>,
67 storage: peek_result_iterator::TraceStorage<ErrsHandle>,
68 ) -> Self {
69 Self {
70 cursor,
71 storage,
72 row_iteration_tracker: PeekRowIterationTracker::new(None, 0),
73 scan_time: Duration::ZERO,
74 }
75 }
76
77 /// Adopts the row-iteration limit that is in effect, without forgetting the rows the walk has
78 /// already examined.
79 pub(super) fn set_row_iteration_limit(&mut self, limit: Option<usize>) {
80 self.row_iteration_tracker.set_limit(limit);
81 }
82
83 /// Advances the walk until it has an answer for the peek, the cursor is exhausted, or `fuel`
84 /// runs out, whichever comes first. Decrements `fuel` by the number of cursor positions
85 /// visited.
86 ///
87 /// A key whose diffs cancel to zero at `peek_timestamp` yields no answer, so fuel is charged
88 /// per position rather than per answer. Otherwise a trace holding many such keys would run to
89 /// its end within a single step.
90 ///
91 /// This walk does not remember that it ended, and stepping it again after it reported
92 /// [`ErrorScanStep::Finished`] walks a spent cursor.
93 /// [`PeekScan`](super::peek_scan::PeekScan)'s error phase remembers instead: it keeps the
94 /// outcome and drops the walk, so a finished peek stops pinning error batches.
95 pub(super) fn step(
96 &mut self,
97 peek_timestamp: Timestamp,
98 target_id: GlobalId,
99 fuel: &mut usize,
100 ) -> ErrorScanStep {
101 let step_start = Instant::now();
102
103 let outcome = loop {
104 // Charged after this, so that finding the trace exhausted costs nothing and an
105 // exhausted walk reports the end rather than asking for a budget it cannot spend.
106 if !self.cursor.key_valid(&self.storage) {
107 break ErrorScanStep::Finished(Ok(self.row_iteration_tracker.rows_iterated()));
108 }
109
110 if *fuel == 0 {
111 break ErrorScanStep::OutOfFuel;
112 }
113 *fuel -= 1;
114
115 if let Err(error) = self.row_iteration_tracker.track_next() {
116 break ErrorScanStep::Finished(Err(error));
117 }
118
119 let mut copies = Diff::ZERO;
120 self.cursor.map_times(&self.storage, |time, diff| {
121 if time.less_equal(&peek_timestamp) {
122 copies += diff;
123 }
124 });
125 if copies.is_negative() {
126 let error = self.cursor.key(&self.storage);
127 error!(
128 target = %target_id, diff = %copies, %error,
129 "index peek encountered negative multiplicities in error trace",
130 );
131 break ErrorScanStep::Finished(Err(PeekError::unstructured(format!(
132 "Invalid data in source errors, \
133 saw retractions ({}) for row that does not exist: {}",
134 -copies, error,
135 ))));
136 }
137 if copies.is_positive() {
138 let error = self.cursor.key(&self.storage).deserialize();
139 break ErrorScanStep::Finished(Err(error.into()));
140 }
141 self.cursor.step_key(&self.storage);
142 };
143
144 self.scan_time += step_start.elapsed();
145 outcome
146 }
147}
148
149#[cfg(test)]
150pub(crate) mod tests;