update

Author: luoluoyuyu

src/coordinator/execution/executor.rs

@@ -225,7 +225,7 @@ impl PlanVisitor for Executor {
     ) -> PlanVisitorResult {
         let result = (|| -> Result<ExecuteResult, ExecuteError> {
             let catalog_table =
-                CatalogTable::ConnectorTable(plan.connector_table.clone());
+                CatalogTable::ConnectorTable(plan.source_table.clone());
             let mut schema_provider = StreamSchemaProvider::new();
             schema_provider.insert_catalog_table(catalog_table.clone());
 

src/coordinator/plan/logical_plan_visitor.rs

@@ -33,21 +33,19 @@ use crate::coordinator::statement::{
 };
 use crate::coordinator::tool::ConnectorOptions;
 use crate::sql::logical_node::logical::{LogicalProgram, ProgramConfig};
-use crate::sql::logical_planner::optimizers::ChainingOptimizer;
+use crate::sql::logical_planner::optimizers::{ChainingOptimizer, produce_optimized_plan};
 use crate::sql::schema::Table;
-use crate::sql::schema::connector::ConnectionType;
-use crate::sql::schema::connector_table::ConnectorTable;
-use crate::sql::schema::field_spec::FieldSpec;
-use crate::sql::schema::optimizer::produce_optimized_plan;
+use crate::sql::schema::ConnectionType;
+use crate::sql::schema::source_table::SourceTable;
+use crate::sql::schema::ColumnDescriptor;
 use crate::sql::functions::{is_json_union, serialize_outgoing_json};
-use crate::sql::extensions::sink::SinkExtension;
+use crate::sql::extensions::sink::StreamEgressNode;
 use crate::sql::logical_planner::planner;
 use crate::sql::analysis::{StreamSchemaProvider, maybe_add_key_extension_to_sink, rewrite_sinks};
 use crate::sql::rewrite_plan;
 
 const CONNECTOR: &str = "connector";
 const PARTITION_BY: &str = "partition_by";
-const IDLE_MICROS: &str = "idle_time";
 
 fn with_options_to_map(options: &[SqlOption]) -> std::collections::HashMap<String, String> {
     options
@@ -108,6 +106,8 @@ impl LogicalPlanVisitor {
             )
         })?;
 
+        let partition_exprs = self.resolve_partition_expressions(&mut opts)?;
+
         let base_plan =
             produce_optimized_plan(&Statement::Query(query.clone()), &self.schema_provider)?;
         let mut plan = rewrite_plan(base_plan, &self.schema_provider)?;
@@ -121,38 +121,33 @@ impl LogicalPlanVisitor {
             plan = serialize_outgoing_json(&self.schema_provider, Arc::new(plan));
         }
 
-        let partition_exprs = self.resolve_partition_expressions(&mut opts)?;
-
-        let fields: Vec<FieldSpec> = plan
+        let fields: Vec<ColumnDescriptor> = plan
             .schema()
             .fields()
             .iter()
-            .map(|f| FieldSpec::Struct((**f).clone()))
+            .map(|f| ColumnDescriptor::from((**f).clone()))
             .collect();
 
-        let connector_table = ConnectorTable {
-            id: None,
-            connector,
-            name: table_name.clone(),
-            connection_type: ConnectionType::Sink,
+        let mut source_table = SourceTable::from_options(
+            &table_name,
+            &connector,
+            false,
             fields,
-            config: "".to_string(),
-            description: comment.clone().unwrap_or_default(),
-            event_time_field: None,
-            watermark_field: None,
-            idle_time: opts.pull_opt_duration(IDLE_MICROS)?,
-            primary_keys: Arc::new(vec![]),
-            inferred_fields: None,
-            partition_exprs: Arc::new(partition_exprs),
-            lookup_cache_ttl:None,
-            lookup_cache_max_bytes:None,
-        };
+            vec![],
+            None,
+            &mut opts,
+            None,
+            &self.schema_provider,
+            Some(ConnectionType::Sink),
+            comment.clone().unwrap_or_default(),
+        )?;
+        source_table.partition_exprs = Arc::new(partition_exprs);
 
-        let sink_extension = SinkExtension::new(
+        let sink_extension = StreamEgressNode::try_new(
             TableReference::bare(table_name.clone()),
-            Table::ConnectorTable(connector_table.clone()),
+            Table::ConnectorTable(source_table.clone()),
             plan.schema().clone(),
-            Arc::new(plan),
+            plan,
         )?;
 
         let plan_with_keys = maybe_add_key_extension_to_sink(LogicalPlan::Extension(Extension {
@@ -196,7 +191,7 @@ impl LogicalPlanVisitor {
         Ok(Box::new(StreamingTable {
             name: table_name,
             comment: comment.clone(),
-            connector_table,
+            source_table,
             logical_plan: final_plan,
         }))
     }
@@ -322,7 +317,7 @@ mod create_streaming_table_tests {
 
     use crate::sql::common::TIMESTAMP_FIELD;
     use crate::sql::rewrite_plan;
-    use crate::sql::schema::optimizer::produce_optimized_plan;
+    use crate::sql::logical_planner::optimizers::produce_optimized_plan;
     use crate::sql::schema::StreamSchemaProvider;
 
     fn schema_provider_with_src() -> StreamSchemaProvider {

src/coordinator/plan/lookup_table_plan.rs

@@ -10,14 +10,14 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use crate::sql::schema::connector_table::ConnectorTable;
+use crate::sql::schema::source_table::SourceTable;
 
 use super::{PlanNode, PlanVisitor, PlanVisitorContext, PlanVisitorResult};
 
 /// Plan node that exposes a lookup table config as a logical plan input.
 #[derive(Debug)]
 pub struct LookupTablePlan {
-    pub table: ConnectorTable,
+    pub table: SourceTable,
 }
 
 impl PlanNode for LookupTablePlan {

src/coordinator/plan/streaming_table_connector_plan.rs

@@ -10,14 +10,14 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use crate::sql::schema::connector_table::ConnectorTable;
+use crate::sql::schema::source_table::SourceTable;
 
 use super::{PlanNode, PlanVisitor, PlanVisitorContext, PlanVisitorResult};
 
 /// Plan node that exposes a connector table config as a logical plan input.
 #[derive(Debug)]
 pub struct StreamingTableConnectorPlan {
-    pub table: ConnectorTable,
+    pub table: SourceTable,
 }
 
 impl PlanNode for StreamingTableConnectorPlan {

src/coordinator/plan/streaming_table_plan.rs

@@ -11,15 +11,15 @@
 // limitations under the License.
 
 use super::{PlanNode, PlanVisitor, PlanVisitorContext, PlanVisitorResult};
-use crate::sql::schema::connector_table::ConnectorTable;
+use crate::sql::schema::source_table::SourceTable;
 use datafusion::logical_expr::LogicalPlan;
 
 /// Plan node representing a fully resolved streaming table (DDL).
 #[derive(Debug)]
 pub struct StreamingTable {
     pub name: String,
     pub comment: Option<String>,
-    pub connector_table: ConnectorTable,
+    pub source_table: SourceTable,
     pub logical_plan: LogicalPlan,
 }
 

src/coordinator/tool/mod.rs

@@ -1,3 +1 @@
-mod connector_options;
-
-pub use connector_options::{ConnectorOptions, FromOpts};
+pub use crate::sql::common::ConnectorOptions;

src/sql/analysis/aggregate_rewriter.rs

@@ -1,13 +1,13 @@
 use datafusion::common::tree_node::{Transformed, TreeNodeRewriter};
 use datafusion::common::{DFSchema, DataFusionError, Result, not_impl_err, plan_err};
 use datafusion::functions_aggregate::expr_fn::max;
-use datafusion::logical_expr::{self, Aggregate, Expr, Extension, LogicalPlan, Projection};
+use datafusion::logical_expr::{Aggregate, Expr, Extension, LogicalPlan, Projection};
 use datafusion::prelude::col;
 use std::sync::Arc;
 
 use crate::sql::schema::StreamSchemaProvider;
-use crate::sql::extensions::aggregate::AggregateExtension;
-use crate::sql::extensions::key_calculation::{KeyCalculationExtension, KeysOrExprs};
+use crate::sql::extensions::aggregate::StreamWindowAggregateNode;
+use crate::sql::extensions::key_calculation::{KeyExtractionNode, KeyExtractionStrategy};
 use crate::sql::analysis::streaming_window_analzer::StreamingWindowAnalzer;
 use crate::sql::types::{
     DFField, TIMESTAMP_FIELD, WindowBehavior, WindowType, fields_with_qualifiers, find_window,
@@ -83,7 +83,7 @@ impl TreeNodeRewriter for AggregateRewriter<'_> {
         let keyed_input =
             self.build_keyed_input(agg.input.clone(), &agg.group_expr, &key_fields)?;
 
-        // 5. Build the final AggregateExtension for the physical planner.
+        // 5. Build the final StreamWindowAggregateNode for the physical planner.
         let mut internal_fields = fields_with_qualifiers(&agg.schema);
         if let WindowBehavior::FromOperator { window_index, .. } = &behavior {
             internal_fields.remove(*window_index);
@@ -100,11 +100,11 @@ impl TreeNodeRewriter for AggregateRewriter<'_> {
             internal_schema,
         )?;
 
-        let extension = AggregateExtension::new(
+        let extension = StreamWindowAggregateNode::try_new(
             behavior,
             LogicalPlan::Aggregate(rewritten_agg),
             (0..key_count).collect(),
-        );
+        )?;
 
         Ok(Transformed::yes(LogicalPlan::Extension(Extension {
             node: Arc::new(extension),
@@ -118,7 +118,7 @@ impl<'a> AggregateRewriter<'a> {
     }
 
     /// [Internal] Builds the physical Key Calculation layer required for distributed Shuffling.
-    /// This wraps the input in a Projection and a KeyCalculationExtension.
+    /// This wraps the input in a Projection and a KeyExtractionNode.
     fn build_keyed_input(
         &self,
         input: Arc<LogicalPlan>,
@@ -151,9 +151,9 @@ impl<'a> AggregateRewriter<'a> {
             LogicalPlan::Projection(Projection::try_new_with_schema(exprs, input, key_schema)?);
 
         Ok(LogicalPlan::Extension(Extension {
-            node: Arc::new(KeyCalculationExtension::new(
+            node: Arc::new(KeyExtractionNode::new(
                 projection,
-                KeysOrExprs::Keys((0..key_count).collect()),
+                KeyExtractionStrategy::ColumnIndices((0..key_count).collect()),
             )),
         }))
     }

src/sql/analysis/async_udf_rewriter.rs

@@ -1,5 +1,5 @@
-use crate::sql::extensions::remote_table::RemoteTableExtension;
-use crate::sql::extensions::{ASYNC_RESULT_FIELD, AsyncUDFExtension};
+use crate::sql::extensions::remote_table::RemoteTableBoundaryNode;
+use crate::sql::extensions::{ASYNC_RESULT_FIELD, AsyncFunctionExecutionNode};
 use crate::sql::schema::StreamSchemaProvider;
 use datafusion::common::tree_node::{Transformed, TreeNode, TreeNodeRewriter};
 use datafusion::common::{Column, Result as DFResult, TableReference, plan_err};
@@ -92,28 +92,28 @@ impl TreeNodeRewriter for AsyncUdfRewriter<'_> {
 
         let input = if matches!(*projection.input, LogicalPlan::Projection(..)) {
             Arc::new(LogicalPlan::Extension(Extension {
-                node: Arc::new(RemoteTableExtension {
-                    input: (*projection.input).clone(),
-                    name: TableReference::bare("subquery_projection"),
-                    schema: projection.input.schema().clone(),
-                    materialize: false,
+                node: Arc::new(RemoteTableBoundaryNode {
+                    upstream_plan: (*projection.input).clone(),
+                    table_identifier: TableReference::bare("subquery_projection"),
+                    resolved_schema: projection.input.schema().clone(),
+                    requires_materialization: false,
                 }),
             }))
         } else {
             projection.input
         };
 
         Ok(Transformed::yes(LogicalPlan::Extension(Extension {
-            node: Arc::new(AsyncUDFExtension {
-                input,
-                name,
-                udf,
-                arg_exprs,
-                final_exprs: projection.expr,
-                ordered: opts.ordered,
-                max_concurrency: opts.max_concurrency,
-                timeout: opts.timeout,
-                final_schema: projection.schema,
+            node: Arc::new(AsyncFunctionExecutionNode {
+                upstream_plan: input,
+                operator_name: name,
+                function_config: udf,
+                invocation_args: arg_exprs,
+                result_projections: projection.expr,
+                preserve_ordering: opts.ordered,
+                concurrency_limit: opts.max_concurrency,
+                execution_timeout: opts.timeout,
+                resolved_schema: projection.schema,
             }),
         })))
     }

src/sql/analysis/join_rewriter.rs

@@ -1,6 +1,6 @@
 use crate::sql::schema::StreamSchemaProvider;
-use crate::sql::extensions::join::JoinExtension;
-use crate::sql::extensions::key_calculation::KeyCalculationExtension;
+use crate::sql::extensions::join::StreamingJoinNode;
+use crate::sql::extensions::key_calculation::KeyExtractionNode;
 use crate::sql::analysis::streaming_window_analzer::StreamingWindowAnalzer;
 use crate::sql::types::{WindowType, fields_with_qualifiers, schema_from_df_fields_with_metadata};
 use crate::sql::common::TIMESTAMP_FIELD;
@@ -62,7 +62,7 @@ impl<'a> JoinRewriter<'a> {
         }
     }
 
-    /// [Internal] Wraps a join input in a KeyCalculation layer to facilitate Shuffle/KeyBy distribution.
+    /// [Internal] Wraps a join input in a key-extraction layer to facilitate shuffle / key-by distribution.
     fn build_keyed_side(
         &self,
         input: Arc<LogicalPlan>,
@@ -85,11 +85,11 @@ impl<'a> JoinRewriter<'a> {
             .collect();
 
         let projection = Projection::try_new(projection_exprs, input)?;
-        let key_ext = KeyCalculationExtension::new_named_and_trimmed(
+        let key_ext = KeyExtractionNode::try_new_with_projection(
             LogicalPlan::Projection(projection),
             (0..key_count).collect(),
             side.to_string(),
-        );
+        )?;
 
         Ok(LogicalPlan::Extension(Extension {
             node: Arc::new(key_ext),
@@ -209,13 +209,13 @@ impl TreeNodeRewriter for JoinRewriter<'_> {
         // 4. Resolve Output Watermark (Timestamp Projection)
         let plan_with_timestamp = self.apply_timestamp_resolution(rewritten_join)?;
 
-        // 5. Wrap in JoinExtension for Physical Planning
-        let ttl = (!is_instant).then_some(self.schema_provider.planning_options.ttl);
-        let extension = JoinExtension {
-            rewritten_join: plan_with_timestamp,
+        // 5. Wrap in StreamingJoinNode for physical planning
+        let state_retention_ttl = (!is_instant).then_some(self.schema_provider.planning_options.ttl);
+        let extension = StreamingJoinNode::new(
+            plan_with_timestamp,
             is_instant,
-            ttl,
-        };
+            state_retention_ttl,
+        );
 
         Ok(Transformed::yes(LogicalPlan::Extension(Extension {
             node: Arc::new(extension),