From 1ee9ad4ca1cd4664348a15fbe11fb1b0eb016564 Mon Sep 17 00:00:00 2001 From: Yingwen Date: Sun, 23 Apr 2023 10:04:09 +0800 Subject: [PATCH] feat: manage multiple engine procedure in the engine manager (#1434) * feat(table): Add engine procedure to engine manager * feat(datanode): Get engine procedure from engine manager * feat(table-procedure): Add source error to SubprocedureFailed * test: Enable procedure in tests and pass all tests * style(table-procedure): Fix clippy --- src/datanode/src/datanode.rs | 8 ++-- src/datanode/src/error.rs | 15 ++++++- src/datanode/src/instance.rs | 25 +++++++++--- src/datanode/src/sql.rs | 23 ++++++----- src/datanode/src/sql/alter.rs | 16 +++++--- src/datanode/src/sql/create.rs | 15 +++++-- src/datanode/src/sql/drop_table.rs | 16 +++++--- src/datanode/src/tests/test_util.rs | 14 +++++-- src/frontend/src/tests.rs | 14 ++++++- src/table-procedure/src/alter.rs | 7 ++-- src/table-procedure/src/create.rs | 7 ++-- src/table-procedure/src/drop.rs | 7 ++-- src/table-procedure/src/error.rs | 7 ++-- src/table/src/engine/manager.rs | 55 +++++++++++++++++++++++--- src/table/src/test_util/mock_engine.rs | 29 +++++++++++++- 15 files changed, 199 insertions(+), 59 deletions(-) diff --git a/src/datanode/src/datanode.rs b/src/datanode/src/datanode.rs index 4b0eb3eff30..fbf47f7d836 100644 --- a/src/datanode/src/datanode.rs +++ b/src/datanode/src/datanode.rs @@ -192,23 +192,23 @@ impl From<&DatanodeOptions> for StorageEngineConfig { #[derive(Debug, Clone, Serialize, Deserialize)] #[serde(default)] pub struct ProcedureConfig { - /// Storage config for procedure manager. - pub store: ObjectStoreConfig, /// Max retry times of procedure. pub max_retry_times: usize, /// Initial retry delay of procedures, increases exponentially. #[serde(with = "humantime_serde")] pub retry_delay: Duration, + /// Storage config for procedure manager. + pub store: ObjectStoreConfig, } impl Default for ProcedureConfig { fn default() -> ProcedureConfig { ProcedureConfig { + max_retry_times: 3, + retry_delay: Duration::from_millis(500), store: ObjectStoreConfig::File(FileConfig { data_dir: "/tmp/greptimedb/procedure/".to_string(), }), - max_retry_times: 3, - retry_delay: Duration::from_millis(500), } } } diff --git a/src/datanode/src/error.rs b/src/datanode/src/error.rs index 86e853857e9..6f25002a7d0 100644 --- a/src/datanode/src/error.rs +++ b/src/datanode/src/error.rs @@ -106,6 +106,17 @@ pub enum Error { source: table::error::Error, }, + #[snafu(display( + "Table engine procedure not found: {}, source: {}", + engine_name, + source + ))] + EngineProcedureNotFound { + engine_name: String, + #[snafu(backtrace)] + source: table::error::Error, + }, + #[snafu(display("Table not found: {}", table_name))] TableNotFound { table_name: String, @@ -432,7 +443,9 @@ impl ErrorExt for Error { Insert { source, .. } => source.status_code(), Delete { source, .. } => source.status_code(), - TableEngineNotFound { source, .. } => source.status_code(), + TableEngineNotFound { source, .. } | EngineProcedureNotFound { source, .. } => { + source.status_code() + } TableNotFound { .. } => StatusCode::TableNotFound, ColumnNotFound { .. } => StatusCode::TableColumnNotFound, diff --git a/src/datanode/src/instance.rs b/src/datanode/src/instance.rs index 1f76350b635..34af29472de 100644 --- a/src/datanode/src/instance.rs +++ b/src/datanode/src/instance.rs @@ -12,6 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. +use std::collections::HashMap; use std::sync::Arc; use std::time::Duration; use std::{fs, path}; @@ -46,6 +47,7 @@ use storage::scheduler::{LocalScheduler, SchedulerConfig}; use storage::EngineImpl; use store_api::logstore::LogStore; use table::engine::manager::MemoryTableEngineManager; +use table::engine::{TableEngine, TableEngineProcedureRef}; use table::requests::FlushTableRequest; use table::table::numbers::NumbersTable; use table::table::TableIdProviderRef; @@ -118,7 +120,17 @@ impl Instance { object_store, )); - let engine_manager = Arc::new(MemoryTableEngineManager::new(table_engine.clone())); + let mut engine_procedures = HashMap::with_capacity(2); + engine_procedures.insert( + table_engine.name().to_string(), + table_engine.clone() as TableEngineProcedureRef, + ); + // TODO(yingwen): Insert the file table engine into `engine_procedures` + // once #1372 is ready. + let engine_manager = Arc::new( + MemoryTableEngineManager::new(table_engine.clone()) + .with_engine_procedures(engine_procedures), + ); // create remote catalog manager let (catalog_manager, table_id_provider) = match opts.mode { @@ -144,7 +156,7 @@ impl Instance { ) } else { let catalog = Arc::new( - catalog::local::LocalCatalogManager::try_new(engine_manager) + catalog::local::LocalCatalogManager::try_new(engine_manager.clone()) .await .context(CatalogSnafu)?, ); @@ -158,7 +170,7 @@ impl Instance { Mode::Distributed => { let catalog = Arc::new(catalog::remote::RemoteCatalogManager::new( - engine_manager, + engine_manager.clone(), opts.node_id.context(MissingNodeIdSnafu)?, Arc::new(MetaKvBackend { client: meta_client.as_ref().unwrap().clone(), @@ -185,21 +197,24 @@ impl Instance { let procedure_manager = create_procedure_manager(&opts.procedure).await?; // Register all procedures. if let Some(procedure_manager) = &procedure_manager { + // Register procedures of the mito engine. table_engine.register_procedure_loaders(&**procedure_manager); + // Register procedures in table-procedure crate. table_procedure::register_procedure_loaders( catalog_manager.clone(), table_engine.clone(), table_engine.clone(), &**procedure_manager, ); + // TODO(yingwen): Register procedures of the file table engine once #1372 + // is ready. } Ok(Self { query_engine: query_engine.clone(), sql_handler: SqlHandler::new( - Arc::new(MemoryTableEngineManager::new(table_engine.clone())), + engine_manager, catalog_manager.clone(), - table_engine, procedure_manager.clone(), ), catalog_manager, diff --git a/src/datanode/src/sql.rs b/src/datanode/src/sql.rs index 47321507583..67aa46b4ba6 100644 --- a/src/datanode/src/sql.rs +++ b/src/datanode/src/sql.rs @@ -12,8 +12,6 @@ // See the License for the specific language governing permissions and // limitations under the License. -use std::sync::Arc; - use catalog::CatalogManagerRef; use common_error::prelude::BoxedError; use common_procedure::ProcedureManagerRef; @@ -24,10 +22,11 @@ use snafu::{OptionExt, ResultExt}; use table::engine::manager::TableEngineManagerRef; use table::engine::{TableEngineProcedureRef, TableEngineRef, TableReference}; use table::requests::*; -use table::{Table, TableRef}; +use table::TableRef; use crate::error::{ - self, CloseTableEngineSnafu, Result, TableEngineNotFoundSnafu, TableNotFoundSnafu, + self, CloseTableEngineSnafu, EngineProcedureNotFoundSnafu, Result, TableEngineNotFoundSnafu, + TableNotFoundSnafu, }; use crate::instance::sql::table_idents_to_full_name; @@ -51,9 +50,6 @@ pub enum SqlRequest { pub struct SqlHandler { table_engine_manager: TableEngineManagerRef, catalog_manager: CatalogManagerRef, - // TODO(yingwen): Support multiple table engine. We need to add a method - // to TableEngineManagerRef to return engine procedure by engine name. - engine_procedure: TableEngineProcedureRef, procedure_manager: Option, } @@ -61,13 +57,11 @@ impl SqlHandler { pub fn new( table_engine_manager: TableEngineManagerRef, catalog_manager: CatalogManagerRef, - engine_procedure: TableEngineProcedureRef, procedure_manager: Option, ) -> Self { Self { table_engine_manager, catalog_manager, - engine_procedure, procedure_manager, } } @@ -111,7 +105,7 @@ impl SqlHandler { self.table_engine_manager.clone() } - pub fn table_engine(&self, table: Arc) -> Result { + pub fn table_engine(&self, table: TableRef) -> Result { let engine_name = &table.table_info().meta.engine; let engine = self .table_engine_manager @@ -120,6 +114,15 @@ impl SqlHandler { Ok(engine) } + pub fn engine_procedure(&self, table: TableRef) -> Result { + let engine_name = &table.table_info().meta.engine; + let engine = self + .table_engine_manager + .engine_procedure(engine_name) + .context(EngineProcedureNotFoundSnafu { engine_name })?; + Ok(engine) + } + pub async fn close(&self) -> Result<()> { self.table_engine_manager .close() diff --git a/src/datanode/src/sql/alter.rs b/src/datanode/src/sql/alter.rs index 6ec8669d464..bb93b4fa7c9 100644 --- a/src/datanode/src/sql/alter.rs +++ b/src/datanode/src/sql/alter.rs @@ -84,11 +84,17 @@ impl SqlHandler { req: AlterTableRequest, ) -> Result { let table_name = req.table_name.clone(); - let procedure = AlterTableProcedure::new( - req, - self.catalog_manager.clone(), - self.engine_procedure.clone(), - ); + let table_ref = TableReference { + catalog: &req.catalog_name, + schema: &req.schema_name, + table: &table_name, + }; + + let table = self.get_table(&table_ref).await?; + let engine_procedure = self.engine_procedure(table)?; + + let procedure = + AlterTableProcedure::new(req, self.catalog_manager.clone(), engine_procedure); let procedure_with_id = ProcedureWithId::with_random_id(Box::new(procedure)); let procedure_id = procedure_with_id.id; diff --git a/src/datanode/src/sql/create.rs b/src/datanode/src/sql/create.rs index db4a76481a2..8196fcce920 100644 --- a/src/datanode/src/sql/create.rs +++ b/src/datanode/src/sql/create.rs @@ -32,9 +32,10 @@ use table_procedure::CreateTableProcedure; use crate::error::{ self, CatalogNotFoundSnafu, CatalogSnafu, ConstraintNotSupportedSnafu, CreateTableSnafu, - IllegalPrimaryKeysDefSnafu, InsertSystemCatalogSnafu, KeyColumnNotFoundSnafu, - RegisterSchemaSnafu, Result, SchemaExistsSnafu, SchemaNotFoundSnafu, SubmitProcedureSnafu, - TableEngineNotFoundSnafu, UnrecognizedTableOptionSnafu, WaitProcedureSnafu, + EngineProcedureNotFoundSnafu, IllegalPrimaryKeysDefSnafu, InsertSystemCatalogSnafu, + KeyColumnNotFoundSnafu, RegisterSchemaSnafu, Result, SchemaExistsSnafu, SchemaNotFoundSnafu, + SubmitProcedureSnafu, TableEngineNotFoundSnafu, UnrecognizedTableOptionSnafu, + WaitProcedureSnafu, }; use crate::sql::SqlHandler; @@ -150,11 +151,17 @@ impl SqlHandler { .context(TableEngineNotFoundSnafu { engine_name: &req.engine, })?; + let engine_procedure = self + .table_engine_manager + .engine_procedure(&req.engine) + .context(EngineProcedureNotFoundSnafu { + engine_name: &req.engine, + })?; let procedure = CreateTableProcedure::new( req, self.catalog_manager.clone(), table_engine.clone(), - self.engine_procedure.clone(), + engine_procedure, ); let procedure_with_id = ProcedureWithId::with_random_id(Box::new(procedure)); let procedure_id = procedure_with_id.id; diff --git a/src/datanode/src/sql/drop_table.rs b/src/datanode/src/sql/drop_table.rs index 0a24d4cb9a6..841af8fd739 100644 --- a/src/datanode/src/sql/drop_table.rs +++ b/src/datanode/src/sql/drop_table.rs @@ -89,11 +89,17 @@ impl SqlHandler { req: DropTableRequest, ) -> Result { let table_name = req.table_name.clone(); - let procedure = DropTableProcedure::new( - req, - self.catalog_manager.clone(), - self.engine_procedure.clone(), - ); + let table_ref = TableReference { + catalog: &req.catalog_name, + schema: &req.schema_name, + table: &table_name, + }; + + let table = self.get_table(&table_ref).await?; + let engine_procedure = self.engine_procedure(table)?; + + let procedure = + DropTableProcedure::new(req, self.catalog_manager.clone(), engine_procedure); let procedure_with_id = ProcedureWithId::with_random_id(Box::new(procedure)); let procedure_id = procedure_with_id.id; diff --git a/src/datanode/src/tests/test_util.rs b/src/datanode/src/tests/test_util.rs index 8477edcb275..83ca20ea2a6 100644 --- a/src/datanode/src/tests/test_util.rs +++ b/src/datanode/src/tests/test_util.rs @@ -12,6 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. +use std::collections::HashMap; use std::sync::Arc; use std::time::Duration; @@ -26,7 +27,7 @@ use mito::table::test_util::{new_test_object_store, MockEngine, MockMitoEngine}; use servers::Mode; use snafu::ResultExt; use table::engine::manager::MemoryTableEngineManager; -use table::engine::{EngineContext, TableEngineRef}; +use table::engine::{EngineContext, TableEngine, TableEngineProcedureRef, TableEngineRef}; use table::requests::{CreateTableRequest, TableOptions}; use crate::datanode::{ @@ -168,11 +169,18 @@ pub async fn create_mock_sql_handler() -> SqlHandler { MockEngine::default(), object_store, )); - let engine_manager = Arc::new(MemoryTableEngineManager::new(mock_engine.clone())); + let mut engine_procedures = HashMap::new(); + engine_procedures.insert( + mock_engine.name().to_string(), + mock_engine.clone() as TableEngineProcedureRef, + ); + let engine_manager = Arc::new( + MemoryTableEngineManager::new(mock_engine).with_engine_procedures(engine_procedures), + ); let catalog_manager = Arc::new( catalog::local::LocalCatalogManager::try_new(engine_manager.clone()) .await .unwrap(), ); - SqlHandler::new(engine_manager, catalog_manager, mock_engine, None) + SqlHandler::new(engine_manager, catalog_manager, None) } diff --git a/src/frontend/src/tests.rs b/src/frontend/src/tests.rs index e9e88648aaf..a770362a915 100644 --- a/src/frontend/src/tests.rs +++ b/src/frontend/src/tests.rs @@ -28,7 +28,7 @@ use common_grpc::channel_manager::ChannelManager; use common_runtime::Builder as RuntimeBuilder; use common_test_util::temp_dir::{create_temp_dir, TempDir}; use datanode::datanode::{ - DatanodeOptions, FileConfig, ObjectStoreConfig, StorageConfig, WalConfig, + DatanodeOptions, FileConfig, ObjectStoreConfig, ProcedureConfig, StorageConfig, WalConfig, }; use datanode::instance::Instance as DatanodeInstance; use meta_client::client::MetaClientBuilder; @@ -56,6 +56,7 @@ use crate::instance::Instance; pub struct TestGuard { _wal_tmp_dir: TempDir, _data_tmp_dir: TempDir, + _procedure_dir: TempDir, } pub(crate) struct MockDistributedInstance { @@ -113,6 +114,7 @@ pub(crate) async fn create_standalone_instance(test_name: &str) -> MockStandalon fn create_tmp_dir_and_datanode_opts(name: &str) -> (DatanodeOptions, TestGuard) { let wal_tmp_dir = create_temp_dir(&format!("gt_wal_{name}")); let data_tmp_dir = create_temp_dir(&format!("gt_data_{name}")); + let procedure_tmp_dir = create_temp_dir(&format!("gt_procedure_{name}")); let opts = DatanodeOptions { wal: WalConfig { dir: wal_tmp_dir.path().to_str().unwrap().to_string(), @@ -125,6 +127,9 @@ fn create_tmp_dir_and_datanode_opts(name: &str) -> (DatanodeOptions, TestGuard) ..Default::default() }, mode: Mode::Standalone, + procedure: Some(ProcedureConfig::from_file_path( + procedure_tmp_dir.path().to_str().unwrap().to_string(), + )), ..Default::default() }; ( @@ -132,6 +137,7 @@ fn create_tmp_dir_and_datanode_opts(name: &str) -> (DatanodeOptions, TestGuard) TestGuard { _wal_tmp_dir: wal_tmp_dir, _data_tmp_dir: data_tmp_dir, + _procedure_dir: procedure_tmp_dir, }, ) } @@ -202,6 +208,8 @@ async fn create_distributed_datanode( ) -> (Arc, TestGuard) { let wal_tmp_dir = create_temp_dir(&format!("gt_wal_{test_name}_dist_dn_{datanode_id}")); let data_tmp_dir = create_temp_dir(&format!("gt_data_{test_name}_dist_dn_{datanode_id}")); + let procedure_tmp_dir = + create_temp_dir(&format!("gt_procedure_{test_name}_dist_dn_{datanode_id}")); let opts = DatanodeOptions { node_id: Some(datanode_id), wal: WalConfig { @@ -215,6 +223,9 @@ async fn create_distributed_datanode( ..Default::default() }, mode: Mode::Distributed, + procedure: Some(ProcedureConfig::from_file_path( + procedure_tmp_dir.path().to_str().unwrap().to_string(), + )), ..Default::default() }; @@ -240,6 +251,7 @@ async fn create_distributed_datanode( TestGuard { _wal_tmp_dir: wal_tmp_dir, _data_tmp_dir: data_tmp_dir, + _procedure_dir: procedure_tmp_dir, }, ) } diff --git a/src/table-procedure/src/alter.rs b/src/table-procedure/src/alter.rs index f6b772a357b..2b33ed87fa1 100644 --- a/src/table-procedure/src/alter.rs +++ b/src/table-procedure/src/alter.rs @@ -223,12 +223,11 @@ impl AlterTableProcedure { Ok(Status::Done) } } - ProcedureState::Failed { .. } => { + ProcedureState::Failed { error } => { // Return error if the subprocedure is failed. - SubprocedureFailedSnafu { + Err(error).context(SubprocedureFailedSnafu { subprocedure_id: sub_id, - } - .fail()? + })? } } } diff --git a/src/table-procedure/src/create.rs b/src/table-procedure/src/create.rs index c81201987c1..6a984852974 100644 --- a/src/table-procedure/src/create.rs +++ b/src/table-procedure/src/create.rs @@ -211,12 +211,11 @@ impl CreateTableProcedure { self.data.state = CreateTableState::RegisterCatalog; Ok(Status::executing(true)) } - ProcedureState::Failed { .. } => { + ProcedureState::Failed { error } => { // Return error if the subprocedure is failed. - SubprocedureFailedSnafu { + Err(error).context(SubprocedureFailedSnafu { subprocedure_id: sub_id, - } - .fail()? + })? } } } diff --git a/src/table-procedure/src/drop.rs b/src/table-procedure/src/drop.rs index 252a136a427..5ea82d8c568 100644 --- a/src/table-procedure/src/drop.rs +++ b/src/table-procedure/src/drop.rs @@ -212,12 +212,11 @@ impl DropTableProcedure { Ok(Status::Done) } - ProcedureState::Failed { .. } => { + ProcedureState::Failed { error } => { // Return error if the subprocedure is failed. - SubprocedureFailedSnafu { + Err(error).context(SubprocedureFailedSnafu { subprocedure_id: sub_id, - } - .fail()? + })? } } } diff --git a/src/table-procedure/src/error.rs b/src/table-procedure/src/error.rs index 9ed582a6a67..5fb2e3b5177 100644 --- a/src/table-procedure/src/error.rs +++ b/src/table-procedure/src/error.rs @@ -13,6 +13,7 @@ // limitations under the License. use std::any::Any; +use std::sync::Arc; use common_error::prelude::*; use common_procedure::ProcedureId; @@ -57,6 +58,7 @@ pub enum Error { #[snafu(display("Subprocedure {} failed", subprocedure_id))] SubprocedureFailed { subprocedure_id: ProcedureId, + source: Arc, location: Location, }, @@ -71,9 +73,8 @@ impl ErrorExt for Error { use Error::*; match self { - SerializeProcedure { .. } | DeserializeProcedure { .. } | SubprocedureFailed { .. } => { - StatusCode::Internal - } + SerializeProcedure { .. } | DeserializeProcedure { .. } => StatusCode::Internal, + SubprocedureFailed { source, .. } => source.status_code(), InvalidRawSchema { source, .. } => source.status_code(), AccessCatalog { source } => source.status_code(), CatalogNotFound { .. } | SchemaNotFound { .. } | TableExists { .. } => { diff --git a/src/table/src/engine/manager.rs b/src/table/src/engine/manager.rs index 671186d38a4..f3d22271f57 100644 --- a/src/table/src/engine/manager.rs +++ b/src/table/src/engine/manager.rs @@ -19,38 +19,58 @@ use async_trait::async_trait; use common_telemetry::error; use snafu::{ensure, OptionExt}; -use crate::engine::TableEngineRef; +use crate::engine::{TableEngineProcedureRef, TableEngineRef}; use crate::error::{EngineExistSnafu, EngineNotFoundSnafu, Result}; #[async_trait::async_trait] pub trait TableEngineManager: Send + Sync { - /// returns `error::EngineNotFound` if engine not found + /// returns [Error::EngineNotFound](crate::error::Error::EngineNotFound) if engine not found fn engine(&self, name: &str) -> Result; - /// returns `error::EngineExist` if engine exists + /// returns [Error::EngineExist](crate::error::Error::EngineExist) if engine exists fn register_engine(&self, name: &str, engine: TableEngineRef) -> Result<()>; /// closes all registered engines async fn close(&self) -> Result<()>; + + /// returns [TableEngineProcedureRef] of specific engine `name` or + /// [Error::EngineNotFound](crate::error::Error::EngineNotFound) if engine not found + fn engine_procedure(&self, name: &str) -> Result; } pub type TableEngineManagerRef = Arc; /// Simple in-memory table engine manager pub struct MemoryTableEngineManager { pub engines: RwLock>, + engine_procedures: RwLock>, } impl MemoryTableEngineManager { + /// Create a new [MemoryTableEngineManager] with single table `engine`. pub fn new(engine: TableEngineRef) -> Self { MemoryTableEngineManager::alias(engine.name().to_string(), engine) } + /// Create a new [MemoryTableEngineManager] with single table `engine` and + /// an alias `name` instead of the engine's name. pub fn alias(name: String, engine: TableEngineRef) -> Self { let mut engines = HashMap::new(); engines.insert(name, engine); let engines = RwLock::new(engines); - MemoryTableEngineManager { engines } + MemoryTableEngineManager { + engines, + engine_procedures: RwLock::new(HashMap::new()), + } + } + + /// Attach the `engine_procedures` to the manager. + pub fn with_engine_procedures( + mut self, + engine_procedures: HashMap, + ) -> Self { + self.engine_procedures = RwLock::new(engine_procedures); + self } } @@ -91,6 +111,14 @@ impl TableEngineManager for MemoryTableEngineManager { Ok(()) } + + fn engine_procedure(&self, name: &str) -> Result { + let engine_procedures = self.engine_procedures.read().unwrap(); + engine_procedures + .get(name) + .cloned() + .context(EngineNotFoundSnafu { engine: name }) + } } #[cfg(test)] @@ -108,6 +136,12 @@ mod tests { let table_engine_ref = Arc::new(table_engine); let table_engine_manager = MemoryTableEngineManager::new(table_engine_ref.clone()); + // Attach engine procedures. + let mut engine_procedures = HashMap::new(); + let engine_procedure: TableEngineProcedureRef = table_engine_ref.clone(); + engine_procedures.insert(table_engine_ref.name().to_string(), engine_procedure); + let table_engine_manager = table_engine_manager.with_engine_procedures(engine_procedures); + table_engine_manager .register_engine("yet_another", table_engine_ref.clone()) .unwrap(); @@ -122,6 +156,17 @@ mod tests { let missing = table_engine_manager.engine("not_exists"); - assert_matches!(missing.err().unwrap(), error::Error::EngineNotFound { .. }) + assert_matches!(missing.err().unwrap(), error::Error::EngineNotFound { .. }); + + assert!(table_engine_manager + .engine_procedure(table_engine_ref.name()) + .is_ok()); + assert_matches!( + table_engine_manager + .engine_procedure("unknown") + .err() + .unwrap(), + error::Error::EngineNotFound { .. } + ); } } diff --git a/src/table/src/test_util/mock_engine.rs b/src/table/src/test_util/mock_engine.rs index ce30dddee2d..a59c04548ac 100644 --- a/src/table/src/test_util/mock_engine.rs +++ b/src/table/src/test_util/mock_engine.rs @@ -16,9 +16,10 @@ use std::collections::HashMap; use std::sync::Arc; use async_trait::async_trait; +use common_procedure::BoxedProcedure; use tokio::sync::Mutex; -use crate::engine::{EngineContext, TableEngine, TableReference}; +use crate::engine::{EngineContext, TableEngine, TableEngineProcedure, TableReference}; use crate::requests::{AlterTableRequest, CreateTableRequest, DropTableRequest, OpenTableRequest}; use crate::test_util::EmptyTable; use crate::{Result, TableRef}; @@ -101,3 +102,29 @@ impl TableEngine for MockTableEngine { Ok(()) } } + +impl TableEngineProcedure for MockTableEngine { + fn create_table_procedure( + &self, + _ctx: &EngineContext, + _request: CreateTableRequest, + ) -> Result { + unimplemented!() + } + + fn alter_table_procedure( + &self, + _ctx: &EngineContext, + _request: AlterTableRequest, + ) -> Result { + unimplemented!() + } + + fn drop_table_procedure( + &self, + _ctx: &EngineContext, + _request: DropTableRequest, + ) -> Result { + unimplemented!() + } +}