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::id::SinkId;
20use risingwave_pb::stream_plan::CompactionResolverNode;
21use risingwave_storage::StateStore;
22
23use crate::error::StreamResult;
24use crate::executor::{CompactionResolverExecutor, Executor, StreamExecutorError};
25use crate::from_proto::ExecutorBuilder;
26use crate::task::ExecutorParams;
27
28pub struct CompactionResolverExecutorBuilder;
29
30impl_stream_node_body!(CompactionResolver(CompactionResolverNode) => CompactionResolverExecutorBuilder);
31
32impl ExecutorBuilder for CompactionResolverExecutorBuilder {
33    type Node = CompactionResolverNode;
34
35    async fn new_boxed_executor(
36        params: ExecutorParams,
37        node: &Self::Node,
38        _store: impl StateStore,
39    ) -> StreamResult<Executor> {
40        assert!(
41            params.input.is_empty(),
42            "compaction resolver executor should not have input"
43        );
44
45        let sink_desc = node.sink_desc.as_ref().unwrap();
46        let sink_id: SinkId = sink_desc.get_id();
47
48        let properties_with_secret = LocalSecretManager::global().fill_secrets(
49            sink_desc.get_properties().clone(),
50            sink_desc.get_secret_refs().clone(),
51        )?;
52        let iceberg_config = IcebergConfig::from_btreemap(properties_with_secret)
53            .map_err(|err| StreamExecutorError::from((err, sink_id)))?;
54
55        // Primary-key column indices within the iceberg data-file row. The writer writes every input
56        // column to iceberg verbatim, so these indices (`SinkDesc.downstream_pk`) also index the
57        // data-file columns.
58        let pk_indices = sink_desc
59            .downstream_pk
60            .iter()
61            .map(|&idx| idx as usize)
62            .collect::<Vec<_>>();
63        if pk_indices.is_empty() {
64            return Err(anyhow!("missing downstream pk in iceberg sink desc").into());
65        }
66
67        // The pk-index state table schema is `[pk.., file_path, position]`, so the first
68        // `pk_indices.len()` columns are the pk columns (in `downstream_pk` order). Derive the output
69        // chunk's pk column data types from them so `Writer_B` consumes a schema identical to the
70        // index key columns.
71        let pk_index_table = node.get_pk_index_table()?;
72        let pk_data_types = pk_index_table
73            .columns
74            .iter()
75            .take(pk_indices.len())
76            .map(|col| {
77                let column_desc = col
78                    .column_desc
79                    .as_ref()
80                    .ok_or_else(|| anyhow!("pk-index table column missing column_desc"))?;
81                Ok::<_, anyhow::Error>(ColumnDesc::from(column_desc).data_type)
82            })
83            .collect::<Result<Vec<_>, _>>()?;
84        if pk_data_types.len() != pk_indices.len() {
85            return Err(anyhow!(
86                "pk-index table has {} columns but sink has {} pk columns",
87                pk_index_table.columns.len(),
88                pk_indices.len()
89            )
90            .into());
91        }
92
93        let barrier_receiver = params
94            .local_barrier_manager
95            .subscribe_barrier(params.actor_context.id);
96        let local_barrier_manager = params.local_barrier_manager.clone();
97        let meta_client = params.env.meta_client().ok_or_else(|| {
98            anyhow!("meta client is required for iceberg pk-index compaction resolver")
99        })?;
100
101        let exec = CompactionResolverExecutor::new(
102            params.actor_context,
103            sink_id,
104            node.compaction_task_id,
105            iceberg_config,
106            pk_indices,
107            pk_data_types,
108            node.output_data_file_paths.clone(),
109            node.input_data_file_paths.clone(),
110            node.read_snapshot_id,
111            params.config.developer.chunk_size,
112            local_barrier_manager,
113            barrier_receiver,
114            meta_client,
115        );
116        Ok((params.info, exec).into())
117    }
118}
risingwave_stream/from_proto/iceberg_with_pk_index/compaction_resolver.rs

risingwave_stream/from_proto/iceberg_with_pk_index/
compaction_resolver.rs
