update

Author: luoluoyuyu

protocol/proto/fs_api.proto

@@ -12,6 +12,17 @@ message ConnectorOp {
   string description = 3;
 }
 
+message ValuePlanOperator {
+  string name = 1;
+  bytes physical_plan = 2;
+}
+
+message KeyPlanOperator {
+  string name = 1;
+  bytes physical_plan = 2;
+  repeated uint64 key_fields = 3;
+}
+
 message ProjectionOperator {
   string name = 1;
   FsSchema input_schema = 2;

src/coordinator/coordinator.rs

@@ -20,7 +20,7 @@ use crate::coordinator::execution::Executor;
 use crate::coordinator::plan::{LogicalPlanVisitor, LogicalPlanner, PlanNode};
 use crate::coordinator::statement::Statement;
 use crate::runtime::taskexecutor::TaskManager;
-use crate::sql::planner::StreamSchemaProvider;
+use crate::sql::schema::StreamSchemaProvider;
 
 use super::execution_context::ExecutionContext;
 

src/coordinator/execution/executor.rs

@@ -19,13 +19,11 @@ use crate::coordinator::plan::{
 };
 use crate::coordinator::statement::{ConfigSource, FunctionSource};
 use crate::runtime::taskexecutor::TaskManager;
+use crate::sql::schema::table::Table as CatalogTable;
+use crate::sql::analysis::{ StreamSchemaProvider};
 use std::sync::Arc;
 use thiserror::Error;
 use tracing::{debug, info};
-use crate::datastream::logical::{LogicalProgram, ProgramConfig};
-use crate::datastream::optimizers::ChainingOptimizer;
-use crate::sql::CompiledSql;
-use crate::sql::planner::{physical_planner, rewrite_sinks};
 
 #[derive(Error, Debug)]
 pub enum ExecuteError {
@@ -222,12 +220,21 @@ impl PlanVisitor for Executor {
 
     fn visit_streaming_table(
         &self,
-        _plan: &StreamingTable,
+        plan: &StreamingTable,
         _context: &PlanVisitorContext,
     ) -> PlanVisitorResult {
-        let result = Err(ExecuteError::Internal(
-            "StreamingTable execution not yet implemented".to_string(),
-        ));
+        let result = (|| -> Result<ExecuteResult, ExecuteError> {
+            let catalog_table =
+                CatalogTable::ConnectorTable(plan.connector_table.clone());
+            let mut schema_provider = StreamSchemaProvider::new();
+            schema_provider.insert_catalog_table(catalog_table.clone());
+
+
+            Ok(ExecuteResult::ok_with_data(
+                format!("Streaming table '{}' compiled successfully", plan.name),
+                empty_record_batch(),
+            ))
+        })();
         PlanVisitorResult::Execute(result)
     }
 

src/coordinator/plan/logical_plan_visitor.rs

@@ -13,8 +13,10 @@
 use std::sync::Arc;
 
 use datafusion::common::{Result, plan_datafusion_err, plan_err};
+use datafusion::execution::SessionStateBuilder;
 use datafusion::sql::sqlparser::ast::{SqlOption, Statement as DFStatement};
 use datafusion_common::TableReference;
+use datafusion_execution::config::SessionConfig;
 use datafusion_expr::{Expr, Extension, LogicalPlan, col};
 use sqlparser::ast::Statement;
 use tracing::debug;
@@ -30,21 +32,24 @@ use crate::coordinator::statement::{
     StreamingTableStatement,
 };
 use crate::coordinator::tool::ConnectorOptions;
-use crate::sql::catalog::Table;
-use crate::sql::catalog::connector::ConnectionType;
-use crate::sql::catalog::connector_table::ConnectorTable;
-use crate::sql::catalog::field_spec::FieldSpec;
-use crate::sql::catalog::optimizer::produce_optimized_plan;
+use crate::sql::logical_node::logical::{LogicalProgram, ProgramConfig};
+use crate::sql::logical_planner::optimizers::ChainingOptimizer;
+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::functions::{is_json_union, serialize_outgoing_json};
-use crate::sql::planner::extension::sink::SinkExtension;
-use crate::sql::planner::{StreamSchemaProvider, maybe_add_key_extension_to_sink, rewrite_sinks};
+use crate::sql::extensions::sink::SinkExtension;
+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";
 
-/// 将 WITH 选项列表转为 key-value map，便于读取 connector 等配置。
+/// Convert `WITH` option list to a key-value map (e.g. connector settings).
 fn with_options_to_map(options: &[SqlOption]) -> std::collections::HashMap<String, String> {
     options
         .iter()
@@ -153,6 +158,8 @@ impl LogicalPlanVisitor {
             primary_keys: Arc::new(vec![]), // PKs are inferred or explicitly set here
             inferred_fields: None,
             partition_exprs: Arc::new(partition_exprs),
+            lookup_cache_ttl:None,
+            lookup_cache_max_bytes:None,
         };
 
         // 6. Sink Extension & Final Rewrites
@@ -173,6 +180,37 @@ impl LogicalPlanVisitor {
         let final_extensions = rewrite_sinks(vec![plan_with_keys])?;
         let final_plan = final_extensions.into_iter().next().unwrap();
 
+
+
+        let mut config = SessionConfig::new();
+        config
+            .options_mut()
+            .optimizer
+            .enable_round_robin_repartition = false;
+        config.options_mut().optimizer.repartition_aggregations = false;
+        config.options_mut().optimizer.repartition_windows = false;
+        config.options_mut().optimizer.repartition_sorts = false;
+        config.options_mut().optimizer.repartition_joins = false;
+        config.options_mut().execution.target_partitions = 1;
+
+        let session_state = SessionStateBuilder::new()
+            .with_config(config)
+            .with_default_features()
+            .with_physical_optimizer_rules(vec![])
+            .build();
+
+        let mut plan_to_graph_visitor =
+            planner::PlanToGraphVisitor::new(&self.schema_provider, &session_state);
+
+        plan_to_graph_visitor.add_plan(final_plan.clone())?;
+
+        let graph = plan_to_graph_visitor.into_graph();
+
+        let mut program = LogicalProgram::new(graph, ProgramConfig::default());
+
+        program.optimize(&ChainingOptimizer {});
+
+
         Ok(Box::new(StreamingTable {
             name: table_name,
             comment: comment.clone(),

src/coordinator/plan/lookup_table_plan.rs

@@ -10,7 +10,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use crate::sql::catalog::connector_table::ConnectorTable;
+use crate::sql::schema::connector_table::ConnectorTable;
 
 use super::{PlanNode, PlanVisitor, PlanVisitorContext, PlanVisitorResult};
 

src/coordinator/plan/streaming_table_connector_plan.rs

@@ -10,7 +10,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use crate::sql::catalog::connector_table::ConnectorTable;
+use crate::sql::schema::connector_table::ConnectorTable;
 
 use super::{PlanNode, PlanVisitor, PlanVisitorContext, PlanVisitorResult};
 

src/coordinator/plan/streaming_table_plan.rs

@@ -11,7 +11,7 @@
 // limitations under the License.
 
 use super::{PlanNode, PlanVisitor, PlanVisitorContext, PlanVisitorResult};
-use crate::sql::catalog::connector_table::ConnectorTable;
+use crate::sql::schema::connector_table::ConnectorTable;
 use datafusion::logical_expr::LogicalPlan;
 
 /// Plan node representing a fully resolved streaming table (DDL).

src/datastream/mod.rs

@@ -17,7 +17,6 @@
 pub mod api;
 pub mod config;
 pub mod coordinator;
-pub mod datastream;
 pub mod logging;
 pub mod runtime;
 pub mod server;

src/lib.rs

@@ -15,7 +15,6 @@
 mod api;
 mod config;
 mod coordinator;
-mod datastream;
 mod logging;
 mod runtime;
 mod server;