feat: Procedure to create table and register table to catalog (#1040)

* feat: Add table-procedures crate

* feat: Implement procedure to create table

* feat: Integrate procedure manager to datanode

* test: Test CreateTableProcedure

* refactor: Rename table-procedures to table-procedure

* feat: Implement create_table_by_procedure

* chore: Remove comment

* chore: Add todo

* feat: Add procedure config to standalone mode

* feat: Register table-procedure loaders

* feat: Address review comments

CreateTableProcedure just return error if the subprocedure is failed

* chore: Address CR comments

Author: evenyag

Cargo.lock

@@ -37,6 +37,7 @@ members = [
     "src/storage",
     "src/store-api",
     "src/table",
+    "src/table-procedure",
     "tests-integration",
     "tests/runner",
 ]

Cargo.toml

@@ -29,3 +29,7 @@ tcp_nodelay = false
 max_inflight_tasks = 4
 max_files_in_level0 = 16
 max_purge_tasks = 32
+
+[procedure.store]
+type = 'File'
+data_dir = '/tmp/greptimedb/procedure/'

config/datanode.example.toml

@@ -14,7 +14,6 @@ purge_threshold = '50GB'
 read_batch_size = 128
 sync_write = false
 
-
 [storage]
 type = 'File'
 data_dir = '/tmp/greptimedb/data/'
@@ -42,3 +41,7 @@ enable = true
 addr = '127.0.0.1:4003'
 runtime_size = 2
 check_pwd = false
+
+[procedure.store]
+type = 'File'
+data_dir = '/tmp/greptimedb/procedure/'

config/standalone.example.toml

@@ -14,7 +14,9 @@
 
 use clap::Parser;
 use common_telemetry::logging;
-use datanode::datanode::{Datanode, DatanodeOptions, FileConfig, ObjectStoreConfig};
+use datanode::datanode::{
+    Datanode, DatanodeOptions, FileConfig, ObjectStoreConfig, ProcedureConfig,
+};
 use meta_client::MetaClientOptions;
 use servers::Mode;
 use snafu::ResultExt;
@@ -65,6 +67,8 @@ struct StartCommand {
     data_dir: Option<String>,
     #[clap(long)]
     wal_dir: Option<String>,
+    #[clap(long)]
+    procedure_dir: Option<String>,
 }
 
 impl StartCommand {
@@ -134,6 +138,11 @@ impl TryFrom<StartCommand> for DatanodeOptions {
         if let Some(wal_dir) = cmd.wal_dir {
             opts.wal.dir = wal_dir;
         }
+
+        if let Some(procedure_dir) = cmd.procedure_dir {
+            opts.procedure = Some(ProcedureConfig::from_file_path(procedure_dir));
+        }
+
         Ok(opts)
     }
 }

src/cmd/src/datanode.rs

@@ -18,7 +18,7 @@ use clap::Parser;
 use common_base::Plugins;
 use common_telemetry::info;
 use datanode::datanode::{
-    CompactionConfig, Datanode, DatanodeOptions, ObjectStoreConfig, WalConfig,
+    CompactionConfig, Datanode, DatanodeOptions, ObjectStoreConfig, ProcedureConfig, WalConfig,
 };
 use datanode::instance::InstanceRef;
 use frontend::frontend::{Frontend, FrontendOptions};
@@ -81,6 +81,7 @@ pub struct StandaloneOptions {
     pub wal: WalConfig,
     pub storage: ObjectStoreConfig,
     pub compaction: CompactionConfig,
+    pub procedure: Option<ProcedureConfig>,
 }
 
 impl Default for StandaloneOptions {
@@ -99,6 +100,7 @@ impl Default for StandaloneOptions {
             wal: WalConfig::default(),
             storage: ObjectStoreConfig::default(),
             compaction: CompactionConfig::default(),
+            procedure: None,
         }
     }
 }
@@ -125,6 +127,7 @@ impl StandaloneOptions {
             wal: self.wal,
             storage: self.storage,
             compaction: self.compaction,
+            procedure: self.procedure,
             ..Default::default()
         }
     }

src/cmd/src/standalone.rs

@@ -123,8 +123,6 @@ pub(crate) struct ManagerContext {
     loaders: Mutex<HashMap<String, BoxedProcedureLoader>>,
     lock_map: LockMap,
     procedures: RwLock<HashMap<ProcedureId, ProcedureMetaRef>>,
-    // TODO(yingwen): Now we never clean the messages. But when the root procedure is done, we
-    // should be able to remove the its message and all its child messages.
     /// Messages loaded from the procedure store.
     messages: Mutex<HashMap<ProcedureId, ProcedureMessage>>,
 }

src/common/procedure/src/local.rs

@@ -21,6 +21,7 @@ common-catalog = { path = "../common/catalog" }
 common-error = { path = "../common/error" }
 common-grpc = { path = "../common/grpc" }
 common-grpc-expr = { path = "../common/grpc-expr" }
+common-procedure = { path = "../common/procedure" }
 common-query = { path = "../common/query" }
 common-recordbatch = { path = "../common/recordbatch" }
 common-runtime = { path = "../common/runtime" }
@@ -52,6 +53,7 @@ storage = { path = "../storage" }
 store-api = { path = "../store-api" }
 substrait = { path = "../common/substrait" }
 table = { path = "../table" }
+table-procedure = { path = "../table-procedure" }
 tokio.workspace = true
 tokio-stream = { version = "0.1", features = ["net"] }
 tonic.workspace = true

src/datanode/Cargo.toml

@@ -144,6 +144,27 @@ impl From<&DatanodeOptions> for StorageEngineConfig {
     }
 }
 
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(default)]
+pub struct ProcedureConfig {
+    /// Storage config for procedure manager.
+    pub store: ObjectStoreConfig,
+}
+
+impl Default for ProcedureConfig {
+    fn default() -> ProcedureConfig {
+        ProcedureConfig::from_file_path("/tmp/greptimedb/procedure/".to_string())
+    }
+}
+
+impl ProcedureConfig {
+    pub fn from_file_path(path: String) -> ProcedureConfig {
+        ProcedureConfig {
+            store: ObjectStoreConfig::File(FileConfig { data_dir: path }),
+        }
+    }
+}
+
 #[derive(Clone, Debug, Serialize, Deserialize)]
 #[serde(default)]
 pub struct DatanodeOptions {
@@ -159,6 +180,7 @@ pub struct DatanodeOptions {
     pub wal: WalConfig,
     pub storage: ObjectStoreConfig,
     pub compaction: CompactionConfig,
+    pub procedure: Option<ProcedureConfig>,
 }
 
 impl Default for DatanodeOptions {
@@ -176,6 +198,7 @@ impl Default for DatanodeOptions {
             wal: WalConfig::default(),
             storage: ObjectStoreConfig::default(),
             compaction: CompactionConfig::default(),
+            procedure: None,
         }
     }
 }

src/datanode/src/datanode.rs

@@ -15,6 +15,7 @@
 use std::any::Any;
 
 use common_error::prelude::*;
+use common_procedure::ProcedureId;
 use common_recordbatch::error::Error as RecordBatchError;
 use datafusion::parquet;
 use datatypes::prelude::ConcreteDataType;
@@ -394,6 +395,31 @@ pub enum Error {
         #[snafu(backtrace)]
         source: table::error::Error,
     },
+
+    #[snafu(display("Failed to recover procedure, source: {}", source))]
+    RecoverProcedure {
+        #[snafu(backtrace)]
+        source: common_procedure::error::Error,
+    },
+
+    #[snafu(display("Failed to submit procedure, source: {}", source))]
+    SubmitProcedure {
+        #[snafu(backtrace)]
+        source: common_procedure::error::Error,
+    },
+
+    #[snafu(display("Failed to wait procedure done, source: {}", source))]
+    WaitProcedure {
+        source: tokio::sync::watch::error::RecvError,
+        backtrace: Backtrace,
+    },
+
+    // TODO(yingwen): Use procedure's error.
+    #[snafu(display("Failed to execute procedure, procedure_id: {}", procedure_id))]
+    ProcedureExec {
+        procedure_id: ProcedureId,
+        backtrace: Backtrace,
+    },
 }
 
 pub type Result<T> = std::result::Result<T, Error>;
@@ -470,6 +496,10 @@ impl ErrorExt for Error {
             CopyTable { source, .. } => source.status_code(),
             TableScanExec { source, .. } => source.status_code(),
             UnrecognizedTableOption { .. } => StatusCode::InvalidArguments,
+            RecoverProcedure { source, .. } | SubmitProcedure { source, .. } => {
+                source.status_code()
+            }
+            WaitProcedure { .. } | ProcedureExec { .. } => StatusCode::Internal,
         }
     }