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
This commit is contained in:
Yingwen
2023-04-23 10:04:09 +08:00
committed by GitHub
parent f2cc912c87
commit 1ee9ad4ca1
15 changed files with 199 additions and 59 deletions
+4 -4
View File
@@ -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),
}
}
}
+14 -1
View File
@@ -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,
+20 -5
View File
@@ -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,
+13 -10
View File
@@ -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<ProcedureManagerRef>,
}
@@ -61,13 +57,11 @@ impl SqlHandler {
pub fn new(
table_engine_manager: TableEngineManagerRef,
catalog_manager: CatalogManagerRef,
engine_procedure: TableEngineProcedureRef,
procedure_manager: Option<ProcedureManagerRef>,
) -> 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<dyn Table>) -> Result<TableEngineRef> {
pub fn table_engine(&self, table: TableRef) -> Result<TableEngineRef> {
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<TableEngineProcedureRef> {
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()
+11 -5
View File
@@ -84,11 +84,17 @@ impl SqlHandler {
req: AlterTableRequest,
) -> Result<Output> {
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;
+11 -4
View File
@@ -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;
+11 -5
View File
@@ -89,11 +89,17 @@ impl SqlHandler {
req: DropTableRequest,
) -> Result<Output> {
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;
+11 -3
View File
@@ -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)
}
+13 -1
View File
@@ -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<DatanodeInstance>, 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,
},
)
}
+3 -4
View File
@@ -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()?
})?
}
}
}
+3 -4
View File
@@ -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()?
})?
}
}
}
+3 -4
View File
@@ -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()?
})?
}
}
}
+4 -3
View File
@@ -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<common_procedure::Error>,
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 { .. } => {
+50 -5
View File
@@ -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<TableEngineRef>;
/// 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<TableEngineProcedureRef>;
}
pub type TableEngineManagerRef = Arc<dyn TableEngineManager>;
/// Simple in-memory table engine manager
pub struct MemoryTableEngineManager {
pub engines: RwLock<HashMap<String, TableEngineRef>>,
engine_procedures: RwLock<HashMap<String, TableEngineProcedureRef>>,
}
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<String, TableEngineProcedureRef>,
) -> 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<TableEngineProcedureRef> {
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 { .. }
);
}
}
+28 -1
View File
@@ -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<BoxedProcedure> {
unimplemented!()
}
fn alter_table_procedure(
&self,
_ctx: &EngineContext,
_request: AlterTableRequest,
) -> Result<BoxedProcedure> {
unimplemented!()
}
fn drop_table_procedure(
&self,
_ctx: &EngineContext,
_request: DropTableRequest,
) -> Result<BoxedProcedure> {
unimplemented!()
}
}