risingwave_stream/executor/iceberg_with_pk_index/
writer_impl.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
15use iceberg::table::Table;
16use iceberg::writer::PositionDeleteInput;
17use risingwave_common::array::DataChunk;
18use risingwave_connector::sink::SinkWriterParam;
19use risingwave_connector::sink::iceberg::{IcebergConfig, IcebergSinkWriterInner};
20use risingwave_pb::connector_service::SinkMetadata;
21use risingwave_pb::id::SinkId;
22
23use super::writer::IcebergWriter;
24use crate::executor::{StreamExecutorError, StreamExecutorResult};
25
26pub struct IcebergWriterImpl {
27    inner: IcebergSinkWriterInner,
28    sink_id: SinkId,
29}
30
31impl IcebergWriterImpl {
32    pub fn build(
33        config: &IcebergConfig,
34        table: Table,
35        writer_param: &SinkWriterParam,
36    ) -> StreamExecutorResult<Self> {
37        let sink_id = writer_param.sink_id;
38        let inner = IcebergSinkWriterInner::build_append_only(config, table, writer_param)
39            .map_err(|e| StreamExecutorError::sink_error(e, sink_id))?;
40
41        Ok(Self { inner, sink_id })
42    }
43}
44
45#[async_trait::async_trait]
46impl IcebergWriter for IcebergWriterImpl {
47    async fn write_chunk(
48        &mut self,
49        chunk: DataChunk,
50    ) -> StreamExecutorResult<Vec<PositionDeleteInput>> {
51        let positions = self
52            .inner
53            .write_batch_with_position(chunk.into())
54            .await
55            .map_err(|e| StreamExecutorError::sink_error(e, self.sink_id))?;
56        Ok(positions)
57    }
58
59    async fn flush(&mut self) -> StreamExecutorResult<Option<SinkMetadata>> {
60        let Some(data_files) = self
61            .inner
62            .close()
63            .await
64            .map_err(|e| StreamExecutorError::sink_error(e, self.sink_id))?
65        else {
66            return Ok(None);
67        };
68        if data_files.is_empty() {
69            return Ok(None);
70        }
71
72        let metadata = self
73            .inner
74            .generate_commit_metadata(data_files)
75            .map_err(|e| StreamExecutorError::sink_error(e, self.sink_id))?;
76
77        Ok(Some(metadata))
78    }
79}
risingwave_stream/executor/iceberg_with_pk_index/writer_impl.rs

risingwave_stream/executor/iceberg_with_pk_index/
writer_impl.rs
