risingwave_stream/executor/backfill/no_shuffle_backfill.rs
1// Copyright 2023 RisingWave Labs
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7// http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use risingwave_common::bail;
16use risingwave_common::hash::VnodeBitmapExt;
17use risingwave_storage::table::batch_table::BatchTable;
18
19use crate::executor::backfill::utils::mapping_message;
20use crate::executor::prelude::*;
21use crate::task::{CreateMviewProgressReporter, FragmentId};
22
23/// Schema: | vnode | pk ... | `backfill_finished` | `row_count` |
24/// We can decode that into `BackfillState` on recovery.
25#[derive(Debug, Eq, PartialEq)]
26pub struct BackfillState {
27 current_pos: Option<OwnedRow>,
28 is_finished: bool,
29 row_count: u64,
30}
31
32/// An implementation of the [RFC: Use Backfill To Let Mv On Mv Stream Again](https://github.com/risingwavelabs/rfcs/pull/13).
33/// `BackfillExecutor` is used to create a materialized view on another materialized view.
34///
35/// It can only buffer chunks between two barriers instead of unbundled memory usage of
36/// `RearrangedChainExecutor`.
37///
38/// It uses the latest epoch to read the snapshot of the upstream mv during two barriers and all the
39/// `StreamChunk` of the snapshot read will forward to the downstream.
40///
41/// It uses `current_pos` to record the progress of the backfill (the pk of the upstream mv) and
42/// `current_pos` is initiated as an empty `Row`.
43///
44/// All upstream messages during the two barriers interval will be buffered and decide to forward or
45/// ignore based on the `current_pos` at the end of the later barrier. Once `current_pos` reaches
46/// the end of the upstream mv pk, the backfill would finish.
47///
48/// Notice:
49/// The pk we are talking about here refers to the storage primary key.
50/// We rely on the scheduler to schedule the `BackfillExecutor` together with the upstream mv/table
51/// in the same worker, so that we can read uncommitted data from the upstream table without
52/// waiting.
53pub struct BackfillExecutor<S: StateStore> {
54 /// Upstream table
55 upstream_table: BatchTable<S>,
56 /// Upstream with the same schema with the upstream table.
57 upstream: Executor,
58
59 /// Internal state table for persisting state of backfill state.
60 state_table: Option<StateTable<S>>,
61
62 /// The column indices need to be forwarded to the downstream from the upstream and table scan.
63 output_indices: Vec<usize>,
64
65 /// PTAL at the docstring for `CreateMviewProgress` to understand how we compute it.
66 progress: CreateMviewProgressReporter,
67
68 actor_id: ActorId,
69
70 fragment_id: FragmentId,
71}
72
73impl<S> BackfillExecutor<S>
74where
75 S: StateStore,
76{
77 pub fn new(
78 upstream_table: BatchTable<S>,
79 upstream: Executor,
80 state_table: Option<StateTable<S>>,
81 output_indices: Vec<usize>,
82 progress: CreateMviewProgressReporter,
83 fragment_id: FragmentId,
84 ) -> Self {
85 let actor_id = progress.actor_id();
86 Self {
87 upstream_table,
88 upstream,
89 state_table,
90 output_indices,
91 progress,
92 actor_id,
93 fragment_id,
94 }
95 }
96
97 #[try_stream(ok = Message, error = StreamExecutorError)]
98 async fn execute_inner(mut self) {
99 // The primary key columns.
100 // We receive a pruned chunk from the upstream table,
101 // which will only contain output columns of the scan on the upstream table.
102 // The pk indices specify the pk columns of the pruned chunk.
103 let pk_indices = self.upstream_table.pk_in_output_indices().unwrap();
104
105 let upstream_table_id = self.upstream_table.table_id();
106
107 let mut upstream = self.upstream.execute();
108
109 // Poll the upstream to get the first barrier.
110 let first_barrier = expect_first_barrier(&mut upstream).await?;
111 let first_epoch = first_barrier.epoch;
112 // The first barrier message should be propagated.
113 yield Message::Barrier(first_barrier);
114
115 if let Some(state_table) = self.state_table.as_mut() {
116 state_table.init_epoch(first_epoch).await?;
117 }
118
119 let BackfillState {
120 current_pos,
121 is_finished,
122 row_count,
123 ..
124 } = Self::recover_backfill_state(self.state_table.as_ref(), pk_indices.len()).await?;
125 tracing::trace!(is_finished, row_count, "backfill state recovered");
126
127 if !is_finished {
128 bail!(
129 "legacy no-shuffle backfill recovered unfinished progress; cancel and recreate the streaming job. upstream_table_id={:?}, fragment_id={:?}, actor_id={}, current_pos={:?}, row_count={}",
130 upstream_table_id,
131 self.fragment_id,
132 self.actor_id,
133 current_pos,
134 row_count,
135 );
136 }
137
138 tracing::trace!("Backfill has finished, waiting for barrier");
139
140 // Wait for first barrier to come after backfill is finished.
141 // So we can update our progress + persist the status.
142 while let Some(Ok(msg)) = upstream.next().await {
143 if let Some(msg) = mapping_message(msg, &self.output_indices) {
144 if let Message::Barrier(barrier) = &msg {
145 // If already finished, no need persist any state, but we need to advance the
146 // epoch of the state table anyway.
147 if let Some(table) = &mut self.state_table {
148 table
149 .commit_assert_no_update_vnode_bitmap(barrier.epoch)
150 .await?;
151 }
152
153 // Backfill progress in meta is not persisted by the executor, so report it
154 // again after recovery.
155 self.progress.finish(barrier.epoch, row_count);
156 tracing::trace!(
157 epoch = ?barrier.epoch,
158 "Updated CreateMaterializedTracker"
159 );
160 yield msg;
161 break;
162 }
163 // Allow other messages to pass through.
164 // We won't yield twice here, since if there's a barrier,
165 // we will always break out of the loop.
166 yield msg;
167 }
168 }
169
170 tracing::trace!(
171 "Backfill has already finished and forward messages directly to the downstream"
172 );
173
174 // After progress finished + state persisted,
175 // we can forward messages directly to the downstream,
176 // as backfill is finished.
177 // We don't need to report backfill progress any longer, as it has finished.
178 // It will always be at 100%.
179 #[for_await]
180 for msg in upstream {
181 if let Some(msg) = mapping_message(msg?, &self.output_indices) {
182 if let Message::Barrier(barrier) = &msg {
183 // If already finished, no need persist any state, but we need to advance the epoch of the state table anyway.
184 if let Some(table) = &mut self.state_table {
185 table
186 .commit_assert_no_update_vnode_bitmap(barrier.epoch)
187 .await?;
188 }
189 }
190
191 yield msg;
192 }
193 }
194 }
195
196 async fn recover_backfill_state(
197 state_table: Option<&StateTable<S>>,
198 pk_len: usize,
199 ) -> StreamExecutorResult<BackfillState> {
200 let Some(state_table) = state_table else {
201 // If no state table, but backfill is present, it must be from an old cluster.
202 // In that case backfill must be finished, otherwise it won't have been persisted.
203 return Ok(BackfillState {
204 current_pos: None,
205 is_finished: true,
206 row_count: 0,
207 });
208 };
209 let mut vnodes = state_table.vnodes().iter_vnodes_scalar();
210 let first_vnode = vnodes.next().unwrap();
211 let key: &[Datum] = &[Some(first_vnode.into())];
212 let row = state_table.get_row(key).await?;
213 let expected_state = Self::deserialize_backfill_state(row, pk_len);
214
215 // All vnode partitions should have same state (no scale-in supported).
216 for vnode in vnodes {
217 let key: &[Datum] = &[Some(vnode.into())];
218 let row = state_table.get_row(key).await?;
219 let state = Self::deserialize_backfill_state(row, pk_len);
220 assert_eq!(state.is_finished, expected_state.is_finished);
221 }
222 Ok(expected_state)
223 }
224
225 fn deserialize_backfill_state(row: Option<OwnedRow>, pk_len: usize) -> BackfillState {
226 let Some(row) = row else {
227 return BackfillState {
228 current_pos: None,
229 is_finished: false,
230 row_count: 0,
231 };
232 };
233 let row = row.into_inner();
234 let current_pos = Some((&row[0..pk_len]).into_owned_row());
235 let is_finished = row[pk_len].clone().is_some_and(|d| d.into_bool());
236 let row_count = row
237 .get(pk_len + 1)
238 .cloned()
239 .unwrap_or(None)
240 .map_or(0, |d| d.into_int64() as u64);
241 BackfillState {
242 current_pos,
243 is_finished,
244 row_count,
245 }
246 }
247}
248
249impl<S> Execute for BackfillExecutor<S>
250where
251 S: StateStore,
252{
253 fn execute(self: Box<Self>) -> BoxedMessageStream {
254 self.execute_inner().boxed()
255 }
256}