update

Author: luoluoyuyu

Cargo.lock

@@ -13,9 +13,9 @@
 use super::Analysis;
 use crate::coordinator::execution_context::ExecutionContext;
 use crate::coordinator::statement::{
-    CreateFunction, CreatePythonFunction, CreateTable, DropFunction, InsertStatement,
-    ShowFunctions, StartFunction, Statement, StatementVisitor, StatementVisitorContext,
-    StatementVisitorResult, StopFunction,
+    CreateFunction, CreatePythonFunction, CreateTable, DropFunction, ShowFunctions, StartFunction,
+    Statement, StatementVisitor, StatementVisitorContext, StatementVisitorResult, StopFunction,
+    StreamingTableStatement,
 };
 use std::fmt;
 
@@ -125,11 +125,13 @@ impl StatementVisitor for Analyzer<'_> {
         StatementVisitorResult::Analyze(Box::new(CreateTable::new(stmt.statement.clone())))
     }
 
-    fn visit_insert_statement(
+    fn visit_streaming_table_statement(
         &self,
-        stmt: &InsertStatement,
+        stmt: &StreamingTableStatement,
         _context: &StatementVisitorContext,
     ) -> StatementVisitorResult {
-        StatementVisitorResult::Analyze(Box::new(InsertStatement::new(stmt.statement.clone())))
+        StatementVisitorResult::Analyze(Box::new(StreamingTableStatement::new(
+            stmt.statement.clone(),
+        )))
     }
 }

src/coordinator/analyze/analyzer.rs

@@ -13,8 +13,9 @@
 use crate::coordinator::dataset::{ExecuteResult, ShowFunctionsResult, empty_record_batch};
 use crate::coordinator::plan::{
     CreateFunctionPlan, CreatePythonFunctionPlan, CreateTablePlan, DropFunctionPlan,
-    InsertStatementPlan, PlanNode, PlanVisitor, PlanVisitorContext, PlanVisitorResult,
-    ShowFunctionsPlan, StartFunctionPlan, StopFunctionPlan,
+    LookupTablePlan, PlanNode, PlanVisitor, PlanVisitorContext, PlanVisitorResult,
+    ShowFunctionsPlan, StartFunctionPlan, StopFunctionPlan, StreamingTable,
+    StreamingTableConnectorPlan,
 };
 use crate::coordinator::statement::{ConfigSource, FunctionSource};
 use crate::runtime::taskexecutor::TaskManager;
@@ -215,17 +216,36 @@ impl PlanVisitor for Executor {
         PlanVisitorResult::Execute(result)
     }
 
-    fn visit_insert_statement_plan(
+    fn visit_streaming_table(
         &self,
-        plan: &InsertStatementPlan,
+        _plan: &StreamingTable,
         _context: &PlanVisitorContext,
     ) -> PlanVisitorResult {
-        // TODO: start streaming pipeline for INSERT / anonymous query
-        let result = Err(ExecuteError::Internal(format!(
-            "INSERT statement execution not yet implemented. Program graph has {} node(s), {} connection(s)",
-            plan.program.graph.node_count(),
-            plan.connection_ids.len(),
-        )));
+        let result = Err(ExecuteError::Internal(
+            "StreamingTable execution not yet implemented".to_string(),
+        ));
+        PlanVisitorResult::Execute(result)
+    }
+
+    fn visit_lookup_table(
+        &self,
+        _plan: &LookupTablePlan,
+        _context: &PlanVisitorContext,
+    ) -> PlanVisitorResult {
+        let result = Err(ExecuteError::Internal(
+            "LookupTable execution not yet implemented".to_string(),
+        ));
+        PlanVisitorResult::Execute(result)
+    }
+
+    fn visit_streaming_connector_table(
+        &self,
+        _plan: &StreamingTableConnectorPlan,
+        _context: &PlanVisitorContext,
+    ) -> PlanVisitorResult {
+        let result = Err(ExecuteError::Internal(
+            "StreamingTableConnector execution not yet implemented".to_string(),
+        ));
         PlanVisitorResult::Execute(result)
     }
 }

src/coordinator/execution/executor.rs

@@ -18,10 +18,11 @@ mod execution;
 mod execution_context;
 mod plan;
 mod statement;
+mod tool;
 
 pub use coordinator::Coordinator;
 pub use dataset::{DataSet, ShowFunctionsResult};
 pub use statement::{
-    CreateFunction, CreatePythonFunction, CreateTable, DropFunction, InsertStatement, PythonModule,
-    ShowFunctions, StartFunction, Statement, StopFunction,
+    CreateFunction, CreatePythonFunction, CreateTable, DropFunction, PythonModule, ShowFunctions,
+    StartFunction, Statement, StopFunction, StreamingTableStatement,
 };