risingwave_frontend/optimizer/plan_node/
logical_iceberg_intermediate_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 std::collections::HashMap;
16
17use educe::Educe;
18use iceberg::expr::Predicate;
19use itertools::Itertools;
20use pretty_xmlish::{Pretty, XmlNode};
21use risingwave_common::catalog::{ColumnCatalog, Field};
22use risingwave_common::types::DataType;
23use risingwave_connector::source::iceberg::IcebergTimeTravelInfo;
24
25use super::generic::GenericPlanRef;
26use super::utils::{Distill, childless_record};
27use super::{
28    ColPrunable, ExprRewritable, Logical, LogicalPlanRef as PlanRef, PlanBase, PredicatePushdown,
29    ToBatch, ToStream, generic,
30};
31use crate::catalog::source_catalog::SourceCatalog;
32use crate::error::Result;
33use crate::optimizer::plan_node::expr_visitable::ExprVisitable;
34use crate::optimizer::plan_node::utils::column_names_pretty;
35use crate::optimizer::plan_node::{
36    ColumnPruningContext, LogicalFilter, LogicalProject, LogicalSource, PredicatePushdownContext,
37    RewriteStreamContext, ToStreamContext,
38};
39use crate::utils::{
40    ColIndexMapping, Condition, ExtractIcebergPredicateResult, extract_iceberg_predicate,
41};
42
43/// Predicate and column mapping needed when rewriting an Iceberg scan to a
44/// Hummock `LogicalScan`. Only consumed by `IcebergEngineStorageSelectionRule`;
45/// the normal Iceberg path ignores these fields.
46#[derive(Debug, Clone, PartialEq, Eq)]
47pub struct HummockRewriteInfo {
48    /// The accumulated predicate over *table* column indices (not output columns),
49    /// built from the extracted portion of pushed-down predicates.
50    pub origin_condition: Condition,
51    /// Maps current output column indices → original table column indices.
52    pub output_column_mapping: ColIndexMapping,
53}
54
55impl HummockRewriteInfo {
56    /// Create with an initial source→table column mapping.
57    /// For non-engine iceberg sources (no associated table), pass an identity mapping.
58    pub fn new(source_to_table_mapping: ColIndexMapping) -> Self {
59        Self {
60            origin_condition: Condition::true_cond(),
61            output_column_mapping: source_to_table_mapping,
62        }
63    }
64
65    /// Accumulate a new predicate, remapping it through the current column mapping.
66    pub fn add_predicate(&self, extracted_condition: Condition) -> Self {
67        let mut mapping = self.output_column_mapping.clone();
68        let remapped = extracted_condition.rewrite_expr(&mut mapping);
69        Self {
70            origin_condition: self.origin_condition.clone().and(remapped),
71            output_column_mapping: mapping,
72        }
73    }
74
75    /// Prune columns: compose the column mapping so that the new output indices
76    /// still map back to the original table column indices.
77    pub fn prune_columns(&self, required_cols: &[usize]) -> Self {
78        let map = required_cols
79            .iter()
80            .map(|&idx| Some(self.output_column_mapping.map(idx)))
81            .collect();
82        Self {
83            origin_condition: self.origin_condition.clone(),
84            output_column_mapping: ColIndexMapping::new(
85                map,
86                self.output_column_mapping.target_size(),
87            ),
88        }
89    }
90}
91
92/// `LogicalIcebergIntermediateScan` is an intermediate plan node used during optimization
93/// of Iceberg scans. It accumulates predicates and column pruning information before
94/// being converted to the final `LogicalIcebergScan` with delete file anti-joins.
95///
96/// This node is introduced to reduce the number of Iceberg metadata reads. Instead of
97/// reading metadata when creating `LogicalIcebergScan`, we defer the metadata read
98/// until all optimizations (predicate pushdown, column pruning) are applied.
99///
100/// The optimization flow is:
101/// 1. `LogicalSource` (iceberg) -> `LogicalIcebergIntermediateScan`
102/// 2. Predicate pushdown and column pruning are applied to `LogicalIcebergIntermediateScan`
103/// 3. `LogicalIcebergIntermediateScan` -> `LogicalIcebergScan` (with anti-joins for delete files)
104#[derive(Debug, Clone, PartialEq, Educe)]
105#[educe(Hash)]
106pub struct LogicalIcebergIntermediateScan {
107    pub base: PlanBase<Logical>,
108    pub core: generic::Source,
109    #[educe(Hash(ignore))]
110    pub iceberg_predicate: Predicate,
111    pub time_travel_info: IcebergTimeTravelInfo,
112    /// For Iceberg engine tables: maps source column name → target Hummock `DataType`.
113    /// This remapping is applied to the output schema so that the intermediate scan's
114    /// output types match the Hummock table types, avoiding unnecessary double casts
115    /// when the storage selection rule rewrites to a Hummock `LogicalScan`.
116    /// Empty for non-engine-table Iceberg sources.
117    #[educe(Hash(ignore))]
118    pub table_column_type_mapping: HashMap<String, DataType>,
119    /// Info needed only when rewriting to a Hummock row-store scan.
120    #[educe(Hash(ignore))]
121    pub hummock_rewrite: HummockRewriteInfo,
122}
123
124impl Eq for LogicalIcebergIntermediateScan {}
125
126impl LogicalIcebergIntermediateScan {
127    pub fn new(
128        logical_source: &LogicalSource,
129        time_travel_info: IcebergTimeTravelInfo,
130        table_column_type_mapping: HashMap<String, DataType>,
131        // Maps source-column indices to Hummock table-column indices so that
132        // `HummockRewriteInfo` tracks predicates and projections in table
133        // index space. Pass `ColIndexMapping::identity(n)` when there is no
134        // associated Hummock table (e.g. standalone iceberg sources).
135        source_to_table_mapping: ColIndexMapping,
136    ) -> Self {
137        assert!(logical_source.core.is_iceberg_connector());
138
139        let mut core = logical_source.core.clone();
140        // Apply type remapping: change the source column types to Hummock table types
141        // so that the output schema has Hummock types.
142        for col in &mut core.column_catalog {
143            if let Some(target_type) = table_column_type_mapping.get(col.name()) {
144                col.column_desc.data_type = target_type.clone();
145            }
146        }
147        let hummock_rewrite = HummockRewriteInfo::new(source_to_table_mapping);
148        let base = PlanBase::new_logical_with_core(&core);
149        assert!(logical_source.output_exprs.is_none());
150        LogicalIcebergIntermediateScan {
151            base,
152            core,
153            iceberg_predicate: Predicate::AlwaysTrue,
154            time_travel_info,
155            table_column_type_mapping,
156            hummock_rewrite,
157        }
158    }
159
160    pub fn source_catalog(&self) -> Option<&SourceCatalog> {
161        self.core.catalog.as_deref()
162    }
163
164    /// Fields carrying the iceberg-side column types (before the engine-table Hummock
165    /// type remapping), for predicate pushdown. Derived from the source catalog, which
166    /// the remapping never touches.
167    fn iceberg_side_fields(&self) -> Vec<Field> {
168        let catalog = self
169            .core
170            .catalog
171            .as_ref()
172            .expect("iceberg intermediate scan must have a source catalog");
173        let by_name: HashMap<&str, &ColumnCatalog> =
174            catalog.columns.iter().map(|c| (c.name(), c)).collect();
175        self.core
176            .column_catalog
177            .iter()
178            .map(|col| {
179                let source_col = by_name
180                    .get(col.name())
181                    .expect("output column must exist in the source catalog");
182                Field::from(&source_col.column_desc)
183            })
184            .collect()
185    }
186
187    pub fn output_columns(&self) -> impl ExactSizeIterator<Item = &str> {
188        self.core.column_catalog.iter().map(|c| c.name.as_str())
189    }
190
191    pub fn add_predicate(
192        &self,
193        iceberg_predicate: Predicate,
194        extracted_condition: Condition,
195    ) -> Self {
196        LogicalIcebergIntermediateScan {
197            iceberg_predicate: self.iceberg_predicate.clone().and(iceberg_predicate),
198            hummock_rewrite: self.hummock_rewrite.add_predicate(extracted_condition),
199            ..self.clone()
200        }
201    }
202
203    /// Returns true if this intermediate scan has type remapping for Iceberg engine tables.
204    pub fn has_type_mapping(&self) -> bool {
205        !self.table_column_type_mapping.is_empty()
206    }
207
208    pub fn clone_with_required_cols(&self, required_cols: &[usize]) -> Self {
209        assert!(!required_cols.is_empty());
210
211        let mut core = self.core.clone();
212        core.column_catalog = required_cols
213            .iter()
214            .map(|idx| core.column_catalog[*idx].clone())
215            .collect();
216        core.row_id_index = required_cols
217            .iter()
218            .position(|idx| Some(*idx) == self.core.row_id_index);
219
220        let base = PlanBase::new_logical_with_core(&core);
221
222        LogicalIcebergIntermediateScan {
223            base,
224            core,
225            iceberg_predicate: self.iceberg_predicate.clone(),
226            time_travel_info: self.time_travel_info.clone(),
227            table_column_type_mapping: self.table_column_type_mapping.clone(),
228            hummock_rewrite: self.hummock_rewrite.prune_columns(required_cols),
229        }
230    }
231}
232
233impl_plan_tree_node_for_leaf! { Logical, LogicalIcebergIntermediateScan }
234
235impl Distill for LogicalIcebergIntermediateScan {
236    fn distill<'a>(&self) -> XmlNode<'a> {
237        let verbose = self.base.ctx().is_explain_verbose();
238        let mut fields = Vec::with_capacity(if verbose { 4 } else { 2 });
239
240        if let Some(catalog) = self.source_catalog() {
241            fields.push(("source", Pretty::from(catalog.name.clone())));
242        } else {
243            fields.push(("source", Pretty::from("unknown")));
244        }
245        fields.push(("columns", column_names_pretty(self.schema())));
246
247        if verbose {
248            fields.push(("predicate", Pretty::debug(&self.iceberg_predicate)));
249            fields.push((
250                "output_column",
251                Pretty::debug(&self.output_columns().collect_vec()),
252            ));
253            fields.push(("time_travel_info", Pretty::debug(&self.time_travel_info)));
254        }
255
256        childless_record("LogicalIcebergIntermediateScan", fields)
257    }
258}
259
260impl ColPrunable for LogicalIcebergIntermediateScan {
261    fn prune_col(&self, required_cols: &[usize], _ctx: &mut ColumnPruningContext) -> PlanRef {
262        if required_cols.is_empty() {
263            // If required_cols is empty, we use the first column of iceberg to avoid the empty schema.
264            LogicalProject::new(self.clone_with_required_cols(&[0]).into(), vec![]).into()
265        } else {
266            self.clone_with_required_cols(required_cols).into()
267        }
268    }
269}
270
271impl ExprRewritable<Logical> for LogicalIcebergIntermediateScan {}
272
273impl ExprVisitable for LogicalIcebergIntermediateScan {}
274
275impl PredicatePushdown for LogicalIcebergIntermediateScan {
276    fn predicate_pushdown(
277        &self,
278        predicate: Condition,
279        _ctx: &mut PredicatePushdownContext,
280    ) -> PlanRef {
281        let ExtractIcebergPredicateResult {
282            iceberg_predicate,
283            extracted_condition,
284            remaining_condition,
285        } = extract_iceberg_predicate(predicate, &self.iceberg_side_fields());
286        let plan = self
287            .add_predicate(iceberg_predicate, extracted_condition)
288            .into();
289        if remaining_condition.always_true() {
290            plan
291        } else {
292            LogicalFilter::create(plan, remaining_condition)
293        }
294    }
295}
296
297impl ToBatch for LogicalIcebergIntermediateScan {
298    fn to_batch(&self) -> Result<crate::optimizer::plan_node::BatchPlanRef> {
299        // This should not be called directly. The intermediate scan should be
300        // converted to LogicalIcebergScan first via the materialization rule.
301        Err(crate::error::ErrorCode::InternalError(
302            "LogicalIcebergIntermediateScan should be converted to LogicalIcebergScan before to_batch".to_owned()
303        )
304        .into())
305    }
306}
307
308impl ToStream for LogicalIcebergIntermediateScan {
309    fn to_stream(
310        &self,
311        _ctx: &mut ToStreamContext,
312    ) -> Result<crate::optimizer::plan_node::StreamPlanRef> {
313        unreachable!("LogicalIcebergIntermediateScan is only for batch queries")
314    }
315
316    fn logical_rewrite_for_stream(
317        &self,
318        _ctx: &mut RewriteStreamContext,
319    ) -> Result<(PlanRef, ColIndexMapping)> {
320        unreachable!("LogicalIcebergIntermediateScan is only for batch queries")
321    }
322}
risingwave_frontend/optimizer/plan_node/logical_iceberg_intermediate_scan.rs

risingwave_frontend/optimizer/plan_node/
logical_iceberg_intermediate_scan.rs
