risingwave_stream/executor/iceberg_with_pk_index/
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//! pk-index sink
16//!
17//! This module implements three core executors for the Iceberg pk-index sink that uses
18//! Deletion Vectors (DVs) instead of Equality Delete files:
19//!
20//! 1. **Writer Executor** (Stateful): Maintains a PK index mapping primary keys to
21//!    (`file_path`, `position`). Writes data files for inserts and emits
22//!    (`file_path`, `position`) messages for deletes.
23//!
24//! 2. **Position-delete merger executor** (Stateless): Consumes the Writer's (`file_path`, `position`)
25//!    messages, merges delete positions with historical deletes, and reports the resulting delete
26//!    files to meta.
27
28mod compaction_resolver;
29mod position_delete_handler_impl;
30mod position_delete_merger;
31mod position_delete_staging;
32mod writer;
33mod writer_impl;
34
35use std::time::Duration;
36
37pub use compaction_resolver::CompactionResolverExecutor;
38use iceberg::table::Table;
39pub use position_delete_handler_impl::PositionDeleteHandlerImpl;
40pub use position_delete_merger::PositionDeleteMergerExecutor;
41use risingwave_connector::sink::iceberg::IcebergConfig;
42use risingwave_connector::sink::{Result as SinkResult, SinkError};
43pub use writer::WriterExecutor;
44pub use writer_impl::IcebergWriterImpl;
45
46/// Load the table, retrying until its snapshot set contains `expected`
47pub async fn load_table_at_least(
48    config: &IcebergConfig,
49    expected: Option<i64>,
50) -> SinkResult<Table> {
51    const MAX_ATTEMPTS: usize = 10;
52    const BACKOFF: Duration = Duration::from_millis(500);
53    let mut last = None;
54    for _ in 0..MAX_ATTEMPTS {
55        let table = config.load_table().await?;
56        let Some(expected) = expected else {
57            return Ok(table);
58        };
59        if table.metadata().snapshot_by_id(expected).is_some() {
60            return Ok(table);
61        }
62        last = Some(table.metadata().current_snapshot_id());
63        tokio::time::sleep(BACKOFF).await;
64    }
65    Err(SinkError::Iceberg(anyhow::anyhow!(
66        "iceberg catalog did not reflect committed pk-index snapshot {:?} after {} attempts (last current_snapshot_id={:?})",
67        expected,
68        MAX_ATTEMPTS,
69        last,
70    )))
71}
risingwave_stream/executor/iceberg_with_pk_index/mod.rs

risingwave_stream/executor/iceberg_with_pk_index/
mod.rs
