risingwave_stream/executor/source/iceberg_update/
list.rs

1// Copyright 2026 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
15//! List enumerates bounded pages of Iceberg update tasks and dispatches them to Fetch.
16
17use std::time::Duration;
18
19use either::Either;
20use futures::{TryStreamExt, stream};
21use risingwave_common::array::Op;
22use risingwave_common::bail;
23use risingwave_common::bitmap::Bitmap;
24use risingwave_common::catalog::{ColumnCatalog, ColumnId};
25use risingwave_common::config::StreamingConfig;
26use risingwave_common::hash::VnodeCountCompat;
27use risingwave_connector::source::iceberg::IcebergProperties;
28use risingwave_connector::source::iceberg::update_planner::{
29    IcebergUpdateBinding, IcebergUpdatePlanner,
30};
31use risingwave_connector::source::iceberg::update_state::{
32    IcebergUpdateAssignment, IcebergUpdateFenceKind, IcebergUpdateListState,
33};
34use risingwave_hummock_sdk::HummockReadEpoch;
35use risingwave_pb::plan_common::StorageTableDesc;
36use risingwave_storage::store::PrefetchOptions;
37use risingwave_storage::table::batch_table::BatchTable;
38use thiserror_ext::AsReport;
39use tokio::sync::mpsc::UnboundedReceiver;
40
41use super::{decode_value, encode, properties, scan_metrics, task_row, task_row_types};
42use crate::executor::prelude::*;
43use crate::executor::source::{StreamSourceCore, barrier_to_message_stream};
44use crate::executor::stream_reader::StreamReaderWithPause;
45
46pub struct IcebergUpdateListExecutor<S: StateStore> {
47    core: StreamSourceCore<S>,
48    downstream_columns: Vec<ColumnCatalog>,
49    fetch_state_table: StorageTableDesc,
50    barriers: UnboundedReceiver<Barrier>,
51    streaming_config: Arc<StreamingConfig>,
52}
53
54impl<S: StateStore> IcebergUpdateListExecutor<S> {
55    pub fn new(
56        core: StreamSourceCore<S>,
57        downstream_columns: Vec<ColumnCatalog>,
58        fetch_state_table: StorageTableDesc,
59        barriers: UnboundedReceiver<Barrier>,
60        streaming_config: Arc<StreamingConfig>,
61    ) -> Self {
62        Self {
63            core,
64            downstream_columns,
65            fetch_state_table,
66            barriers,
67            streaming_config,
68        }
69    }
70
71    #[try_stream(ok = Message, error = StreamExecutorError)]
72    async fn into_stream(mut self) {
73        let developer = &self.streaming_config.developer;
74        let list_interval = Duration::from_secs(developer.iceberg_list_interval_sec);
75        let batch_size = developer.iceberg_fetch_batch_size as usize;
76        let ordered_max_tasks = developer.iceberg_update_ordered_max_tasks as usize;
77        let properties = properties(&mut self.core)?;
78        let scan_metrics = scan_metrics(&self.core, &properties);
79        let fetch_reader = fetch_state_reader(
80            self.core
81                .split_state_store
82                .state_table()
83                .state_store()
84                .clone(),
85            &self.fetch_state_table,
86        );
87        let ctx = Arc::new(ListContext {
88            properties,
89            downstream_columns: self.downstream_columns,
90            batch_size,
91            ordered_max_tasks,
92            job_id: self.fetch_state_table.table_id.as_raw_id(),
93            fetch_reader,
94        });
95
96        let state_table = self.core.split_state_store.state_table_mut();
97        let mut barriers = barrier_to_message_stream(self.barriers).boxed();
98        let first = expect_first_barrier(&mut barriers).await?;
99        let first_epoch = first.epoch;
100        let paused = first.is_pause_on_startup();
101        yield Message::Barrier(first);
102        state_table.init_epoch(first_epoch).await?;
103
104        let mut persisted_state = state_table.get_from_one_value_table().await?;
105        let mut state = persisted_state
106            .as_ref()
107            .map(|value| decode_value::<IcebergUpdateListState>(value.as_jsonb().clone()))
108            .transpose()?;
109        if let Some(state) = &state {
110            state.validate_columns(ctx.job_id, &ctx.downstream_columns)?;
111        }
112        let mut dirty = false;
113        let mut planner = None;
114        let mut epoch = first_epoch.curr;
115        // On recovery the previous epoch is already the recovery checkpoint.
116        let mut checkpoint_epoch = first_epoch.prev;
117
118        let mut stream = ListStream::new(
119            barriers,
120            list_work(
121                ctx.clone(),
122                state.clone(),
123                None,
124                checkpoint_epoch,
125                Duration::ZERO,
126            )
127            .map(Ok),
128        );
129        if paused {
130            stream.pause_stream();
131        }
132        let mut next_poll = NextPoll::Running;
133        while let Some(message) = stream.next().await {
134            match message? {
135                Either::Left(Message::Barrier(barrier)) => {
136                    match barrier.mutation.as_deref() {
137                        Some(Mutation::Pause) => stream.pause_stream(),
138                        Some(Mutation::Resume) => stream.resume_stream(),
139                        _ => {}
140                    }
141                    if dirty {
142                        write_list_state(
143                            state_table,
144                            &mut persisted_state,
145                            state.as_ref().expect("dirty List state"),
146                        )?;
147                        dirty = false;
148                    }
149                    state_table
150                        .commit_assert_no_update_vnode_bitmap(barrier.epoch)
151                        .await?;
152                    epoch = barrier.epoch.curr;
153                    if barrier.is_checkpoint() {
154                        checkpoint_epoch = barrier.epoch.prev;
155                    }
156                    // A committed read must run only after yielding its barrier. Waiting on
157                    // the control arm would prevent the very checkpoint it needs to observe.
158                    yield Message::Barrier(barrier);
159                    if let Some(idle_delay) = next_poll.idle_delay(list_interval) {
160                        stream.replace_data_stream(
161                            list_work(
162                                ctx.clone(),
163                                state.clone(),
164                                planner.take(),
165                                checkpoint_epoch,
166                                idle_delay,
167                            )
168                            .map(Ok),
169                        );
170                        next_poll = NextPoll::Running;
171                    }
172                }
173                Either::Left(_) => bail!("unexpected input to Iceberg update List"),
174                Either::Right(Err(e)) => {
175                    tracing::warn!(
176                        error = %e.as_report(),
177                        "Iceberg update List errored, retrying"
178                    );
179                    scan_metrics.record_scan_error("list_error");
180                    // The failed poll consumed the planner. The next one rebuilds it from the
181                    // cursor in `state`, which this poll did not change.
182                    stream.replace_data_stream(stream::pending());
183                    next_poll = NextPoll::Throttled;
184                }
185                Either::Right(Ok(ListResult {
186                    state: mut new_state,
187                    planner: next_planner,
188                    assignments,
189                    fence,
190                })) => {
191                    planner = next_planner;
192                    if let Some(kind) = fence {
193                        new_state.record_fence(kind, epoch)?;
194                    }
195                    let changed = state.as_ref() != Some(&new_state);
196                    dirty |= changed;
197                    let waiting = new_state.wait_epoch().is_some();
198                    next_poll = if !assignments.is_empty() && !waiting {
199                        // More pages or ordered phases may follow, so keep listing without a
200                        // barrier.
201                        NextPoll::Running
202                    } else if assignments.is_empty() && !changed && !waiting {
203                        NextPoll::Throttled
204                    } else {
205                        NextPoll::Now
206                    };
207                    state = Some(new_state);
208                    if !assignments.is_empty() {
209                        yield Message::Chunk(assignments_chunk(&assignments)?);
210                    }
211                    if next_poll == NextPoll::Running {
212                        stream.replace_data_stream(
213                            list_work(
214                                ctx.clone(),
215                                state.clone(),
216                                planner.take(),
217                                checkpoint_epoch,
218                                Duration::ZERO,
219                            )
220                            .map(Ok),
221                        );
222                    } else {
223                        stream.replace_data_stream(stream::pending());
224                    }
225                }
226            }
227        }
228    }
229}
230
231impl<S: StateStore> Execute for IcebergUpdateListExecutor<S> {
232    fn execute(self: Box<Self>) -> BoxedMessageStream {
233        self.into_stream().boxed()
234    }
235}
236
237impl<S: StateStore> Debug for IcebergUpdateListExecutor<S> {
238    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
239        f.debug_struct("IcebergUpdateListExecutor")
240            .finish_non_exhaustive()
241    }
242}
243
244type ListStream = StreamReaderWithPause<true, StreamExecutorResult<ListResult>>;
245
246/// When List starts its next poll.
247#[derive(Clone, Copy, PartialEq, Eq)]
248enum NextPoll {
249    /// A poll is installed on the data arm.
250    Running,
251    /// Poll right after the next barrier.
252    Now,
253    /// The last poll found nothing new or failed. Wait `list_interval` after the next barrier.
254    Throttled,
255}
256
257impl NextPoll {
258    /// The delay before the next poll, or `None` if one is already running.
259    fn idle_delay(self, list_interval: Duration) -> Option<Duration> {
260        match self {
261            NextPoll::Running => None,
262            NextPoll::Now => Some(Duration::ZERO),
263            NextPoll::Throttled => Some(list_interval),
264        }
265    }
266}
267
268/// Inputs that stay fixed for the lifetime of a List executor.
269pub(super) struct ListContext<S: StateStore> {
270    pub(super) properties: IcebergProperties,
271    pub(super) downstream_columns: Vec<ColumnCatalog>,
272    pub(super) batch_size: usize,
273    pub(super) ordered_max_tasks: usize,
274    pub(super) job_id: u32,
275    pub(super) fetch_reader: BatchTable<S>,
276}
277
278pub(super) struct ListResult {
279    pub(super) state: IcebergUpdateListState,
280    pub(super) planner: Option<IcebergUpdatePlanner>,
281    pub(super) assignments: Vec<IcebergUpdateAssignment>,
282    pub(super) fence: Option<IcebergUpdateFenceKind>,
283}
284
285/// Metadata I/O runs on the data arm so it cannot prevent List from forwarding barriers.
286/// `idle_delay` throttles only catalog polls; completion checks never wait for it.
287#[try_stream(boxed, ok = ListResult, error = StreamExecutorError)]
288pub(super) async fn list_work<S: StateStore>(
289    ctx: Arc<ListContext<S>>,
290    state: Option<IcebergUpdateListState>,
291    mut planner: Option<IcebergUpdatePlanner>,
292    checkpoint_epoch: u64,
293    idle_delay: Duration,
294) {
295    let polls_catalog = state
296        .as_ref()
297        .is_none_or(|state| state.wait_epoch().is_none());
298    if polls_catalog && !idle_delay.is_zero() {
299        tokio::time::sleep(idle_delay).await;
300    }
301    let Some(mut state) = state else {
302        // Bind only once, on the pausable data arm. Persist it before enumerating tasks, so
303        // their checkpoint always includes the binding even if latest advances on recovery.
304        let table = ctx.properties.load_table().await?;
305        let binding = IcebergUpdateBinding::bind_columns(&table, &ctx.downstream_columns)?;
306        yield ListResult {
307            state: IcebergUpdateListState::new(ctx.job_id, binding),
308            planner: None,
309            assignments: vec![],
310            fence: None,
311        };
312        return Ok(());
313    };
314    let mut assignments = vec![];
315    let mut fence = None;
316    if let Some(dispatched_epoch) = state.wait_epoch() {
317        if checkpoint_epoch >= dispatched_epoch {
318            let rows = ctx
319                .fetch_reader
320                .batch_iter(
321                    HummockReadEpoch::Committed(checkpoint_epoch),
322                    false,
323                    PrefetchOptions::prefetch_for_small_range_scan(),
324                )
325                .await?;
326            pin_mut!(rows);
327            if rows.try_next().await?.is_none() {
328                state.complete_fence(checkpoint_epoch)?;
329            }
330        }
331    } else {
332        let table = ctx.properties.load_table().await?;
333        (assignments, fence) = state
334            .enumerate(&table, ctx.batch_size, ctx.ordered_max_tasks, &mut planner)
335            .await?;
336    }
337    yield ListResult {
338        state,
339        planner,
340        assignments,
341        fence,
342    };
343}
344
345/// Reads the task keys of the Fetch state table across all vnodes, so that completion
346/// observes the tasks of every Fetch actor.
347fn fetch_state_reader<S: StateStore>(store: S, desc: &StorageTableDesc) -> BatchTable<S> {
348    BatchTable::new_partial(
349        store,
350        vec![ColumnId::new(desc.columns[0].column_id)],
351        Some(Arc::new(Bitmap::ones(desc.vnode_count()))),
352        desc,
353    )
354}
355
356/// Writes the single List state row, inserting it on the first write.
357fn write_list_state<S: StateStore>(
358    table: &mut StateTable<S>,
359    persisted: &mut Option<ScalarImpl>,
360    state: &IcebergUpdateListState,
361) -> StreamExecutorResult<()> {
362    let value = ScalarImpl::Jsonb(encode(state)?);
363    if let Some(previous) = persisted.replace(value.clone()) {
364        table.update([Some(previous)], [Some(value)]);
365    } else {
366        table.insert([Some(value)]);
367    }
368    Ok(())
369}
370
371fn assignments_chunk(assignments: &[IcebergUpdateAssignment]) -> StreamExecutorResult<StreamChunk> {
372    let rows = assignments
373        .iter()
374        .map(|assignment| Ok((Op::Insert, task_row(assignment, assignment)?)))
375        .collect::<StreamExecutorResult<Vec<_>>>()?;
376    Ok(StreamChunk::from_rows(&rows, &task_row_types()))
377}
risingwave_stream/executor/source/iceberg_update/list.rs

risingwave_stream/executor/source/iceberg_update/
list.rs
