risingwave_stream/from_proto/iceberg_with_pk_index/
compaction_resolver.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 anyhow::anyhow;
16use risingwave_common::catalog::ColumnDesc;
17use risingwave_common::secret::LocalSecretManager;
18use risingwave_connector::sink::iceberg::IcebergConfig;
19use risingwave_pb::stream_plan::CompactionResolverNode;
20use risingwave_storage::StateStore;
21
22use crate::error::StreamResult;
23use crate::executor::{CompactionResolverExecutor, Executor, StreamExecutorError};
24use crate::from_proto::ExecutorBuilder;
25use crate::task::ExecutorParams;
26
27pub struct CompactionResolverExecutorBuilder;
28
29impl_stream_node_body!(CompactionResolver(CompactionResolverNode) => CompactionResolverExecutorBuilder);
30
31impl ExecutorBuilder for CompactionResolverExecutorBuilder {
32    type Node = CompactionResolverNode;
33
34    async fn new_boxed_executor(
35        params: ExecutorParams,
36        node: &Self::Node,
37        _store: impl StateStore,
38    ) -> StreamResult<Executor> {
39        assert!(
40            params.input.is_empty(),
41            "compaction resolver executor should not have input"
42        );
43
44        let sink_id = node.sink_id;
45
46        let properties_with_secret = LocalSecretManager::global()
47            .fill_secrets(node.properties.clone(), node.secret_refs.clone())?;
48        let iceberg_config = IcebergConfig::from_btreemap(properties_with_secret)
49            .map_err(|err| StreamExecutorError::from((err, sink_id)))?;
50
51        let pk_indices = node
52            .pk_columns
53            .iter()
54            .map(|column| column.data_file_index as usize)
55            .collect::<Vec<_>>();
56        if pk_indices.is_empty() {
57            return Err(anyhow!("missing primary-key columns in compaction resolver").into());
58        }
59
60        let pk_data_types = node
61            .pk_columns
62            .iter()
63            .map(|column| {
64                column
65                    .column_desc
66                    .as_ref()
67                    .map(ColumnDesc::from)
68                    .map(|column| column.data_type)
69                    .ok_or_else(|| anyhow!("compaction resolver PK column missing column_desc"))
70            })
71            .collect::<Result<Vec<_>, _>>()?;
72
73        let barrier_receiver = params
74            .local_barrier_manager
75            .subscribe_barrier(params.actor_context.id);
76        let local_barrier_manager = params.local_barrier_manager.clone();
77        let meta_client = params.env.meta_client().ok_or_else(|| {
78            anyhow!("meta client is required for iceberg pk-index compaction resolver")
79        })?;
80        let exec = CompactionResolverExecutor::new(
81            params.actor_context,
82            sink_id,
83            iceberg_config,
84            pk_indices,
85            pk_data_types,
86            params.config.developer.chunk_size,
87            local_barrier_manager,
88            barrier_receiver,
89            meta_client,
90        );
91        Ok((params.info, exec).into())
92    }
93}
risingwave_stream/from_proto/iceberg_with_pk_index/compaction_resolver.rs

risingwave_stream/from_proto/iceberg_with_pk_index/
compaction_resolver.rs
