fix(iceberg): only convert iceberg table to iceberg source for batch …

…dql (#20045)
risingwavelabs · Jan 7, 2025 · ec36f40 · ec36f40
1 parent 2497760
commit ec36f40
Show file tree

Hide file tree

Showing 3 changed files with 21 additions and 4 deletions.
diff --git a/src/frontend/src/handler/query.rs b/src/frontend/src/handler/query.rs
@@ -247,7 +247,11 @@ fn gen_batch_query_plan(
         ..
     } = bind_result;
 
-    let mut planner = Planner::new_for_batch(context);
+    let mut planner = if matches!(bound, BoundStatement::Query(_)) {
+        Planner::new_for_batch_dql(context)
+    } else {
+        Planner::new_for_batch(context)
+    };
 
     let mut logical = planner.plan(bound)?;
     let schema = logical.schema();

diff --git a/src/frontend/src/planner/mod.rs b/src/frontend/src/planner/mod.rs
@@ -48,9 +48,20 @@ pub struct Planner {
 pub enum PlanFor {
     Stream,
     Batch,
+    /// `BatchDql` is a special mode for batch.
+    /// Iceberg engine table will be converted to iceberg source based on this mode.
+    BatchDql,
 }
 
 impl Planner {
+    pub fn new_for_batch_dql(ctx: OptimizerContextRef) -> Planner {
+        Planner {
+            ctx,
+            share_cache: Default::default(),
+            plan_for: PlanFor::BatchDql,
+        }
+    }
+
     pub fn new_for_batch(ctx: OptimizerContextRef) -> Planner {
         Planner {
             ctx,

diff --git a/src/frontend/src/planner/relation.rs b/src/frontend/src/planner/relation.rs
@@ -92,7 +92,9 @@ impl Planner {
         );
 
         match (base_table.table_catalog.engine, self.plan_for()) {
-            (Engine::Hummock, PlanFor::Stream) | (Engine::Hummock, PlanFor::Batch) => {
+            (Engine::Hummock, PlanFor::Stream)
+            | (Engine::Hummock, PlanFor::Batch)
+            | (Engine::Hummock, PlanFor::BatchDql) => {
                 match as_of {
                     None
                     | Some(AsOf::ProcessTime)
@@ -105,7 +107,7 @@ impl Planner {
                 };
                 Ok(scan.into())
             }
-            (Engine::Iceberg, PlanFor::Stream) => {
+            (Engine::Iceberg, PlanFor::Stream) | (Engine::Iceberg, PlanFor::Batch) => {
                 match as_of {
                     None
                     | Some(AsOf::VersionNum(_))
@@ -120,7 +122,7 @@ impl Planner {
                 }
                 Ok(scan.into())
             }
-            (Engine::Iceberg, PlanFor::Batch) => {
+            (Engine::Iceberg, PlanFor::BatchDql) => {
                 match as_of {
                     None
                     | Some(AsOf::VersionNum(_))