feat(flow): flow node manager

src/cmd/Cargo.toml

@@ -46,6 +46,7 @@ datatypes.workspace = true
 either = "1.8"
 etcd-client.workspace = true
 file-engine.workspace = true
+flow.workspace = true
 frontend.workspace = true
 futures.workspace = true
 human-panic = "1.2.2"

src/cmd/src/standalone.rs

@@ -45,6 +45,7 @@ use common_wal::config::StandaloneWalConfig;
 use datanode::config::{DatanodeOptions, ProcedureConfig, RegionEngineConfig, StorageConfig};
 use datanode::datanode::{Datanode, DatanodeBuilder};
 use file_engine::config::EngineConfig as FileEngineConfig;
+use flow::FlownodeBuilder;
 use frontend::error::{Result as FeResult, TomlFormatSnafu};
 use frontend::frontend::FrontendOptions;
 use frontend::instance::builder::FrontendBuilder;
@@ -418,11 +419,26 @@ impl StartCommand {
         )
         .await;
 
+        let table_metadata_manager =
+            Self::create_table_metadata_manager(kv_backend.clone()).await?;
+
+        let flow_builder = FlownodeBuilder::new(
+            Default::default(),
+            fe_plugins.clone(),
+            table_metadata_manager.clone(),
+            catalog_manager.clone(),
+        )
+        .with_kv_backend(kv_backend.clone());
+        let flownode = Arc::new(flow_builder.build().await);
+
         let builder =
             DatanodeBuilder::new(dn_opts, fe_plugins.clone()).with_kv_backend(kv_backend.clone());
         let datanode = builder.build().await.context(StartDatanodeSnafu)?;
 
-        let node_manager = Arc::new(StandaloneDatanodeManager(datanode.region_server()));
+        let node_manager = Arc::new(StandaloneDatanodeManager {
+            region_server: datanode.region_server(),
+            flow_server: flownode.clone(),
+        });
 
         let table_id_sequence = Arc::new(
             SequenceBuilder::new(TABLE_ID_SEQ, kv_backend.clone())
@@ -440,8 +456,6 @@ impl StartCommand {
             opts.wal.into(),
             kv_backend.clone(),
         ));
-        let table_metadata_manager =
-            Self::create_table_metadata_manager(kv_backend.clone()).await?;
         let flow_metadata_manager = Arc::new(FlowMetadataManager::new(kv_backend.clone()));
         let table_meta_allocator = Arc::new(TableMetadataAllocator::new(
             table_id_sequence,
@@ -474,6 +488,12 @@ impl StartCommand {
         .await
         .context(StartFrontendSnafu)?;
 
+        // flow server need to be able to use frontend to write insert requests back
+        flownode
+            .set_frontend_invoker(Box::new(frontend.clone()))
+            .await;
+        let _handle = flownode.clone().run_background();
+
         let servers = Services::new(fe_opts.clone(), Arc::new(frontend.clone()), fe_plugins)
             .build()
             .await

src/common/meta/src/key/table_info.rs

@@ -128,6 +128,7 @@ impl TableInfoValue {
 }
 
 pub type TableInfoManagerRef = Arc<TableInfoManager>;
+#[derive(Clone)]
 pub struct TableInfoManager {
     kv_backend: KvBackendRef,
 }