risingwave_stream/executor/source/iceberg_update/
mod.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//! Delete-aware Iceberg ingestion (`streaming_updates = 'true'`).
16//!
17//! List dispatches bounded pages continuously within a phase. Fetch retires tasks at EOF.
18//! Small phases share one ordered Fetch route and follow each other without waiting. Other
19//! phase transitions wait asynchronously for an empty committed Fetch queue, while ordinary
20//! barriers remain responsive. Output and physical cursors recover at the same checkpoint.
21//!
22//! List output rows and Fetch state rows are both `(route, seq, json)`. The Fetch state table
23//! is keyed by `(route, seq)` and distributed by `route`, like the List-to-Fetch exchange.
24
25mod fetch;
26mod list;
27
28use anyhow::Context;
29pub use fetch::IcebergUpdateFetchExecutor;
30pub use list::IcebergUpdateListExecutor;
31use risingwave_common::types::{JsonbVal, ScalarRef};
32use risingwave_common::{bail, ensure};
33use risingwave_connector::source::ConnectorProperties;
34use risingwave_connector::source::iceberg::update_state::IcebergUpdateAssignment;
35use risingwave_connector::source::iceberg::{IcebergProperties, IcebergScanMetricsLabels};
36use serde::Serialize;
37use serde::de::DeserializeOwned;
38
39use super::StreamSourceCore;
40use crate::executor::prelude::*;
41
42pub(crate) fn validate_update_state_table(
43    table: &risingwave_pb::catalog::Table,
44    singleton: bool,
45) -> StreamExecutorResult<()> {
46    use risingwave_pb::data::data_type::TypeName;
47    let types = table
48        .columns
49        .iter()
50        .map(|column| {
51            column
52                .column_desc
53                .as_ref()
54                .and_then(|desc| desc.column_type.as_ref())
55                .map(|ty| ty.type_name())
56        })
57        .collect::<Vec<_>>();
58    if singleton {
59        ensure!(
60            types == vec![Some(TypeName::Jsonb)]
61                && table.pk.is_empty()
62                && table.value_indices == vec![0]
63                && table.distribution_key.is_empty(),
64            "Iceberg update List state requires a single JSON value without a primary key"
65        );
66    } else {
67        ensure!(
68            types
69                == vec![
70                    Some(TypeName::Int64),
71                    Some(TypeName::Int64),
72                    Some(TypeName::Jsonb)
73                ]
74                && table.pk.iter().map(|order| order.column_index).eq([0, 1])
75                && table.value_indices == vec![0, 1, 2]
76                && table.distribution_key == vec![0],
77            "Iceberg update Fetch state requires (route, seq, JSON) rows keyed by (route, seq) and distributed by route"
78        );
79    }
80    Ok(())
81}
82
83fn properties(
84    core: &mut StreamSourceCore<impl StateStore>,
85) -> StreamExecutorResult<IcebergProperties> {
86    let builder = core
87        .source_desc_builder
88        .take()
89        .context("missing Iceberg source description")?;
90    let ConnectorProperties::Iceberg(properties) =
91        ConnectorProperties::extract(builder.with_properties(), false)?
92    else {
93        bail!("update source requires Iceberg");
94    };
95    Ok(*properties)
96}
97
98fn scan_metrics(
99    core: &StreamSourceCore<impl StateStore>,
100    properties: &IcebergProperties,
101) -> IcebergScanMetricsLabels {
102    IcebergScanMetricsLabels::new(
103        core.source_id.to_string(),
104        core.source_name.clone(),
105        properties.table.table_name().to_owned(),
106    )
107}
108
109fn encode(value: &impl Serialize) -> StreamExecutorResult<JsonbVal> {
110    Ok(serde_json::to_value(value)
111        .map_err(anyhow::Error::from)?
112        .into())
113}
114
115/// The data types of List output rows and Fetch state rows.
116fn task_row_types() -> [DataType; 3] {
117    [DataType::Int64, DataType::Int64, DataType::Jsonb]
118}
119
120/// The Fetch state primary key of an assignment.
121fn task_pk(assignment: &IcebergUpdateAssignment) -> OwnedRow {
122    OwnedRow::new(vec![
123        Some(ScalarImpl::Int64(assignment.route)),
124        Some(ScalarImpl::Int64(assignment.seq)),
125    ])
126}
127
128fn task_row(
129    assignment: &IcebergUpdateAssignment,
130    value: &impl Serialize,
131) -> StreamExecutorResult<OwnedRow> {
132    let mut row = task_pk(assignment).into_inner().into_vec();
133    row.push(Some(ScalarImpl::Jsonb(encode(value)?)));
134    Ok(OwnedRow::new(row))
135}
136
137/// Decodes the JSON value of a task row and checks it against the row key.
138fn decode<T: DeserializeOwned>(
139    row: impl Row,
140    assignment: impl FnOnce(&T) -> &IcebergUpdateAssignment,
141) -> StreamExecutorResult<T> {
142    let value = row
143        .datum_at(2)
144        .context("missing Iceberg update state")?
145        .into_jsonb();
146    let value: T = decode_value(value.to_owned_scalar())?;
147    let key = assignment(&value);
148    ensure!(
149        row.datum_at(0).map(ScalarRefImpl::into_int64) == Some(key.route)
150            && row.datum_at(1).map(ScalarRefImpl::into_int64) == Some(key.seq),
151        "Iceberg task row key differs from its assignment"
152    );
153    Ok(value)
154}
155
156fn decode_value<T: DeserializeOwned>(value: JsonbVal) -> StreamExecutorResult<T> {
157    Ok(serde_json::from_value(value.take()).map_err(anyhow::Error::from)?)
158}
159
160#[cfg(test)]
161mod tests;
risingwave_stream/executor/source/iceberg_update/mod.rs

risingwave_stream/executor/source/iceberg_update/
mod.rs
