risingwave_frontend/optimizer/rule/
iceberg_engine_storage_selection_rule.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//! When `iceberg_query_storage_mode` is `auto`, this rule may rewrite
16//! `LogicalIcebergIntermediateScan` (columnar Iceberg) to `LogicalScan` (row Hummock)
17//! for Iceberg engine tables.
18
19use std::collections::HashSet;
20use std::sync::Arc;
21
22use risingwave_common::session_config::IcebergQueryStorageMode;
23
24use super::prelude::{PlanRef, *};
25use crate::TableCatalog;
26use crate::catalog::source_catalog::SourceCatalog;
27use crate::optimizer::plan_node::generic::GenericPlanRef;
28use crate::optimizer::plan_node::{Logical, LogicalIcebergIntermediateScan, LogicalScan, generic};
29use crate::optimizer::rule::InfallibleRule;
30use crate::session::SessionImpl;
31
32pub struct IcebergEngineStorageSelectionRule;
33
34impl InfallibleRule<Logical> for IcebergEngineStorageSelectionRule {
35    fn apply(&self, plan: PlanRef) -> Option<PlanRef> {
36        let scan = plan.as_logical_iceberg_intermediate_scan()?;
37        let ctx = plan.ctx();
38        let session = ctx.session_ctx();
39
40        // Only apply when storage mode is auto.
41        if session.config().iceberg_query_storage_mode() != IcebergQueryStorageMode::Auto {
42            return None;
43        }
44        let source_catalog = scan.source_catalog()?;
45        let table = get_table_from_iceberg_source(session, source_catalog)?;
46
47        // Append-only Iceberg engine tables do not materialize rows in their Hummock table. Their
48        // batch plan combines Iceberg with the sink log store, so rewriting the Iceberg side to
49        // the dummy Hummock table would silently drop committed rows.
50        if table.append_only {
51            return None;
52        }
53
54        let prefer_rowstore = check_point_lookup(scan, &table);
55        if !prefer_rowstore {
56            return None;
57        }
58
59        rewrite_to_table_scan(scan, &table)
60    }
61}
62
63impl IcebergEngineStorageSelectionRule {
64    pub fn create() -> BoxedRule {
65        Box::new(IcebergEngineStorageSelectionRule)
66    }
67}
68
69/// Rewrite the intermediate Iceberg scan to a Hummock `LogicalScan`.
70fn rewrite_to_table_scan(
71    scan: &LogicalIcebergIntermediateScan,
72    table: &Arc<TableCatalog>,
73) -> Option<PlanRef> {
74    // output_column_mapping already maps to table-column indices (built at
75    // construction time), so we can use it and origin_condition directly.
76    let output_col_idx = scan
77        .hummock_rewrite
78        .output_column_mapping
79        .to_parts()
80        .0
81        .iter()
82        .copied()
83        .try_collect()?;
84    let table_scan = generic::TableScan::new(
85        output_col_idx,
86        table.clone(),
87        vec![],
88        vec![],
89        scan.ctx(),
90        scan.hummock_rewrite.origin_condition.clone(),
91        scan.core.as_of.clone(),
92    );
93    Some(LogicalScan::from(table_scan).into())
94}
95
96fn get_table_from_iceberg_source(
97    session: &SessionImpl,
98    source_catalog: &SourceCatalog,
99) -> Option<Arc<TableCatalog>> {
100    let catalog_reader = session.env().catalog_reader().read_guard();
101    let schema = catalog_reader
102        .get_schema_by_id(source_catalog.database_id, source_catalog.schema_id)
103        .ok()?;
104    let table_name = source_catalog.iceberg_table_name()?;
105    let table = schema.get_created_table_by_name(&table_name)?;
106    Some(table.clone())
107}
108
109/// Returns `true` when the predicate has equality-to-constant conditions on
110/// *all* PK columns of the table, making this a point lookup that benefits
111/// from the row store's key-value access pattern.
112fn check_point_lookup(scan: &LogicalIcebergIntermediateScan, table: &TableCatalog) -> bool {
113    let pk_column_names: HashSet<&str> = table.pk_column_names().into_iter().collect();
114    if pk_column_names.is_empty() {
115        return false;
116    }
117
118    // origin_condition is already in table-column index space.
119    let eq_input_refs = scan
120        .hummock_rewrite
121        .origin_condition
122        .get_eq_const_input_refs();
123    let eq_col_names: HashSet<&str> = eq_input_refs
124        .iter()
125        .filter_map(|input_ref| table.columns().get(input_ref.index()))
126        .filter(|c| !c.is_hidden())
127        .map(|c| c.name.as_str())
128        .collect();
129
130    // All PK columns must be covered by equality predicates.
131    pk_column_names.is_subset(&eq_col_names)
132}
risingwave_frontend/optimizer/rule/iceberg_engine_storage_selection_rule.rs

risingwave_frontend/optimizer/rule/
iceberg_engine_storage_selection_rule.rs
