risingwave_connector/source/iceberg/
update_state.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//! Durable List/Fetch handoff for Insert/Delete updates.
16//!
17//! Every task has a Fetch route and a dispatch sequence number. The Fetch state table is
18//! distributed by route and ordered by sequence number within a route, so one Fetch actor applies
19//! the tasks of a route in dispatch order, also across rescaling. A phase that fits in one small
20//! page goes to the shared ordered route and needs no completion fence: later ordered phases queue
21//! behind it. Other phases fan out across routes. Before a fan-out page, and after the last page
22//! of a fan-out phase, List confirms that the Fetch queue is empty at a committed epoch covering
23//! the earlier dispatch.
24
25use anyhow::Context;
26use iceberg::table::Table;
27use risingwave_common::catalog::ColumnCatalog;
28use risingwave_common::ensure;
29use serde::{Deserialize, Serialize};
30
31use super::update_planner::{
32    IcebergUpdateBinding, IcebergUpdateCursor, IcebergUpdatePhase, IcebergUpdatePlanner,
33    IcebergUpdateTask,
34};
35use crate::error::ConnectorResult;
36
37#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
38pub struct IcebergUpdateAssignment {
39    /// The dedicated Fetch state table ID. Distinct ingestion jobs never share this table.
40    pub job_id: u32,
41    /// The Fetch state distribution key.
42    pub route: i64,
43    /// Unique within the job and increasing in dispatch order. Orders tasks within a route.
44    pub seq: i64,
45    pub task: IcebergUpdateTask,
46}
47
48impl IcebergUpdateAssignment {
49    /// Shared by all ordered phases of a job. Fan-out tasks use their nonnegative `seq`.
50    pub const ORDERED_ROUTE: i64 = -1;
51}
52
53#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
54pub struct IcebergUpdateFetchState {
55    pub assignment: IcebergUpdateAssignment,
56    pub next_position: u64,
57    pub finished: bool,
58}
59
60impl IcebergUpdateFetchState {
61    pub fn new(assignment: IcebergUpdateAssignment) -> Self {
62        Self {
63            assignment,
64            next_position: 0,
65            finished: false,
66        }
67    }
68
69    pub fn advance(&mut self, next_position: u64, finished: bool) -> ConnectorResult<()> {
70        ensure!(
71            !self.finished
72                && next_position >= self.next_position
73                && next_position <= self.assignment.task.record_count(),
74            "invalid Iceberg fetch cursor advance"
75        );
76        ensure!(
77            !finished || next_position == self.assignment.task.record_count(),
78            "Iceberg task completion before EOF"
79        );
80        self.next_position = next_position;
81        self.finished = finished;
82        Ok(())
83    }
84}
85
86/// Which outstanding tasks a dispatch must wait for, recorded at the epoch that emits a page.
87#[derive(Debug, Clone, Copy, PartialEq, Eq)]
88pub enum IcebergUpdateFenceKind {
89    /// A complete phase on the ordered route. Only fan-out dispatch waits for it.
90    Ordered,
91    /// The last page of a fan-out phase. All later dispatch waits for it.
92    FanOut,
93}
94
95/// Outstanding tasks, confirmed by an empty committed Fetch queue at `epoch` or later. The
96/// queue never holds ordered and fan-out tasks at once, so emptiness covers both.
97#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
98pub struct IcebergUpdateFence {
99    /// The graph epoch of the last covered dispatch. An older empty Fetch snapshot says
100    /// nothing about completion, since assignments may not yet be visible.
101    epoch: u64,
102    /// Only ordered tasks are outstanding and no fan-out page waits for them, so ordered
103    /// dispatch may continue behind them.
104    ordered: bool,
105}
106
107#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
108pub struct IcebergUpdateListState {
109    pub job_id: u32,
110    binding: IcebergUpdateBinding,
111    bootstrap_complete: bool,
112    /// The target of the last range whose tasks are all dispatched, and the base of the next.
113    dispatched_snapshot: Option<i64>,
114    cursor: Option<IcebergUpdateCursor>,
115    next_seq: i64,
116    fence: Option<IcebergUpdateFence>,
117}
118
119impl IcebergUpdateListState {
120    pub fn new(job_id: u32, binding: IcebergUpdateBinding) -> Self {
121        Self {
122            job_id,
123            binding,
124            bootstrap_complete: false,
125            dispatched_snapshot: None,
126            cursor: None,
127            next_seq: 0,
128            fence: None,
129        }
130    }
131
132    pub fn validate_columns(&self, job_id: u32, columns: &[ColumnCatalog]) -> ConnectorResult<()> {
133        ensure!(
134            self.job_id == job_id,
135            "Iceberg List job changed on recovery"
136        );
137        self.binding.validate_columns(columns)
138    }
139
140    /// The epoch whose committed Fetch queue must be empty before List dispatches again.
141    pub fn wait_epoch(&self) -> Option<u64> {
142        self.fence
143            .filter(|fence| !fence.ordered)
144            .map(|fence| fence.epoch)
145    }
146
147    /// Build one bounded page and advance the enumeration cursor, without waiting for Fetch.
148    /// A phase that fits in its first page with at most `ordered_max_tasks` tasks goes to the
149    /// ordered route. The caller must record the returned fence at the epoch emitting the page
150    /// before persisting this state. Assignments and the cursor share a graph epoch.
151    pub async fn enumerate(
152        &mut self,
153        table: &Table,
154        batch_size: usize,
155        ordered_max_tasks: usize,
156        planner: &mut Option<IcebergUpdatePlanner>,
157    ) -> ConnectorResult<(Vec<IcebergUpdateAssignment>, Option<IcebergUpdateFenceKind>)> {
158        ensure!(batch_size > 0, "Iceberg fetch batch size must be positive");
159        ensure!(
160            self.wait_epoch().is_none(),
161            "cannot enumerate until dispatched Iceberg tasks are applied"
162        );
163        let planner =
164            planner.get_or_insert_with(|| IcebergUpdatePlanner::new(self.binding.clone()));
165        if self.cursor.is_none() {
166            self.cursor =
167                planner.start_cursor(table, self.bootstrap_complete, self.dispatched_snapshot)?;
168            if self.cursor.is_none() {
169                // Binding an empty table is itself a completed empty bootstrap, not permission
170                // to choose a different bootstrap on the next poll.
171                self.bootstrap_complete = true;
172                return Ok((vec![], None));
173            }
174        }
175        let cursor = self.cursor.clone().expect("cursor set");
176        let page = planner.plan_page(table, &cursor, batch_size).await?;
177        if page.tasks.is_empty() {
178            ensure!(page.phase_finished, "empty non-final Iceberg page");
179            self.advance_phase();
180            return Ok((vec![], None));
181        }
182        // The route of a phase is chosen before its first dispatch, so only a phase that fits
183        // in its first page can be ordered.
184        let ordered = cursor.after_path.is_none()
185            && page.phase_finished
186            && page.tasks.len() <= ordered_max_tasks;
187        if !ordered && let Some(fence) = &mut self.fence {
188            // Fan-out tasks on other routes could overtake the outstanding ordered tasks. Keep
189            // the page for the next poll and wait until the ordered route drains.
190            planner.unread(cursor, page.tasks);
191            fence.ordered = false;
192            return Ok((vec![], None));
193        }
194        let first_seq = self.next_seq;
195        self.next_seq = i64::try_from(page.tasks.len())
196            .ok()
197            .and_then(|len| first_seq.checked_add(len))
198            .context("Iceberg task sequence overflow")?;
199        self.cursor.as_mut().expect("cursor set").after_path = Some(
200            page.tasks
201                .last()
202                .expect("nonempty page")
203                .id
204                .data_file_path
205                .clone(),
206        );
207        let assignments = page
208            .tasks
209            .into_iter()
210            .enumerate()
211            .map(|(offset, task)| (task, first_seq + offset as i64))
212            .map(|(task, seq)| IcebergUpdateAssignment {
213                job_id: self.job_id,
214                route: if ordered {
215                    IcebergUpdateAssignment::ORDERED_ROUTE
216                } else {
217                    seq
218                },
219                seq,
220                task,
221            })
222            .collect();
223        let fence = if ordered {
224            Some(IcebergUpdateFenceKind::Ordered)
225        } else {
226            page.phase_finished
227                .then_some(IcebergUpdateFenceKind::FanOut)
228        };
229        if page.phase_finished {
230            self.advance_phase();
231        }
232        Ok((assignments, fence))
233    }
234
235    /// Called when the page is emitted, not when its asynchronous planning started.
236    pub fn record_fence(
237        &mut self,
238        kind: IcebergUpdateFenceKind,
239        epoch: u64,
240    ) -> ConnectorResult<()> {
241        let ordered = kind == IcebergUpdateFenceKind::Ordered;
242        ensure!(
243            match self.fence {
244                None => true,
245                Some(fence) => ordered && fence.ordered && fence.epoch <= epoch,
246            },
247            "invalid Iceberg dispatch fence"
248        );
249        self.fence = Some(IcebergUpdateFence { epoch, ordered });
250        Ok(())
251    }
252
253    /// The caller has observed an empty Fetch table at this committed epoch.
254    pub fn complete_fence(&mut self, committed_epoch: u64) -> ConnectorResult<()> {
255        ensure!(
256            self.wait_epoch()
257                .is_some_and(|epoch| epoch <= committed_epoch),
258            "Iceberg completion snapshot predates dispatch"
259        );
260        self.fence = None;
261        Ok(())
262    }
263
264    fn advance_phase(&mut self) {
265        let mut cursor = self.cursor.take().expect("enumerated page has a cursor");
266        if cursor.phase == IcebergUpdatePhase::Delete {
267            cursor.phase = IcebergUpdatePhase::Insert;
268            cursor.after_path = None;
269            self.cursor = Some(cursor);
270        } else {
271            self.dispatched_snapshot = Some(cursor.snapshot_id);
272            self.bootstrap_complete = true;
273            self.cursor = None;
274        }
275    }
276}
risingwave_connector/source/iceberg/update_state.rs

risingwave_connector/source/iceberg/
update_state.rs
