feat(flow): use batching mode&fix sqlness

Cargo.toml

@@ -129,7 +129,7 @@ etcd-client = "0.14"
 fst = "0.4.7"
 futures = "0.3"
 futures-util = "0.3"
-greptime-proto = { git = "https://github.com/GreptimeTeam/greptime-proto.git", rev = "b6d9cffd43c4e6358805a798f17e03e232994b82" }
+greptime-proto = { git = "https://github.com/GreptimeTeam/greptime-proto.git", rev = "e82b0158cd38d4021edb4e4c0ae77f999051e62f" }
 hex = "0.4"
 http = "1"
 humantime = "2.1"

src/api/src/helper.rs

@@ -514,6 +514,7 @@ fn query_request_type(request: &QueryRequest) -> &'static str {
         Some(Query::Sql(_)) => "query.sql",
         Some(Query::LogicalPlan(_)) => "query.logical_plan",
         Some(Query::PromRangeQuery(_)) => "query.prom_range",
+        Some(Query::InsertIntoPlan(_)) => "query.insert_into_plan",
         None => "query.empty",
     }
 }

src/catalog/src/table_source.rs

@@ -27,7 +27,7 @@ use session::context::QueryContextRef;
 use snafu::{ensure, OptionExt, ResultExt};
 use table::metadata::TableType;
 use table::table::adapter::DfTableProviderAdapter;
-mod dummy_catalog;
+pub mod dummy_catalog;
 use dummy_catalog::DummyCatalogList;
 use table::TableRef;
 

src/cmd/src/flownode.rs

@@ -345,7 +345,7 @@ impl StartCommand {
         let client = Arc::new(NodeClients::new(channel_config));
 
         let invoker = FrontendInvoker::build_from(
-            flownode.flow_worker_manager().clone(),
+            flownode.flow_engine().streaming_engine(),
             catalog_manager.clone(),
             cached_meta_backend.clone(),
             layered_cache_registry.clone(),
@@ -355,7 +355,9 @@ impl StartCommand {
         .await
         .context(StartFlownodeSnafu)?;
         flownode
-            .flow_worker_manager()
+            .flow_engine()
+            .streaming_engine()
+            // TODO(discord9): refactor and avoid circular reference
             .set_frontend_invoker(invoker)
             .await;
 

src/cmd/src/standalone.rs

@@ -56,8 +56,8 @@ use datanode::datanode::{Datanode, DatanodeBuilder};
 use datanode::region_server::RegionServer;
 use file_engine::config::EngineConfig as FileEngineConfig;
 use flow::{
-    FlowConfig, FlowWorkerManager, FlownodeBuilder, FlownodeInstance, FlownodeOptions,
-    FrontendClient, FrontendInvoker,
+    FlowConfig, FlownodeBuilder, FlownodeInstance, FlownodeOptions, FrontendClient,
+    FrontendInvoker, GrpcQueryHandlerWithBoxedError, StreamingEngine,
 };
 use frontend::frontend::{Frontend, FrontendOptions};
 use frontend::instance::builder::FrontendBuilder;
@@ -524,17 +524,17 @@ impl StartCommand {
             ..Default::default()
         };
 
-        // TODO(discord9): for standalone not use grpc, but just somehow get a handler to frontend grpc client without
+        // for standalone not use grpc, but get a handler to frontend grpc client without
         // actually make a connection
-        let fe_server_addr = fe_opts.grpc.bind_addr.clone();
-        let frontend_client = FrontendClient::from_static_grpc_addr(fe_server_addr);
+        let (frontend_client, frontend_instance_handler) =
+            FrontendClient::from_empty_grpc_handler();
         let flow_builder = FlownodeBuilder::new(
             flownode_options,
             plugins.clone(),
             table_metadata_manager.clone(),
             catalog_manager.clone(),
             flow_metadata_manager.clone(),
-            Arc::new(frontend_client),
+            Arc::new(frontend_client.clone()),
         );
         let flownode = flow_builder
             .build()
@@ -544,15 +544,15 @@ impl StartCommand {
 
         // set the ref to query for the local flow state
         {
-            let flow_worker_manager = flownode.flow_worker_manager();
+            let flow_streaming_engine = flownode.flow_engine().streaming_engine();
             information_extension
-                .set_flow_worker_manager(flow_worker_manager.clone())
+                .set_flow_streaming_engine(flow_streaming_engine)
                 .await;
         }
 
         let node_manager = Arc::new(StandaloneDatanodeManager {
             region_server: datanode.region_server(),
-            flow_server: flownode.flow_worker_manager(),
+            flow_server: flownode.flow_engine(),
         });
 
         let table_id_sequence = Arc::new(
@@ -606,10 +606,19 @@ impl StartCommand {
         .context(error::StartFrontendSnafu)?;
         let fe_instance = Arc::new(fe_instance);
 
-        let flow_worker_manager = flownode.flow_worker_manager();
+        // set the frontend client for flownode
+        let grpc_handler = fe_instance.clone() as Arc<dyn GrpcQueryHandlerWithBoxedError>;
+        let weak_grpc_handler = Arc::downgrade(&grpc_handler);
+        frontend_instance_handler
+            .lock()
+            .unwrap()
+            .replace(weak_grpc_handler);
+
+        // set the frontend invoker for flownode
+        let flow_streaming_engine = flownode.flow_engine().streaming_engine();
         // flow server need to be able to use frontend to write insert requests back
         let invoker = FrontendInvoker::build_from(
-            flow_worker_manager.clone(),
+            flow_streaming_engine.clone(),
             catalog_manager.clone(),
             kv_backend.clone(),
             layered_cache_registry.clone(),
@@ -618,7 +627,7 @@ impl StartCommand {
         )
         .await
         .context(error::StartFlownodeSnafu)?;
-        flow_worker_manager.set_frontend_invoker(invoker).await;
+        flow_streaming_engine.set_frontend_invoker(invoker).await;
 
         let export_metrics_task = ExportMetricsTask::try_new(&opts.export_metrics, Some(&plugins))
             .context(error::ServersSnafu)?;
@@ -694,7 +703,7 @@ pub struct StandaloneInformationExtension {
     region_server: RegionServer,
     procedure_manager: ProcedureManagerRef,
     start_time_ms: u64,
-    flow_worker_manager: RwLock<Option<Arc<FlowWorkerManager>>>,
+    flow_streaming_engine: RwLock<Option<Arc<StreamingEngine>>>,
 }
 
 impl StandaloneInformationExtension {
@@ -703,14 +712,14 @@ impl StandaloneInformationExtension {
             region_server,
             procedure_manager,
             start_time_ms: common_time::util::current_time_millis() as u64,
-            flow_worker_manager: RwLock::new(None),
+            flow_streaming_engine: RwLock::new(None),
         }
     }
 
-    /// Set the flow worker manager for the standalone instance.
-    pub async fn set_flow_worker_manager(&self, flow_worker_manager: Arc<FlowWorkerManager>) {
-        let mut guard = self.flow_worker_manager.write().await;
-        *guard = Some(flow_worker_manager);
+    /// Set the flow streaming engine for the standalone instance.
+    pub async fn set_flow_streaming_engine(&self, flow_streaming_engine: Arc<StreamingEngine>) {
+        let mut guard = self.flow_streaming_engine.write().await;
+        *guard = Some(flow_streaming_engine);
     }
 }
 
@@ -789,7 +798,7 @@ impl InformationExtension for StandaloneInformationExtension {
 
     async fn flow_stats(&self) -> std::result::Result<Option<FlowStat>, Self::Error> {
         Ok(Some(
-            self.flow_worker_manager
+            self.flow_streaming_engine
                 .read()
                 .await
                 .as_ref()

src/common/meta/src/ddl/create_flow.rs

@@ -38,7 +38,7 @@ use table::metadata::TableId;
 use crate::cache_invalidator::Context;
 use crate::ddl::utils::{add_peer_context_if_needed, handle_retry_error};
 use crate::ddl::DdlContext;
-use crate::error::{self, Result};
+use crate::error::{self, Result, UnexpectedSnafu};
 use crate::instruction::{CacheIdent, CreateFlow};
 use crate::key::flow::flow_info::FlowInfoValue;
 use crate::key::flow::flow_route::FlowRouteValue;
@@ -171,7 +171,7 @@ impl CreateFlowProcedure {
         }
         self.data.state = CreateFlowState::CreateFlows;
         // determine flow type
-        self.data.flow_type = Some(determine_flow_type(&self.data.task));
+        self.data.flow_type = Some(get_flow_type_from_options(&self.data.task)?);
 
         Ok(Status::executing(true))
     }
@@ -196,8 +196,8 @@ impl CreateFlowProcedure {
             });
         }
         info!(
-            "Creating flow({:?}) on flownodes with peers={:?}",
-            self.data.flow_id, self.data.peers
+            "Creating flow({:?}, type={:?}) on flownodes with peers={:?}",
+            self.data.flow_id, self.data.flow_type, self.data.peers
         );
         join_all(create_flow)
             .await
@@ -306,8 +306,20 @@ impl Procedure for CreateFlowProcedure {
     }
 }
 
-pub fn determine_flow_type(_flow_task: &CreateFlowTask) -> FlowType {
-    FlowType::Batching
+pub fn get_flow_type_from_options(flow_task: &CreateFlowTask) -> Result<FlowType> {
+    let flow_type = flow_task
+        .flow_options
+        .get(FlowType::FLOW_TYPE_KEY)
+        .map(|s| s.as_str());
+    match flow_type {
+        Some(FlowType::BATCHING) => Ok(FlowType::Batching),
+        Some(FlowType::STREAMING) => Ok(FlowType::Streaming),
+        Some(unknown) => UnexpectedSnafu {
+            err_msg: format!("Unknown flow type: {}", unknown),
+        }
+        .fail(),
+        None => Ok(FlowType::Batching),
+    }
 }
 
 /// The state of [CreateFlowProcedure].

src/common/meta/src/ddl/tests/create_flow.rs

@@ -46,7 +46,7 @@ pub(crate) fn test_create_flow_task(
         create_if_not_exists,
         expire_after: Some(300),
         comment: "".to_string(),
-        sql: "raw_sql".to_string(),
+        sql: "select 1".to_string(),
         flow_options: Default::default(),
     }
 }

src/common/meta/src/error.rs

@@ -401,6 +401,13 @@ pub enum Error {
         location: Location,
     },
 
+    #[snafu(display("Invalid flow request body: {:?}", body))]
+    InvalidFlowRequestBody {
+        body: Box<Option<api::v1::flow::flow_request::Body>>,
+        #[snafu(implicit)]
+        location: Location,
+    },
+
     #[snafu(display("Failed to get kv cache, err: {}", err_msg))]
     GetKvCache { err_msg: String },
 
@@ -863,6 +870,7 @@ impl ErrorExt for Error {
             | InvalidUnsetDatabaseOption { .. }
             | InvalidTopicNamePrefix { .. }
             | InvalidTimeZone { .. } => StatusCode::InvalidArguments,
+            InvalidFlowRequestBody { .. } => StatusCode::InvalidArguments,
 
             FlowNotFound { .. } => StatusCode::FlowNotFound,
             FlowRouteNotFound { .. } => StatusCode::Unexpected,

src/common/query/src/logical_plan.rs

@@ -18,16 +18,19 @@ mod udaf;
 
 use std::sync::Arc;
 
+use api::v1::TableName;
 use datafusion::catalog::CatalogProviderList;
 use datafusion::error::Result as DatafusionResult;
 use datafusion::logical_expr::{LogicalPlan, LogicalPlanBuilder};
-use datafusion_common::Column;
-use datafusion_expr::col;
+use datafusion_common::{Column, TableReference};
+use datafusion_expr::dml::InsertOp;
+use datafusion_expr::{col, DmlStatement, WriteOp};
 pub use expr::{build_filter_from_timestamp, build_same_type_ts_filter};
+use snafu::ResultExt;
 
 pub use self::accumulator::{Accumulator, AggregateFunctionCreator, AggregateFunctionCreatorRef};
 pub use self::udaf::AggregateFunction;
-use crate::error::Result;
+use crate::error::{GeneralDataFusionSnafu, Result};
 use crate::logical_plan::accumulator::*;
 use crate::signature::{Signature, Volatility};
 
@@ -79,6 +82,74 @@ pub fn rename_logical_plan_columns(
     LogicalPlanBuilder::from(plan).project(projection)?.build()
 }
 
+/// Convert a insert into logical plan to an (table_name, logical_plan)
+/// where table_name is the name of the table to insert into.
+/// logical_plan is the plan to be executed.
+///
+/// if input logical plan is not `insert into table_name <input>`, return None
+///
+/// Returned TableName will use provided catalog and schema if not specified in the logical plan,
+/// if table scan in logical plan have full table name, will **NOT** override it.
+pub fn breakup_insert_plan(
+    plan: &LogicalPlan,
+    default_catalog: &str,
+    default_schema: &str,
+) -> Option<(TableName, Arc<LogicalPlan>)> {
+    if let LogicalPlan::Dml(dml) = plan {
+        if dml.op != WriteOp::Insert(InsertOp::Append) {
+            return None;
+        }
+        let table_name = &dml.table_name;
+        let table_name = match table_name {
+            TableReference::Bare { table } => TableName {
+                catalog_name: default_catalog.to_string(),
+                schema_name: default_schema.to_string(),
+                table_name: table.to_string(),
+            },
+            TableReference::Partial { schema, table } => TableName {
+                catalog_name: default_catalog.to_string(),
+                schema_name: schema.to_string(),
+                table_name: table.to_string(),
+            },
+            TableReference::Full {
+                catalog,
+                schema,
+                table,
+            } => TableName {
+                catalog_name: catalog.to_string(),
+                schema_name: schema.to_string(),
+                table_name: table.to_string(),
+            },
+        };
+        let logical_plan = dml.input.clone();
+        Some((table_name, logical_plan))
+    } else {
+        None
+    }
+}
+
+/// create a `insert into table_name <input>` logical plan
+pub fn add_insert_to_logical_plan(
+    table_name: TableName,
+    table_schema: datafusion_common::DFSchemaRef,
+    input: LogicalPlan,
+) -> Result<LogicalPlan> {
+    let table_name = TableReference::Full {
+        catalog: table_name.catalog_name.into(),
+        schema: table_name.schema_name.into(),
+        table: table_name.table_name.into(),
+    };
+
+    let plan = LogicalPlan::Dml(DmlStatement::new(
+        table_name,
+        table_schema,
+        WriteOp::Insert(InsertOp::Append),
+        Arc::new(input),
+    ));
+    let plan = plan.recompute_schema().context(GeneralDataFusionSnafu)?;
+    Ok(plan)
+}
+
 /// The datafusion `[LogicalPlan]` decoder.
 #[async_trait::async_trait]
 pub trait SubstraitPlanDecoder {