update

Author: luoluoyuyu

Cargo.lock

@@ -14,6 +14,7 @@ members = [
     "src/servicer",
     "src/sqlparser",
     "src/streaming_runtime",
+    "src/streaming_planner",
     "src/wasm_runtime",
 ]
 
@@ -50,6 +51,7 @@ protocol = { path = "./protocol" }
 function-stream-config = { path = "src/config" }
 function-stream-logger = { path = "src/logger" }
 function-stream-runtime-common = { path = "src/runtime_common" }
+function-stream-streaming-planner = { path = "src/streaming_planner" }
 prost = "0.13"
 rdkafka = { version = "0.38", features = ["cmake-build", "ssl", "gssapi", "curl"] }
 crossbeam-channel = "0.5"

Cargo.toml

@@ -1,7 +1,7 @@
 //! Persistent catalog storage implementations.
 //!
 //! The stream catalog manager and task persistence (`stream_catalog/`, `task/`) live in this
-//! package and are compiled as part of `function-stream` via `#[path]` in `src/storage/mod.rs`.
+//! package and are compiled as part of `function-stream` via `#[path]` in `src/storage.rs`.
 
 pub mod memory;
 pub mod rocksdb;

src/catalog_storage/src/lib.rs

@@ -13,13 +13,13 @@
 mod analyze;
 #[allow(clippy::module_inception)]
 mod coordinator;
-mod sql_classify;
 mod dataset;
 mod execution;
 mod execution_context;
 mod plan;
 mod runtime_context;
 mod statement;
+mod sql_classify;
 mod streaming_table_options;
 mod tool;
 

src/coordinator/src/legacy/mod.rs

@@ -15,13 +15,13 @@
 
 use std::collections::HashMap;
 
-use datafusion::common::{Result, plan_err};
+use datafusion::common::{plan_err, Result};
 use datafusion::error::DataFusionError;
 use datafusion::sql::sqlparser::ast::{
     ObjectType, ShowCreateObject, SqlOption, Statement as DFStatement,
 };
 
-use super::statement::{
+use super::{
     CreateFunction, CreateTable, DropFunction, DropStreamingTableStatement, DropTableStatement,
     ShowCatalogTables, ShowCreateStreamingTable, ShowCreateTable, ShowFunctions,
     ShowStreamingTables, StartFunction, Statement, StopFunction, StreamingTableStatement,
@@ -298,8 +298,7 @@ mod tests {
 
     #[test]
     fn test_insert_not_supported() {
-        let mut stmts =
-            parse_sql("INSERT INTO sink SELECT * FROM src").unwrap();
+        let mut stmts = parse_sql("INSERT INTO sink SELECT * FROM src").unwrap();
         let err = classify_statement(stmts.remove(0)).unwrap_err();
         let msg = err.to_string();
         assert!(

src/coordinator/src/legacy/sql_classify.rs

@@ -21,5 +21,5 @@ pub use function_stream_logger as logging;
 pub mod runtime;
 #[path = "servicer/src/legacy/mod.rs"]
 pub mod server;
-pub mod sql;
+pub use function_stream_streaming_planner as sql;
 pub mod storage;

src/lib.rs

@@ -19,7 +19,7 @@ pub use function_stream_logger as logging;
 mod runtime;
 #[path = "servicer/src/legacy/mod.rs"]
 mod server;
-mod sql;
+pub use function_stream_streaming_planner as sql;
 mod storage;
 
 use anyhow::{Context, Result};

src/main.rs

@@ -9,6 +9,7 @@ path = "src/lib.rs"
 
 [dependencies]
 arrow-array = "55"
+bincode = { version = "2", features = ["serde"] }
 parking_lot = "0.12"
 serde = { version = "1.0", features = ["derive"] }
 tokio = { version = "1.0", features = ["sync"] }

src/runtime/mod.rs

@@ -14,3 +14,4 @@
 
 pub mod common;
 pub mod memory;
+pub mod streaming_protocol;