risingwave_meta/manager/iceberg_pk_index_sink/
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//! Manager for the Iceberg pk-index sink path. Owns per-sink commit coordinators that
16//! drive iceberg `commit_epoch` ahead of hummock `commit_epoch` and persist
17//! exactly-once state via `pending_sink_state`.
18//!
19//! This is intentionally separate from [`crate::manager::sink_coordination`]
20//! (which serves V1/V2 sinks via gRPC). Future responsibilities such as
21//! per-sink compaction will live alongside the per-sink commit coordinator here.
22
23mod committed_epoch;
24mod coordinator;
25mod manager;
26
27use std::collections::{BTreeMap, HashMap};
28
29use anyhow::anyhow;
30use iceberg::spec::SerializedDataFile;
31pub use manager::IcebergPkIndexSinkManager;
32pub(crate) use manager::IcebergPkIndexSinkRegistration;
33use risingwave_common::secret::LocalSecretManager;
34use risingwave_connector::sink::catalog::SinkId;
35use risingwave_connector::sink::iceberg::{ENABLE_PK_INDEX, IcebergConfig};
36use risingwave_connector::source::UPSTREAM_SOURCE_KEY;
37use risingwave_pb::catalog::PbSink;
38use risingwave_pb::stream_service::barrier_complete_response::IcebergPkIndexSinkMetadata as PbIcebergPkIndexSinkMetadata;
39
40#[derive(educe::Educe)]
41#[educe(Debug)]
42pub(crate) struct CompactionOverwrite {
43    pub sink_id: SinkId,
44    pub epoch: u64,
45    pub schema_id: i32,
46    pub partition_spec_id: i32,
47    #[educe(Debug(ignore))]
48    pub output_files: Vec<SerializedDataFile>,
49    pub input_file_paths: Vec<String>,
50    pub read_snapshot_id: i64,
51}
52
53/// Metadata collected for one sink/epoch before the Hummock checkpoint is committed.
54/// Ordinary reports and the optional compaction overwrite share one transport shape so the
55/// barrier path does not need separate metadata variants.
56#[derive(Debug)]
57pub(crate) struct IcebergPkIndexPreCommitMetadata {
58    pub sink_id: SinkId,
59    pub prev_epoch: u64,
60    pub reports: Vec<PbIcebergPkIndexSinkMetadata>,
61    pub compaction: Option<CompactionOverwrite>,
62}
63
64impl From<PbIcebergPkIndexSinkMetadata> for IcebergPkIndexPreCommitMetadata {
65    fn from(report: PbIcebergPkIndexSinkMetadata) -> Self {
66        Self {
67            sink_id: report.sink_id,
68            prev_epoch: report.prev_epoch,
69            reports: vec![report],
70            compaction: None,
71        }
72    }
73}
74
75impl From<CompactionOverwrite> for IcebergPkIndexPreCommitMetadata {
76    fn from(overwrite: CompactionOverwrite) -> Self {
77        Self {
78            sink_id: overwrite.sink_id,
79            prev_epoch: overwrite.epoch,
80            reports: vec![],
81            compaction: Some(overwrite),
82        }
83    }
84}
85
86pub(crate) fn group_pre_commit_metadata(
87    metadata: Vec<IcebergPkIndexPreCommitMetadata>,
88) -> anyhow::Result<Vec<IcebergPkIndexPreCommitMetadata>> {
89    let mut grouped = HashMap::<SinkId, IcebergPkIndexPreCommitMetadata>::new();
90    for metadata in metadata {
91        let IcebergPkIndexPreCommitMetadata {
92            sink_id,
93            prev_epoch,
94            reports,
95            compaction,
96        } = metadata;
97        let input = grouped
98            .entry(sink_id)
99            .or_insert_with(|| IcebergPkIndexPreCommitMetadata {
100                sink_id,
101                prev_epoch,
102                reports: Vec::new(),
103                compaction: None,
104            });
105        if input.prev_epoch != prev_epoch {
106            anyhow::bail!(
107                "iceberg v3 sink {} pre-commit metadata disagrees on prev_epoch: {} vs {}",
108                sink_id,
109                input.prev_epoch,
110                prev_epoch
111            );
112        }
113        input.reports.extend(reports);
114        if let Some(overwrite) = compaction
115            && input.compaction.replace(overwrite).is_some()
116        {
117            anyhow::bail!(
118                "iceberg v3 sink {} has multiple compaction overwrites in one barrier",
119                sink_id
120            );
121        }
122    }
123    Ok(grouped.into_values().collect())
124}
125
126/// Returns true if the given sink properties identify a Iceberg pk-index sink
127/// (i.e. an iceberg sink with `enable_pk_index = 'true'`).
128pub fn is_iceberg_pk_index_sink(properties: &BTreeMap<String, String>) -> bool {
129    let connector_match = properties
130        .get(UPSTREAM_SOURCE_KEY)
131        .map(|v| v.eq_ignore_ascii_case("iceberg"))
132        .unwrap_or(false);
133    let pk_index_enabled = properties
134        .get(ENABLE_PK_INDEX)
135        .map(|v| v.eq_ignore_ascii_case("true"))
136        .unwrap_or(false);
137    connector_match && pk_index_enabled
138}
139
140/// Build an [`IcebergConfig`] from a [`PbSink`], filling secret refs along the
141/// way. Used at CREATE SINK time and during recovery to (re-)register the
142/// commit coordinator.
143pub fn build_iceberg_config(pb_sink: &PbSink) -> anyhow::Result<IcebergConfig> {
144    let properties: BTreeMap<String, String> = pb_sink.properties.clone().into_iter().collect();
145    let secret_refs: BTreeMap<_, _> = pb_sink.secret_refs.clone().into_iter().collect();
146    let with_secrets = LocalSecretManager::global()
147        .fill_secrets(properties, secret_refs)
148        .map_err(|e| anyhow!(e).context("fill secrets for iceberg"))?;
149    IcebergConfig::from_btreemap(with_secrets)
150        .map_err(|e| anyhow!(e).context("parse iceberg config"))
151}
152
153#[cfg(test)]
154mod tests {
155    use risingwave_pb::stream_service::PbIcebergPkIndexSinkRole;
156
157    use super::*;
158
159    fn report(
160        sink_id: u32,
161        prev_epoch: u64,
162        role: PbIcebergPkIndexSinkRole,
163    ) -> PbIcebergPkIndexSinkMetadata {
164        PbIcebergPkIndexSinkMetadata {
165            sink_id: SinkId::new(sink_id),
166            prev_epoch,
167            role: role as i32,
168            ..Default::default()
169        }
170    }
171
172    fn overwrite(sink_id: u32, epoch: u64) -> CompactionOverwrite {
173        CompactionOverwrite {
174            sink_id: SinkId::new(sink_id),
175            epoch,
176            schema_id: 3,
177            partition_spec_id: 4,
178            output_files: vec![],
179            input_file_paths: vec![],
180            read_snapshot_id: 1,
181        }
182    }
183
184    #[test]
185    fn group_pre_commit_metadata_combines_reports_and_compaction() {
186        let inputs = group_pre_commit_metadata(vec![
187            report(7, 60, PbIcebergPkIndexSinkRole::Writer).into(),
188            overwrite(7, 60).into(),
189            report(7, 60, PbIcebergPkIndexSinkRole::PositionDeleteMerger).into(),
190        ])
191        .unwrap();
192
193        assert_eq!(inputs.len(), 1);
194        let input = &inputs[0];
195        assert_eq!(input.sink_id, SinkId::new(7));
196        assert_eq!(input.prev_epoch, 60);
197        assert_eq!(input.reports.len(), 2);
198        let overwrite = input.compaction.as_ref().unwrap();
199        assert_eq!(overwrite.schema_id, 3);
200        assert_eq!(overwrite.partition_spec_id, 4);
201    }
202
203    #[test]
204    fn group_pre_commit_metadata_preserves_ordinary_only_sink() {
205        let inputs =
206            group_pre_commit_metadata(vec![report(8, 61, PbIcebergPkIndexSinkRole::Writer).into()])
207                .unwrap();
208
209        assert_eq!(inputs.len(), 1);
210        assert_eq!(inputs[0].sink_id, SinkId::new(8));
211        assert_eq!(inputs[0].prev_epoch, 61);
212        assert_eq!(inputs[0].reports.len(), 1);
213        assert!(inputs[0].compaction.is_none());
214    }
215
216    #[test]
217    fn group_pre_commit_metadata_rejects_duplicate_compaction() {
218        let error =
219            group_pre_commit_metadata(vec![overwrite(7, 60).into(), overwrite(7, 60).into()])
220                .unwrap_err();
221        assert!(error.to_string().contains("multiple compaction overwrites"));
222    }
223
224    #[test]
225    fn group_pre_commit_metadata_rejects_epoch_mismatch() {
226        let error = group_pre_commit_metadata(vec![
227            report(7, 59, PbIcebergPkIndexSinkRole::Writer).into(),
228            overwrite(7, 60).into(),
229        ])
230        .unwrap_err();
231        assert!(error.to_string().contains("disagrees on prev_epoch"));
232    }
233}
risingwave_meta/manager/iceberg_pk_index_sink/mod.rs

risingwave_meta/manager/iceberg_pk_index_sink/
mod.rs
