feat: implement the CreateFlowProcedure

src/cmd/src/standalone.rs

@@ -18,12 +18,14 @@ use std::{fs, path};
 use async_trait::async_trait;
 use catalog::kvbackend::KvBackendCatalogManager;
 use clap::Parser;
-use common_catalog::consts::MIN_USER_TABLE_ID;
+use common_catalog::consts::{MIN_USER_FLOW_TASK_ID, MIN_USER_TABLE_ID};
 use common_config::{metadata_store_dir, KvBackendConfig};
 use common_meta::cache_invalidator::{CacheInvalidatorRef, MultiCacheInvalidator};
 use common_meta::ddl::table_meta::{TableMetadataAllocator, TableMetadataAllocatorRef};
-use common_meta::ddl::ProcedureExecutorRef;
+use common_meta::ddl::task_meta::{FlowTaskMetadataAllocator, FlowTaskMetadataAllocatorRef};
+use common_meta::ddl::{DdlContext, ProcedureExecutorRef};
 use common_meta::ddl_manager::DdlManager;
+use common_meta::key::flow_task::{FlowTaskMetadataManager, FlowTaskMetadataManagerRef};
 use common_meta::key::{TableMetadataManager, TableMetadataManagerRef};
 use common_meta::kv_backend::KvBackendRef;
 use common_meta::node_manager::NodeManagerRef;
@@ -45,6 +47,7 @@ use frontend::server::Services;
 use frontend::service_config::{
  GrpcOptions, InfluxdbOptions, MysqlOptions, OpentsdbOptions, PostgresOptions, PromStoreOptions,
 };
+use meta_srv::metasrv::{FLOW_TASK_ID_SEQ, TABLE_ID_SEQ};
 use mito2::config::MitoConfig;
 use serde::{Deserialize, Serialize};
 use servers::export_metrics::ExportMetricsOption;
@@ -411,30 +414,40 @@ impl StartCommand {
  let node_manager = Arc::new(StandaloneDatanodeManager(datanode.region_server()));
 
  let table_id_sequence = Arc::new(
- SequenceBuilder::new("table_id", kv_backend.clone())
+ SequenceBuilder::new(TABLE_ID_SEQ, kv_backend.clone())
  .initial(MIN_USER_TABLE_ID as u64)
  .step(10)
  .build(),
  );
+ let flow_task_id_sequence = Arc::new(
+ SequenceBuilder::new(FLOW_TASK_ID_SEQ, kv_backend.clone())
+ .initial(MIN_USER_FLOW_TASK_ID as u64)
+ .step(10)
+ .build(),
+ );
  let wal_options_allocator = Arc::new(WalOptionsAllocator::new(
  opts.wal_meta.clone(),
  kv_backend.clone(),
  ));
-
  let table_metadata_manager =
  Self::create_table_metadata_manager(kv_backend.clone()).await?;
-
+ let flow_task_metadata_manager = Arc::new(FlowTaskMetadataManager::new(kv_backend.clone()));
  let table_meta_allocator = Arc::new(TableMetadataAllocator::new(
  table_id_sequence,
  wal_options_allocator.clone(),
  ));
+ let flow_task_meta_allocator = Arc::new(
+ FlowTaskMetadataAllocator::with_noop_peer_allocator(flow_task_id_sequence),
+ );
 
  let ddl_task_executor = Self::create_ddl_task_executor(
- table_metadata_manager,
  procedure_manager.clone(),
  node_manager.clone(),
  multi_cache_invalidator,
+ table_metadata_manager,
  table_meta_allocator,
+ flow_task_metadata_manager,
+ flow_task_meta_allocator,
  )
  .await?;
 
@@ -462,20 +475,26 @@ impl StartCommand {
  }
 
  pub async fn create_ddl_task_executor(
- table_metadata_manager: TableMetadataManagerRef,
  procedure_manager: ProcedureManagerRef,
  node_manager: NodeManagerRef,
  cache_invalidator: CacheInvalidatorRef,
- table_meta_allocator: TableMetadataAllocatorRef,
+ table_metadata_manager: TableMetadataManagerRef,
+ table_metadata_allocator: TableMetadataAllocatorRef,
+ flow_task_metadata_manager: FlowTaskMetadataManagerRef,
+ flow_task_metadata_allocator: FlowTaskMetadataAllocatorRef,
  ) -> Result<ProcedureExecutorRef> {
  let procedure_executor: ProcedureExecutorRef = Arc::new(
  DdlManager::try_new(
+ DdlContext {
+ node_manager,
+ cache_invalidator,
+ memory_region_keeper: Arc::new(MemoryRegionKeeper::default()),
+ table_metadata_manager,
+ table_metadata_allocator,
+ flow_task_metadata_manager,
+ flow_task_metadata_allocator,
+ },
  procedure_manager,
- node_manager,
- cache_invalidator,
- table_metadata_manager,
- table_meta_allocator,
- Arc::new(MemoryRegionKeeper::default()),
  true,
  )
  .context(InitDdlManagerSnafu)?,

src/common/catalog/src/consts.rs

@@ -19,6 +19,9 @@ pub const DEFAULT_CATALOG_NAME: &str = "greptime";
 pub const DEFAULT_SCHEMA_NAME: &str = "public";
 pub const DEFAULT_PRIVATE_SCHEMA_NAME: &str = "greptime_private";
 
+/// Reserves [0,MIN_USER_FLOW_TASK_ID) for internal usage.
+/// User defined table id starts from this value.
+pub const MIN_USER_FLOW_TASK_ID: u32 = 1024;
 /// Reserves [0,MIN_USER_TABLE_ID) for internal usage.
 /// User defined table id starts from this value.
 pub const MIN_USER_TABLE_ID: u32 = 1024;

src/common/meta/src/ddl.rs

@@ -18,9 +18,11 @@ use std::sync::Arc;
 use common_telemetry::tracing_context::W3cTrace;
 use store_api::storage::{RegionNumber, TableId};
 
-use self::table_meta::TableMetadataAllocatorRef;
 use crate::cache_invalidator::CacheInvalidatorRef;
+use crate::ddl::table_meta::TableMetadataAllocatorRef;
+use crate::ddl::task_meta::FlowTaskMetadataAllocatorRef;
 use crate::error::Result;
+use crate::key::flow_task::FlowTaskMetadataManagerRef;
 use crate::key::table_route::PhysicalTableRouteValue;
 use crate::key::TableMetadataManagerRef;
 use crate::node_manager::NodeManagerRef;
@@ -31,13 +33,15 @@ use crate::rpc::procedure::{MigrateRegionRequest, MigrateRegionResponse, Procedu
 pub mod alter_logical_tables;
 pub mod alter_table;
 pub mod create_database;
+pub mod create_flow;
 pub mod create_logical_tables;
 pub mod create_table;
 mod create_table_template;
 pub mod drop_database;
 pub mod drop_table;
 mod physical_table_metadata;
 pub mod table_meta;
+pub mod task_meta;
 #[cfg(any(test, feature = "testing"))]
 pub mod test_util;
 #[cfg(test)]
@@ -93,11 +97,21 @@ pub struct TableMetadata {
  pub region_wal_options: HashMap<RegionNumber, String>,
 }
 
+/// The context of ddl.
 #[derive(Clone)]
 pub struct DdlContext {
+ /// Sends querying and requests to nodes.
  pub node_manager: NodeManagerRef,
+ /// Cache invalidation.
  pub cache_invalidator: CacheInvalidatorRef,
- pub table_metadata_manager: TableMetadataManagerRef,
+ /// Keep tracking operating regions.
  pub memory_region_keeper: MemoryRegionKeeperRef,
+ /// Table metadata manager.
+ pub table_metadata_manager: TableMetadataManagerRef,
+ /// Allocator for table metadata.
  pub table_metadata_allocator: TableMetadataAllocatorRef,
+ /// Flow task metadata manager.
+ pub flow_task_metadata_manager: FlowTaskMetadataManagerRef,
+ /// Allocator for flow task metadata.
+ pub flow_task_metadata_allocator: FlowTaskMetadataAllocatorRef,
 }