risingwave_frontend/optimizer/plan_node/
logical_iceberg_scan.rs

1// Copyright 2024 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::hash::{Hash, Hasher};
16
17use pretty_xmlish::{Pretty, XmlNode};
18use risingwave_connector::source::iceberg::IcebergFileScanTask;
19use risingwave_pb::batch_plan::iceberg_scan_node::IcebergScanType;
20
21use super::generic::GenericPlanRef;
22use super::utils::{Distill, childless_record};
23use super::{
24    ColPrunable, ExprRewritable, Logical, LogicalPlanRef as PlanRef, PlanBase, PredicatePushdown,
25    ToBatch, ToStream, generic,
26};
27use crate::catalog::source_catalog::SourceCatalog;
28use crate::error::Result;
29use crate::optimizer::plan_node::expr_visitable::ExprVisitable;
30use crate::optimizer::plan_node::utils::column_names_pretty;
31use crate::optimizer::plan_node::{
32    BatchIcebergScan, ColumnPruningContext, LogicalFilter, PredicatePushdownContext,
33    RewriteStreamContext, ToStreamContext,
34};
35use crate::utils::{ColIndexMapping, Condition};
36
37/// `LogicalIcebergScan` is only used by batch queries.
38/// It represents a scan of Iceberg data files, with delete files handled via anti-joins
39/// added on top of this scan.
40///
41/// The conversion flow is:
42/// 1. `LogicalSource` (iceberg) -> `LogicalIcebergIntermediateScan`
43/// 2. Predicate pushdown and column pruning on `LogicalIcebergIntermediateScan`
44/// 3. `LogicalIcebergIntermediateScan` -> `LogicalIcebergScan`
45#[derive(Debug, Clone, PartialEq)]
46pub struct LogicalIcebergScan {
47    pub base: PlanBase<Logical>,
48    pub core: generic::Source,
49    pub task: IcebergFileScanTask,
50}
51
52impl Eq for LogicalIcebergScan {}
53
54impl Hash for LogicalIcebergScan {
55    fn hash<H: Hasher>(&self, state: &mut H) {
56        self.base.hash(state);
57        self.core.hash(state);
58        self.iceberg_scan_type().hash(state);
59    }
60}
61
62impl LogicalIcebergScan {
63    pub fn new(core: generic::Source, task: IcebergFileScanTask) -> Self {
64        let base = PlanBase::new_logical_with_core(&core);
65        LogicalIcebergScan { base, core, task }
66    }
67
68    pub fn source_catalog(&self) -> Option<&SourceCatalog> {
69        self.core.catalog.as_deref()
70    }
71
72    pub fn iceberg_scan_type(&self) -> IcebergScanType {
73        self.task.scan_type()
74    }
75}
76
77impl_plan_tree_node_for_leaf! { Logical, LogicalIcebergScan}
78impl Distill for LogicalIcebergScan {
79    fn distill<'a>(&self) -> XmlNode<'a> {
80        let fields = if let Some(catalog) = self.source_catalog() {
81            let src = Pretty::from(catalog.name.clone());
82            vec![
83                ("source", src),
84                ("columns", column_names_pretty(self.schema())),
85                (
86                    "iceberg_scan_type",
87                    Pretty::debug(&self.iceberg_scan_type()),
88                ),
89            ]
90        } else {
91            vec![]
92        };
93        childless_record("LogicalIcebergScan", fields)
94    }
95}
96
97impl ColPrunable for LogicalIcebergScan {
98    fn prune_col(&self, _: &[usize], _: &mut ColumnPruningContext) -> PlanRef {
99        // Column pruning should have been done in LogicalIcebergIntermediateScan.
100        unreachable!()
101    }
102}
103
104impl ExprRewritable<Logical> for LogicalIcebergScan {}
105
106impl ExprVisitable for LogicalIcebergScan {}
107
108impl PredicatePushdown for LogicalIcebergScan {
109    fn predicate_pushdown(
110        &self,
111        predicate: Condition,
112        _ctx: &mut PredicatePushdownContext,
113    ) -> PlanRef {
114        // Predicate pushdown should have been done in LogicalIcebergIntermediateScan.
115        LogicalFilter::create(self.clone().into(), predicate)
116    }
117}
118
119impl ToBatch for LogicalIcebergScan {
120    fn to_batch(&self) -> Result<crate::optimizer::plan_node::BatchPlanRef> {
121        let plan = BatchIcebergScan::new(self.core.clone(), self.task.clone()).into();
122        Ok(plan)
123    }
124}
125
126impl ToStream for LogicalIcebergScan {
127    fn to_stream(
128        &self,
129        _ctx: &mut ToStreamContext,
130    ) -> Result<crate::optimizer::plan_node::StreamPlanRef> {
131        unreachable!()
132    }
133
134    fn logical_rewrite_for_stream(
135        &self,
136        _ctx: &mut RewriteStreamContext,
137    ) -> Result<(PlanRef, ColIndexMapping)> {
138        unreachable!()
139    }
140}
risingwave_frontend/optimizer/plan_node/logical_iceberg_scan.rs

risingwave_frontend/optimizer/plan_node/
logical_iceberg_scan.rs
