risingwave_batch_executors/executor/
iceberg_metadata_scan.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::Context;
16use futures_async_stream::try_stream;
17use futures_util::stream::StreamExt;
18use risingwave_common::array::DataChunk;
19use risingwave_common::catalog::Schema;
20use risingwave_connector::WithOptionsSecResolved;
21use risingwave_connector::sink::iceberg::{IcebergMetadataTableType, scan_iceberg_metadata};
22use risingwave_connector::source::ConnectorProperties;
23use risingwave_connector::source::iceberg::IcebergTimeTravelInfo;
24use risingwave_pb::batch_plan::iceberg_metadata_scan_node::{MetadataType, TimeTravel};
25use risingwave_pb::batch_plan::plan_node::NodeBody;
26
27use crate::error::{BatchError, anyhow};
28use crate::executor::{BoxedExecutor, BoxedExecutorBuilder, Executor, ExecutorBuilder};
29
30pub struct IcebergMetadataScanExecutor {
31    schema: Schema,
32    properties: risingwave_connector::source::iceberg::IcebergProperties,
33    metadata_type: IcebergMetadataTableType,
34    time_travel_info: Option<IcebergTimeTravelInfo>,
35    identity: String,
36    chunk_size: usize,
37}
38
39impl Executor for IcebergMetadataScanExecutor {
40    fn schema(&self) -> &Schema {
41        &self.schema
42    }
43
44    fn identity(&self) -> &str {
45        &self.identity
46    }
47
48    fn execute(self: Box<Self>) -> super::BoxedDataChunkStream {
49        self.do_execute().boxed()
50    }
51}
52
53impl IcebergMetadataScanExecutor {
54    #[try_stream(ok = DataChunk, error = BatchError)]
55    async fn do_execute(self: Box<Self>) {
56        let table = self.properties.load_table().await?;
57        #[for_await]
58        for chunk in scan_iceberg_metadata(
59            table,
60            self.metadata_type,
61            self.time_travel_info,
62            self.chunk_size,
63        ) {
64            yield chunk?;
65        }
66    }
67}
68
69pub struct IcebergMetadataScanExecutorBuilder;
70
71impl BoxedExecutorBuilder for IcebergMetadataScanExecutorBuilder {
72    async fn new_boxed_executor(
73        source: &ExecutorBuilder<'_>,
74        inputs: Vec<BoxedExecutor>,
75    ) -> crate::error::Result<BoxedExecutor> {
76        ensure!(
77            inputs.is_empty(),
78            "Iceberg metadata scan should not have input executors"
79        );
80        let node = try_match_expand!(
81            source.plan_node().get_node_body().unwrap(),
82            NodeBody::IcebergMetadataScan
83        )?;
84
85        let metadata_type = match MetadataType::try_from(node.metadata_type)
86            .context("invalid Iceberg metadata type")?
87        {
88            MetadataType::Snapshots => IcebergMetadataTableType::Snapshots,
89            MetadataType::Manifests => IcebergMetadataTableType::Manifests,
90            MetadataType::Files => IcebergMetadataTableType::Files,
91            MetadataType::Unspecified => {
92                return Err(anyhow!("Iceberg metadata type is unspecified").into());
93            }
94        };
95        let time_travel_info = node
96            .time_travel
97            .as_ref()
98            .map(|time_travel| match time_travel {
99                TimeTravel::SnapshotId(snapshot_id) => IcebergTimeTravelInfo::Version(*snapshot_id),
100                TimeTravel::TimestampMs(timestamp_ms) => {
101                    IcebergTimeTravelInfo::TimestampMs(*timestamp_ms)
102                }
103            });
104        let config = ConnectorProperties::extract(
105            WithOptionsSecResolved::new(node.with_properties.clone(), node.secret_refs.clone()),
106            false,
107        )?;
108        let ConnectorProperties::Iceberg(properties) = config else {
109            return Err(anyhow!("Iceberg metadata scan received a non-Iceberg connector").into());
110        };
111
112        Ok(Box::new(IcebergMetadataScanExecutor {
113            schema: metadata_type.schema(),
114            properties: *properties,
115            metadata_type,
116            time_travel_info,
117            identity: source.plan_node().get_identity().clone(),
118            chunk_size: source.context().get_config().developer.chunk_size,
119        }))
120    }
121}
risingwave_batch_executors/executor/iceberg_metadata_scan.rs

risingwave_batch_executors/executor/
iceberg_metadata_scan.rs
