refactor(iceberg): Separate iceberg source pb from source pb

proto/batch_plan.proto

@@ -66,12 +66,11 @@ message SourceNode {
   map<string, secret.SecretRef> secret_refs = 6;
 }
 
-message IcebergSourceNode {
-  uint32 source_id = 1;
-  repeated plan_common.ColumnCatalog columns = 2;
-  map<string, string> with_properties = 3;
-  repeated bytes split = 4;
-  map<string, secret.SecretRef> secret_refs = 5;
+message IcebergScanNode {
+  repeated plan_common.ColumnCatalog columns = 1;
+  map<string, string> with_properties = 2;
+  repeated bytes split = 3;
+  map<string, secret.SecretRef> secret_refs = 4;
 }
 
 message FileScanNode {
@@ -373,7 +372,7 @@ message PlanNode {
     MaxOneRowNode max_one_row = 36;
     LogRowSeqScanNode log_row_seq_scan = 37;
     FileScanNode file_scan = 38;
-    IcebergSourceNode iceberg_source = 39;
+    IcebergScanNode iceberg_scan = 39;
     // The following nodes are used for testing.
     bool block_executor = 100;
     bool busy_loop_executor = 101;

src/batch/src/executor/iceberg_scan.rs

@@ -131,7 +131,7 @@ impl BoxedExecutorBuilder for IcebergScanExecutorBuilder {
         );
         let source_node = try_match_expand!(
             source.plan_node().get_node_body().unwrap(),
-            NodeBody::IcebergSource
+            NodeBody::IcebergScan
         )?;
 
         // prepare connector source

src/batch/src/executor/mod.rs

@@ -243,7 +243,7 @@ impl<'a, C: BatchTaskContext> ExecutorBuilder<'a, C> {
             NodeBody::SortOverWindow => SortOverWindowExecutor,
             NodeBody::MaxOneRow => MaxOneRowExecutor,
             NodeBody::FileScan => FileScanExecutorBuilder,
-            NodeBody::IcebergSource => IcebergScanExecutorBuilder,
+            NodeBody::IcebergScan => IcebergScanExecutorBuilder,
             // Follow NodeBody only used for test
             NodeBody::BlockExecutor => BlockExecutorBuilder,
             NodeBody::BusyLoopExecutor => BusyLoopExecutorBuilder,

src/frontend/src/optimizer/plan_node/batch_iceberg_scan.rs

@@ -16,7 +16,7 @@ use std::rc::Rc;
 
 use pretty_xmlish::{Pretty, XmlNode};
 use risingwave_pb::batch_plan::plan_node::NodeBody;
-use risingwave_pb::batch_plan::IcebergSourceNode;
+use risingwave_pb::batch_plan::IcebergScanNode;
 use risingwave_sqlparser::ast::AsOf;
 
 use super::batch::prelude::*;
@@ -99,8 +99,7 @@ impl ToBatchPb for BatchIcebergScan {
     fn to_batch_prost_body(&self) -> NodeBody {
         let source_catalog = self.source_catalog().unwrap();
         let (with_properties, secret_refs) = source_catalog.with_properties.clone().into_parts();
-        NodeBody::IcebergSource(IcebergSourceNode {
-            source_id: source_catalog.id,
+        NodeBody::IcebergScan(IcebergScanNode {
             columns: self
                 .core
                 .column_catalog

src/frontend/src/scheduler/distributed/stage.rs

@@ -1052,36 +1052,44 @@ impl StageRunner {
                     node_body: Some(NodeBody::LogRowSeqScan(scan_node)),
                 }
             }
-            PlanNodeType::BatchSource
-            | PlanNodeType::BatchKafkaScan
-            | PlanNodeType::BatchIcebergScan => {
+            PlanNodeType::BatchSource | PlanNodeType::BatchKafkaScan => {
                 let node_body = execution_plan_node.node.clone();
+                let NodeBody::Source(mut source_node) = node_body else {
+                    unreachable!();
+                };
+
                 let partition = partition
                     .expect("no partition info for seq scan")
                     .into_source()
                     .expect("PartitionInfo should be SourcePartitionInfo");
-                if let NodeBody::Source(mut source_node) = node_body {
-                    source_node.split = partition
-                        .into_iter()
-                        .map(|split| split.encode_to_bytes().into())
-                        .collect_vec();
-                    PbPlanNode {
-                        children: vec![],
-                        identity,
-                        node_body: Some(NodeBody::Source(source_node)),
-                    }
-                } else if let NodeBody::IcebergSource(mut source_node) = node_body {
-                    source_node.split = partition
-                        .into_iter()
-                        .map(|split| split.encode_to_bytes().into())
-                        .collect_vec();
-                    PbPlanNode {
-                        children: vec![],
-                        identity,
-                        node_body: Some(NodeBody::IcebergSource(source_node)),
-                    }
-                } else {
+                source_node.split = partition
+                    .into_iter()
+                    .map(|split| split.encode_to_bytes().into())
+                    .collect_vec();
+                PbPlanNode {
+                    children: vec![],
+                    identity,
+                    node_body: Some(NodeBody::Source(source_node)),
+                }
+            }
+            PlanNodeType::BatchIcebergScan => {
+                let node_body = execution_plan_node.node.clone();
+                let NodeBody::IcebergScan(mut iceberg_source_node) = node_body else {
                     unreachable!();
+                };
+
+                let partition = partition
+                    .expect("no partition info for seq scan")
+                    .into_source()
+                    .expect("PartitionInfo should be SourcePartitionInfo");
+                iceberg_source_node.split = partition
+                    .into_iter()
+                    .map(|split| split.encode_to_bytes().into())
+                    .collect_vec();
+                PbPlanNode {
+                    children: vec![],
+                    identity,
+                    node_body: Some(NodeBody::IcebergScan(iceberg_source_node)),
                 }
             }
             _ => {

src/frontend/src/scheduler/local.rs

@@ -554,12 +554,10 @@ impl LocalQueryExecution {
                     node_body: Some(node_body),
                 })
             }
-            PlanNodeType::BatchSource
-            | PlanNodeType::BatchKafkaScan
-            | PlanNodeType::BatchIcebergScan => {
-                let node_body = execution_plan_node.node.clone();
-                match node_body {
-                    NodeBody::Source(mut source_node) => {
+            PlanNodeType::BatchSource | PlanNodeType::BatchKafkaScan => {
+                let mut node_body = execution_plan_node.node.clone();
+                match &mut node_body {
+                    NodeBody::Source(ref mut source_node) => {
                         if let Some(partition) = partition {
                             let partition = partition
                                 .into_source()
@@ -569,13 +567,20 @@ impl LocalQueryExecution {
                                 .map(|split| split.encode_to_bytes().into())
                                 .collect_vec();
                         }
-                        Ok(PbPlanNode {
-                            children: vec![],
-                            identity,
-                            node_body: Some(NodeBody::Source(source_node)),
-                        })
                     }
-                    NodeBody::IcebergSource(mut iceberg_source_node) => {
+                    _ => unreachable!(),
+                }
+
+                Ok(PbPlanNode {
+                    children: vec![],
+                    identity,
+                    node_body: Some(node_body),
+                })
+            }
+            PlanNodeType::BatchIcebergScan => {
+                let mut node_body = execution_plan_node.node.clone();
+                match &mut node_body {
+                    NodeBody::IcebergScan(ref mut iceberg_source_node) => {
                         if let Some(partition) = partition {
                             let partition = partition
                                 .into_source()
@@ -585,15 +590,15 @@ impl LocalQueryExecution {
                                 .map(|split| split.encode_to_bytes().into())
                                 .collect_vec();
                         }
-
-                        Ok(PbPlanNode {
-                            children: vec![],
-                            identity,
-                            node_body: Some(NodeBody::IcebergSource(iceberg_source_node)),
-                        })
                     }
                     _ => unreachable!(),
                 }
+
+                Ok(PbPlanNode {
+                    children: vec![],
+                    identity,
+                    node_body: Some(node_body),
+                })
             }
             PlanNodeType::BatchLookupJoin => {
                 let mut node_body = execution_plan_node.node.clone();

src/prost/build.rs

@@ -80,7 +80,7 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
         ".stream_plan.SourceBackfillNode",
         ".stream_plan.StreamSource",
         ".batch_plan.SourceNode",
-        ".batch_plan.IcebergSourceNode",
+        ".batch_plan.IcebergScanNode",
     ];
 
     // Build protobuf structs.