diff --git a/.github/workflows/develop.yml b/.github/workflows/develop.yml index d9f6b0c061..24a74a2d76 100644 --- a/.github/workflows/develop.yml +++ b/.github/workflows/develop.yml @@ -795,6 +795,12 @@ jobs: run: tar -xvf ./bins.tar.gz - name: Run sqlness run: RUST_BACKTRACE=1 ./bins/sqlness-runner bare ${{ matrix.mode.opts }} -c ./tests/cases --bins-dir ./bins --preserve-state + - if: matrix.mode.name == 'Basic' + name: Run soft-drop lifecycle sqlness + env: + GREPTIMEDB_METASRV__GC__SOFT_DROP__ENABLE: "true" + GREPTIMEDB_METASRV__GC__SOFT_DROP__RETENTION: "7d" + run: RUST_BACKTRACE=1 ./bins/sqlness-runner bare --enable-gc -c ./tests/cases-soft-drop --bins-dir ./bins --preserve-state -t soft_drop_table - name: Upload sqlness logs if: failure() uses: actions/upload-artifact@v4 diff --git a/config/config.md b/config/config.md index 93575fcce4..86374927dd 100644 --- a/config/config.md +++ b/config/config.md @@ -438,8 +438,8 @@ | `gc.enable` | Bool | `false` | Whether GC is enabled. Default to false. Need to be the same with datanode's `mito.gc.enable`
If set to false, no GC will be performed | | `gc.gc_cooldown_period` | String | `5m` | Cooldown period between GC operations on the same region. | | `gc.experimental_soft_drop` | -- | -- | -- | -| `gc.experimental_soft_drop.enable` | Bool | `false` | Reserved experimental option. Currently ignored. | -| `gc.experimental_soft_drop.retention` | String | `7d` | Reserved retention duration. Currently ignored. | +| `gc.experimental_soft_drop.enable` | Bool | `false` | Whether soft drop is enabled. Requires `gc.enable = true`. | +| `gc.experimental_soft_drop.retention` | String | `7d` | How long soft-dropped tables are retained before automatic purge. | | `logging` | -- | -- | The logging options. | | `logging.dir` | String | `./greptimedb_data/logs` | The directory to store the log files. If set to empty, logs will not be written to files. | | `logging.level` | String | Unset | The log level. Can be `info`/`debug`/`warn`/`error`. | diff --git a/config/metasrv.example.toml b/config/metasrv.example.toml index 55e6497078..33f3378cbc 100644 --- a/config/metasrv.example.toml +++ b/config/metasrv.example.toml @@ -327,9 +327,9 @@ enable = false gc_cooldown_period = "5m" [gc.experimental_soft_drop] -## Reserved experimental option. Currently ignored. +## Whether soft drop is enabled. Requires `gc.enable = true`. enable = false -## Reserved retention duration. Currently ignored. +## How long soft-dropped tables are retained before automatic purge. retention = "7d" ## The logging options. diff --git a/src/cmd/tests/load_config_test.rs b/src/cmd/tests/load_config_test.rs index 5814cc33b4..78c10eb14d 100644 --- a/src/cmd/tests/load_config_test.rs +++ b/src/cmd/tests/load_config_test.rs @@ -233,6 +233,25 @@ fn test_load_metasrv_example_config() { similar_asserts::assert_eq!(options, expected); } +#[test] +fn test_load_metasrv_soft_drop_config() { + let config = tempfile::NamedTempFile::new().unwrap(); + std::fs::write( + config.path(), + "[gc]\nenable = true\n[gc.experimental_soft_drop]\nenable = true\nretention = \"1d\"\n", + ) + .unwrap(); + + let options = + GreptimeOptions::::load_layered_options(config.path().to_str(), "") + .unwrap(); + assert!(options.component.gc.experimental_soft_drop.enable); + assert_eq!( + Duration::from_secs(24 * 60 * 60), + options.component.gc.experimental_soft_drop.retention + ); +} + #[test] fn test_load_flownode_example_config() { let example_config = common_test_util::find_workspace_path("config/flownode.example.toml"); diff --git a/src/common/function/src/admin.rs b/src/common/function/src/admin.rs index c31b608967..02fd32bd77 100644 --- a/src/common/function/src/admin.rs +++ b/src/common/function/src/admin.rs @@ -17,6 +17,7 @@ mod flush_compact_region; mod flush_compact_table; mod gc; mod migrate_region; +mod purge_table; mod reconcile_catalog; mod reconcile_database; mod reconcile_table; @@ -25,6 +26,7 @@ use flush_compact_region::{CompactRegionFunction, FlushRegionFunction}; use flush_compact_table::{CompactTableFunction, FlushTableFunction}; use gc::{GcRegionsFunction, GcTableFunction}; use migrate_region::MigrateRegionFunction; +use purge_table::PurgeTableFunction; use reconcile_catalog::ReconcileCatalogFunction; use reconcile_database::ReconcileDatabaseFunction; use reconcile_table::ReconcileTableFunction; @@ -52,4 +54,9 @@ impl AdminFunction { registry.register(ReconcileDatabaseFunction::factory()); registry.register(ReconcileTableFunction::factory()); } + + /// Register functions that must only be resolved by an ADMIN statement. + pub fn register_admin_only(registry: &FunctionRegistry) { + registry.register(PurgeTableFunction::factory()); + } } diff --git a/src/common/function/src/admin/gc.rs b/src/common/function/src/admin/gc.rs index b5e067ce72..e9ba33a050 100644 --- a/src/common/function/src/admin/gc.rs +++ b/src/common/function/src/admin/gc.rs @@ -266,6 +266,14 @@ mod tests { #[async_trait] impl ProcedureServiceHandler for MockProcedureServiceHandler { + async fn purge_table( + &self, + _table_name: table::table_name::TableName, + _query_ctx: QueryContextRef, + ) -> Result<()> { + unreachable!() + } + async fn migrate_region(&self, _request: MigrateRegionRequest) -> Result> { unreachable!() } diff --git a/src/common/function/src/admin/purge_table.rs b/src/common/function/src/admin/purge_table.rs new file mode 100644 index 0000000000..bc4b74da72 --- /dev/null +++ b/src/common/function/src/admin/purge_table.rs @@ -0,0 +1,364 @@ +// Copyright 2023 Greptime Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use arrow::datatypes::DataType as ArrowDataType; +use common_error::ext::BoxedError; +use common_macro::admin_fn; +use common_query::error::{ + InvalidFuncArgsSnafu, MissingProcedureServiceHandlerSnafu, ProcedureServiceSnafu, Result, + UnsupportedInputDataTypeSnafu, +}; +use datafusion_expr::{Signature, Volatility}; +use datatypes::prelude::{Value, ValueRef}; +use session::context::QueryContextRef; +use session::table_name::table_name_to_full_name; +use snafu::{ResultExt, ensure}; +use table::table_name::TableName; + +use crate::handlers::ProcedureServiceHandlerRef; + +/// Purges a dropped table after the ADMIN statement layer authorizes the request. +/// Name validation retains the session helper's existing catalog-only permission check. +#[admin_fn( + name = PurgeTableFunction, + display_name = purge_table, + sig_fn = purge_table_signature, + ret = uint64, + single_row +)] +pub(crate) async fn purge_table( + procedure_service_handler: &ProcedureServiceHandlerRef, + query_ctx: &QueryContextRef, + params: &[ValueRef<'_>], +) -> Result { + ensure!( + params.len() == 1, + InvalidFuncArgsSnafu { + err_msg: format!( + "The length of the args is not correct, expect 1, have: {}", + params.len() + ), + } + ); + let ValueRef::String(table_name) = params[0] else { + return UnsupportedInputDataTypeSnafu { + function: "purge_table", + datatypes: params + .iter() + .map(|value| value.data_type()) + .collect::>(), + } + .fail(); + }; + let (catalog_name, schema_name, table_name) = table_name_to_full_name(table_name, query_ctx) + .map_err(BoxedError::new) + .context(ProcedureServiceSnafu)?; + + procedure_service_handler + .purge_table( + TableName::new(catalog_name, schema_name, table_name), + query_ctx.clone(), + ) + .await?; + Ok(Value::from(0_u64)) +} + +fn purge_table_signature() -> Signature { + Signature::uniform(1, vec![ArrowDataType::Utf8], Volatility::Immutable) +} + +#[cfg(test)] +mod tests { + use std::sync::{Arc, Mutex}; + + use api::v1::meta::ReconcileRequest; + use arrow::array::StringArray; + use arrow::datatypes::{DataType as ArrowDataType, Field}; + use async_trait::async_trait; + use catalog::CatalogManagerRef; + use common_macro::admin_fn; + use common_meta::rpc::procedure::{ + GcRegionsRequest, GcResponse, GcTableRequest, ManageRegionFollowerRequest, + MigrateRegionRequest, ProcedureStateResponse, + }; + use common_query::error::{InvalidFuncArgsSnafu, MissingProcedureServiceHandlerSnafu, Result}; + use datafusion_expr::{ColumnarValue, Signature, TypeSignature, Volatility}; + use datatypes::prelude::{Value, ValueRef}; + use session::context::{QueryContext, QueryContextBuilder, QueryContextRef}; + use table::table_name::TableName; + + use super::{PurgeTableFunction, purge_table}; + use crate::function_factory::ScalarFunctionFactory; + use crate::function_registry::{FUNCTION_REGISTRY, get_admin_function}; + use crate::handlers::{ProcedureServiceHandler, ProcedureServiceHandlerRef}; + use crate::state::FunctionState; + + #[admin_fn( + name = ZeroArgSingleRowFunction, + display_name = zero_arg_single_row, + sig_fn = zero_arg_single_row_signature, + ret = uint64, + single_row + )] + async fn zero_arg_single_row( + _procedure_service_handler: &ProcedureServiceHandlerRef, + _query_ctx: &QueryContextRef, + params: &[ValueRef<'_>], + ) -> Result { + assert!(params.is_empty()); + Ok(Value::from(0_u64)) + } + + fn zero_arg_single_row_signature() -> Signature { + Signature::nullary(Volatility::Immutable) + } + + #[derive(Default)] + struct RecordingHandler { + calls: Mutex>, + fail: bool, + } + + #[async_trait] + impl ProcedureServiceHandler for RecordingHandler { + async fn purge_table( + &self, + table_name: TableName, + query_ctx: QueryContextRef, + ) -> Result<()> { + if self.fail { + return InvalidFuncArgsSnafu { + err_msg: "purge failed", + } + .fail(); + } + self.calls.lock().unwrap().push((table_name, query_ctx)); + Ok(()) + } + + async fn migrate_region(&self, _: MigrateRegionRequest) -> Result> { + unreachable!() + } + async fn reconcile(&self, _: ReconcileRequest) -> Result> { + unreachable!() + } + async fn query_procedure_state(&self, _: &str) -> Result { + unreachable!() + } + async fn manage_region_follower(&self, _: ManageRegionFollowerRequest) -> Result<()> { + unreachable!() + } + fn catalog_manager(&self) -> &CatalogManagerRef { + unreachable!() + } + async fn gc_regions(&self, _: GcRegionsRequest) -> Result { + unreachable!() + } + async fn gc_table(&self, _: GcTableRequest) -> Result { + unreachable!() + } + } + + #[test] + fn test_purge_table_is_admin_only() { + assert!(get_admin_function("purge_table").is_some()); + assert!(FUNCTION_REGISTRY.get_function("purge_table").is_none()); + } + + #[tokio::test] + async fn test_admin_only_factory_executes_without_registry_leak() { + let handler = Arc::new(RecordingHandler::default()); + let state = FunctionState { + procedure_service_handler: Some(handler.clone()), + ..Default::default() + }; + let function = + get_admin_function("purge_table") + .unwrap() + .provide(crate::function::FunctionContext { + query_ctx: QueryContext::arc(), + state: Arc::new(state), + }); + let args = datafusion_expr::ScalarFunctionArgs { + args: vec![ColumnarValue::Array(Arc::new(StringArray::from(vec![ + "foo", + ])))], + arg_fields: vec![Arc::new(Field::new("arg_0", ArrowDataType::Utf8, false))], + return_field: Arc::new(Field::new("result", ArrowDataType::UInt64, false)), + number_rows: 1, + config_options: Arc::new(datafusion_common::config::ConfigOptions::default()), + }; + + function + .as_async() + .unwrap() + .invoke_async_with_args(args) + .await + .unwrap(); + + assert_eq!(1, handler.calls.lock().unwrap().len()); + assert!(FUNCTION_REGISTRY.get_function("purge_table").is_none()); + } + + #[test] + fn test_purge_table_signature() { + let factory: ScalarFunctionFactory = PurgeTableFunction::factory().into(); + let function = factory.provide(crate::function::FunctionContext::mock()); + assert_eq!("purge_table", function.name()); + assert_eq!(ArrowDataType::UInt64, function.return_type(&[]).unwrap()); + assert!(matches!( + function.signature(), + Signature { type_signature: TypeSignature::Uniform(1, types), volatility: Volatility::Immutable, .. } + if types == &vec![ArrowDataType::Utf8] + )); + } + + #[tokio::test] + async fn test_purge_table_rejects_multi_row_before_handler_call() { + let handler = Arc::new(RecordingHandler::default()); + let state = FunctionState { + procedure_service_handler: Some(handler.clone()), + ..Default::default() + }; + let factory: ScalarFunctionFactory = PurgeTableFunction::factory().into(); + let function = factory.provide(crate::function::FunctionContext { + query_ctx: QueryContext::arc(), + state: Arc::new(state), + }); + let args = datafusion_expr::ScalarFunctionArgs { + args: vec![ColumnarValue::Array(Arc::new(StringArray::from(vec![ + Some("foo"), + None, + ])))], + arg_fields: vec![Arc::new(Field::new("arg_0", ArrowDataType::Utf8, true))], + return_field: Arc::new(Field::new("result", ArrowDataType::UInt64, true)), + number_rows: 2, + config_options: Arc::new(datafusion_common::config::ConfigOptions::default()), + }; + + assert!( + function + .as_async() + .unwrap() + .invoke_async_with_args(args) + .await + .is_err() + ); + assert!(handler.calls.lock().unwrap().is_empty()); + } + + #[tokio::test] + async fn test_single_row_admin_fn_rejects_zero_arg_multi_row_invocation() { + let handler = Arc::new(RecordingHandler::default()); + let state = FunctionState { + procedure_service_handler: Some(handler), + ..Default::default() + }; + let factory: ScalarFunctionFactory = ZeroArgSingleRowFunction::factory().into(); + let function = factory.provide(crate::function::FunctionContext { + query_ctx: QueryContext::arc(), + state: Arc::new(state), + }); + let args = datafusion_expr::ScalarFunctionArgs { + args: vec![], + arg_fields: vec![], + return_field: Arc::new(Field::new("result", ArrowDataType::UInt64, false)), + number_rows: 2, + config_options: Arc::new(datafusion_common::config::ConfigOptions::default()), + }; + + let err = function + .as_async() + .unwrap() + .invoke_async_with_args(args) + .await + .unwrap_err(); + assert!(err.to_string().contains("received 2")); + } + + #[tokio::test] + async fn test_purge_table_resolves_names_and_returns_zero() { + let handler = Arc::new(RecordingHandler::default()); + let handler_ref: ProcedureServiceHandlerRef = handler.clone(); + let query_ctx = QueryContextBuilder::default() + .current_catalog("catalog".to_string()) + .current_schema("schema".to_string()) + .build() + .into(); + + for (input, expected) in [ + ("foo", TableName::new("catalog", "schema", "foo")), + ("other.foo", TableName::new("catalog", "other", "foo")), + ( + "catalog.other.foo", + TableName::new("catalog", "other", "foo"), + ), + ] { + assert_eq!( + Value::from(0_u64), + purge_table(&handler_ref, &query_ctx, &[ValueRef::String(input)]) + .await + .unwrap() + ); + let calls = handler.calls.lock().unwrap(); + assert_eq!(expected, calls.last().unwrap().0); + assert!(Arc::ptr_eq(&query_ctx, &calls.last().unwrap().1)); + } + } + + #[tokio::test] + async fn test_purge_table_rejects_invalid_arguments() { + let recording_handler = Arc::new(RecordingHandler::default()); + let handler: ProcedureServiceHandlerRef = recording_handler.clone(); + for params in [vec![], vec![ValueRef::String("a"), ValueRef::String("b")]] { + assert!( + purge_table(&handler, &QueryContext::arc(), ¶ms) + .await + .is_err() + ); + } + assert!( + purge_table(&handler, &QueryContext::arc(), &[ValueRef::Int64(1)]) + .await + .is_err() + ); + assert!( + purge_table(&handler, &QueryContext::arc(), &[ValueRef::Null]) + .await + .is_err() + ); + assert!( + purge_table( + &handler, + &QueryContext::arc(), + &[ValueRef::String("a.b.c.d")] + ) + .await + .is_err() + ); + assert!(recording_handler.calls.lock().unwrap().is_empty()); + } + + #[tokio::test] + async fn test_purge_table_propagates_handler_error() { + let handler: ProcedureServiceHandlerRef = Arc::new(RecordingHandler { + fail: true, + ..Default::default() + }); + let error = purge_table(&handler, &QueryContext::arc(), &[ValueRef::String("foo")]) + .await + .unwrap_err(); + assert!(error.to_string().contains("purge failed")); + } +} diff --git a/src/common/function/src/function_registry.rs b/src/common/function/src/function_registry.rs index a3beca5403..7ad73396c8 100644 --- a/src/common/function/src/function_registry.rs +++ b/src/common/function/src/function_registry.rs @@ -221,6 +221,17 @@ pub static FUNCTION_REGISTRY: LazyLock> = LazyLock::new(|| Arc::new(function_registry) }); +static ADMIN_FUNCTION_REGISTRY: LazyLock = LazyLock::new(|| { + let registry = FunctionRegistry::default(); + AdminFunction::register_admin_only(®istry); + registry +}); + +/// Returns a function that is only available to the ADMIN statement executor. +pub fn get_admin_function(name: &str) -> Option { + ADMIN_FUNCTION_REGISTRY.get_function(name) +} + #[cfg(test)] mod tests { use super::*; diff --git a/src/common/function/src/handlers.rs b/src/common/function/src/handlers.rs index 50318fd14f..47de2824b8 100644 --- a/src/common/function/src/handlers.rs +++ b/src/common/function/src/handlers.rs @@ -30,6 +30,7 @@ use store_api::storage::RegionId; use table::requests::{ BuildIndexTableRequest, CompactTableRequest, DeleteRequest, FlushTableRequest, InsertRequest, }; +use table::table_name::TableName; /// A trait for handling table mutations in `QueryEngine`. #[async_trait] @@ -73,6 +74,9 @@ pub trait TableMutationHandler: Send + Sync { /// A trait for handling procedure service requests in `QueryEngine`. #[async_trait] pub trait ProcedureServiceHandler: Send + Sync { + /// Permanently purge a dropped table. + async fn purge_table(&self, table_name: TableName, query_ctx: QueryContextRef) -> Result<()>; + /// Migrate a region from source peer to target peer, returns the procedure id if success. async fn migrate_region(&self, request: MigrateRegionRequest) -> Result>; diff --git a/src/common/function/src/state.rs b/src/common/function/src/state.rs index 06dff44e79..726785f074 100644 --- a/src/common/function/src/state.rs +++ b/src/common/function/src/state.rs @@ -57,6 +57,14 @@ impl FunctionState { #[async_trait] impl ProcedureServiceHandler for MockProcedureServiceHandler { + async fn purge_table( + &self, + _table_name: table::table_name::TableName, + _query_ctx: QueryContextRef, + ) -> Result<()> { + Ok(()) + } + async fn migrate_region( &self, _request: MigrateRegionRequest, diff --git a/src/common/macro/src/admin_fn.rs b/src/common/macro/src/admin_fn.rs index efeb77f1d1..c580389f3c 100644 --- a/src/common/macro/src/admin_fn.rs +++ b/src/common/macro/src/admin_fn.rs @@ -45,6 +45,7 @@ pub(crate) fn process_admin_fn(args: TokenStream, input: TokenStream) -> TokenSt let mut sig_fn: Option = None; let mut ret: Option = None; let mut user_path: Option = None; + let mut single_row = false; let parser = syn::meta::parser(|meta| { if meta.path.is_ident("name") { @@ -62,6 +63,9 @@ pub(crate) fn process_admin_fn(args: TokenStream, input: TokenStream) -> TokenSt } else if meta.path.is_ident("user_path") { user_path = Some(meta.value()?.parse()?); Ok(()) + } else if meta.path.is_ident("single_row") { + single_row = true; + Ok(()) } else { Err(meta.error("unsupported property")) } @@ -113,6 +117,7 @@ pub(crate) fn process_admin_fn(args: TokenStream, input: TokenStream) -> TokenSt handler_type, display_name, user_path.expect("user_path required"), + single_row, ); result.extend(struct_code); } @@ -158,10 +163,20 @@ fn build_struct( handler_type: &Ident, display_name_ident: Ident, user_path: Path, + single_row: bool, ) -> TokenStream { let display_name = display_name_ident.to_string(); let ret = Ident::new(&format!("{ret}_datatype"), ret.span()); let uppcase_display_name = display_name.to_uppercase(); + let validate_rows = single_row.then(|| { + quote! { + if args.number_rows != 1 { + return Err(datafusion_common::DataFusionError::Execution( + format!("{} expects exactly one row, received {}", #display_name, args.number_rows) + )); + } + } + }); // Get the handler name in function state by the argument ident // TODO(discord9): consider simple depend injection if more handlers are needed let (handler, snafu_type) = match handler_type.to_string().as_str() { @@ -306,6 +321,7 @@ fn build_struct( } else { columns[0].len() }; + #validate_rows use snafu::{OptionExt, ResultExt}; use datatypes::data_type::DataType; diff --git a/src/common/meta/src/cache_invalidator.rs b/src/common/meta/src/cache_invalidator.rs index 4fe0699ba5..edf84b00b8 100644 --- a/src/common/meta/src/cache_invalidator.rs +++ b/src/common/meta/src/cache_invalidator.rs @@ -17,7 +17,6 @@ use std::sync::Arc; use crate::error::Result; use crate::flow_name::FlowName; use crate::instruction::{CacheIdent, DropFlow}; -use crate::key::MetadataKey; use crate::key::flow::flow_info::FlowInfoKey; use crate::key::flow::flow_name::FlowNameKey; use crate::key::flow::flow_route::FlowRouteKey; @@ -28,7 +27,11 @@ use crate::key::schema_name::SchemaNameKey; use crate::key::table_info::TableInfoKey; use crate::key::table_name::TableNameKey; use crate::key::table_route::TableRouteKey; +use crate::key::tombstone::to_tombstone_key; use crate::key::view_info::ViewInfoKey; +use crate::key::{ + MetadataKey, drop_generation_key, dropped_at_key, purging_key, retention_expires_at_key, +}; /// KvBackend cache invalidator #[async_trait::async_trait] @@ -100,6 +103,15 @@ where let key = ViewInfoKey::new(*table_id); self.invalidate_key(&key.to_bytes()).await; + + for key in [ + dropped_at_key(*table_id), + retention_expires_at_key(*table_id), + drop_generation_key(*table_id), + purging_key(*table_id), + ] { + self.invalidate_key(&to_tombstone_key(&key)).await; + } } CacheIdent::TableName(table_name) => { let key: TableNameKey = table_name.into(); diff --git a/src/common/meta/src/ddl/drop_table.rs b/src/common/meta/src/ddl/drop_table.rs index f3f085d1f2..177edd2553 100644 --- a/src/common/meta/src/ddl/drop_table.rs +++ b/src/common/meta/src/ddl/drop_table.rs @@ -47,6 +47,14 @@ use crate::region_keeper::OperatingRegionGuard; use crate::rpc::ddl::DropTableTask; use crate::rpc::router::{RegionRoute, operating_leader_region_roles}; +fn ensure_retry_later(err: error::Error) -> error::Error { + if err.is_retry_later() { + err + } else { + error::Error::retry_later(err) + } +} + pub struct DropTableProcedure { /// The context of procedure runtime. pub context: DdlContext, @@ -94,15 +102,13 @@ impl DropTableProcedure { } pub(crate) async fn on_prepare(&mut self) -> Result { - if self - .executor - .on_prepare(&self.context, self.data.soft_drop_enabled) - .await? - .stop() - { + if self.executor.on_prepare(&self.context).await?.stop() { return Ok(Status::done()); } self.fill_table_metadata().await?; + self.executor + .check_tombstone_conflict(&self.context, self.data.soft_drop_enabled) + .await?; if self.data.soft_drop_enabled && self.data.dropped_at.is_none() { let dropped_at = current_time_millis(); let retention_millis = @@ -155,12 +161,34 @@ impl DropTableProcedure { Ok(()) } - /// Removes the table metadata. + /// Closes soft-drop regions before removing the table metadata. + /// + /// Both operations stay in this procedure state so retries always close and flush regions + /// before moving their live metadata to tombstones. pub(crate) async fn on_delete_metadata(&mut self) -> Result { self.register_dropping_regions()?; - // NOTES: If the meta server is crashed after the `RemoveMetadata`, - // Corresponding regions of this table on the Datanode will be closed automatically. - // Then any future dropping operation will fail. + if self.data.soft_drop_enabled { + let storage = self + .context + .table_metadata_manager + .table_route_manager() + .table_route_storage(); + storage + .remap_region_routes(&mut self.data.physical_region_routes) + .await + .map_err(ensure_retry_later)?; + self.executor + .on_close_regions( + &self.context.node_manager, + &self.context.leader_region_registry, + &self.data.physical_region_routes, + true, + ) + .await + .map_err(ensure_retry_later)?; + } + // Hard drop still relies on regions being closed automatically if metasrv crashes after + // metadata removal. Soft drop has already closed them above. // TODO(weny): Considers introducing a RegionStatus to indicate the region is dropping. let table_id = self.data.table_id(); @@ -171,7 +199,8 @@ impl DropTableProcedure { self.data.physical_region_routes.clone(), ); // Deletes table metadata logically. - self.executor + let result = self + .executor .on_delete_metadata( &self.context, table_route_value, @@ -180,7 +209,13 @@ impl DropTableProcedure { self.data.retention_expires_at, self.data.drop_generation.as_deref(), ) - .await?; + .await; + if self.data.soft_drop_enabled { + result.map_err(ensure_retry_later)?; + self.data.allow_rollback = false; + } else { + result?; + } info!("Deleted table metadata for table {table_id}"); self.data.state = DropTableState::InvalidateTableCache; Ok(Status::executing(true)) @@ -188,7 +223,12 @@ impl DropTableProcedure { /// Broadcasts invalidate table cache instruction. async fn on_broadcast(&mut self) -> Result { - self.executor.invalidate_table_cache(&self.context).await?; + let result = self.executor.invalidate_table_cache(&self.context).await; + if self.data.soft_drop_enabled { + result.map_err(ensure_retry_later)?; + } else { + result?; + } self.data.state = DropTableState::DatanodeDropRegions; @@ -214,14 +254,6 @@ impl DropTableProcedure { } if self.data.soft_drop_enabled { - self.executor - .on_close_regions( - &self.context.node_manager, - &self.context.leader_region_registry, - &self.data.physical_region_routes, - true, - ) - .await?; self.context .deregister_failure_detectors(convert_region_routes_to_detecting_regions( &self.data.physical_region_routes, diff --git a/src/common/meta/src/ddl/drop_table/executor.rs b/src/common/meta/src/ddl/drop_table/executor.rs index 61e74d1b5c..0fdcf03fce 100644 --- a/src/common/meta/src/ddl/drop_table/executor.rs +++ b/src/common/meta/src/ddl/drop_table/executor.rs @@ -85,13 +85,7 @@ impl DropTableExecutor { /// Checks whether table exists. /// - Early returns if table not exists and `drop_if_exists` is `true`. /// - Throws an error if table not exists and `drop_if_exists` is `false`. - /// - Rejects dropping a recreated live table while an older tombstone still owns the same - /// fully qualified name. - pub async fn on_prepare( - &self, - ctx: &DdlContext, - soft_drop_enabled: bool, - ) -> Result> { + pub async fn on_prepare(&self, ctx: &DdlContext) -> Result> { let table_ref = self.table.table_ref(); let exist = ctx @@ -115,6 +109,17 @@ impl DropTableExecutor { } ); + Ok(Control::Continue(())) + } + + /// Rejects dropping a recreated live table while an older tombstone still owns the same + /// fully qualified name. + pub async fn check_tombstone_conflict( + &self, + ctx: &DdlContext, + soft_drop_enabled: bool, + ) -> Result<()> { + let table_ref = self.table.table_ref(); if let Some(dropped_table) = ctx .table_metadata_manager .get_dropped_table(&self.table) @@ -130,7 +135,7 @@ impl DropTableExecutor { .fail(); } - Ok(Control::Continue(())) + Ok(()) } /// Deletes the table metadata **logically**. @@ -573,7 +578,7 @@ mod tests { 1024, true, ); - let ctrl = executor.on_prepare(&ctx, false).await.unwrap(); + let ctrl = executor.on_prepare(&ctx).await.unwrap(); assert!(ctrl.stop()); // Drops a non-exists table @@ -582,7 +587,7 @@ mod tests { 1024, false, ); - let err = executor.on_prepare(&ctx, false).await.unwrap_err(); + let err = executor.on_prepare(&ctx).await.unwrap_err(); assert_matches!(err, error::Error::TableNotFound { .. }); // Drops a exists table @@ -600,7 +605,7 @@ mod tests { ) .await .unwrap(); - let ctrl = executor.on_prepare(&ctx, false).await.unwrap(); + let ctrl = executor.on_prepare(&ctx).await.unwrap(); assert!(!ctrl.stop()); } } diff --git a/src/common/meta/src/ddl/drop_table/metadata.rs b/src/common/meta/src/ddl/drop_table/metadata.rs index 213ebb461b..a3c9b278c9 100644 --- a/src/common/meta/src/ddl/drop_table/metadata.rs +++ b/src/common/meta/src/ddl/drop_table/metadata.rs @@ -14,7 +14,7 @@ use common_catalog::consts::FILE_ENGINE; use common_catalog::format_full_table_name; -use snafu::{OptionExt, ensure}; +use snafu::OptionExt; use store_api::metric_engine_consts::METRIC_ENGINE_NAME; use crate::ddl::drop_table::DropTableProcedure; @@ -49,25 +49,17 @@ impl DropTableProcedure { physical_table_route_value.region_routes.clone(), ); // TODO(hl): support soft-dropping logical tables. - ensure!( - !(self.data.soft_drop_enabled - && is_metric_engine_logical_table( - &table_info_value.table_info, - &table_route_value - )), - error::UnsupportedSnafu { - operation: "soft-dropping metric logical tables".to_string() - } - ); + if self.data.soft_drop_enabled + && is_metric_engine_logical_table(&table_info_value.table_info, &table_route_value) + { + self.data.soft_drop_enabled = false; + } if physical_table_id == self.data.table_id() { let engine = table_info_value.table_info.meta.engine; - ensure!( - !(self.data.soft_drop_enabled && engine == FILE_ENGINE), - error::UnsupportedSnafu { - operation: "soft-dropping file-engine tables".to_string() - } - ); + if self.data.soft_drop_enabled && engine == FILE_ENGINE { + self.data.soft_drop_enabled = false; + } // rollback only if dropping the metric physical table fails self.data.allow_rollback = engine.as_str() == METRIC_ENGINE_NAME; diff --git a/src/common/meta/src/ddl/purge_dropped_table.rs b/src/common/meta/src/ddl/purge_dropped_table.rs index 040a0d8e60..1548c72241 100644 --- a/src/common/meta/src/ddl/purge_dropped_table.rs +++ b/src/common/meta/src/ddl/purge_dropped_table.rs @@ -19,6 +19,7 @@ use common_procedure::error::{FromJsonSnafu, ToJsonSnafu}; use common_procedure::{ Context as ProcedureContext, LockKey, Procedure, Result as ProcedureResult, Status, }; +use common_telemetry::info; use common_time::util::current_time_millis; use common_wal::options::WalOptions; use serde::{Deserialize, Serialize}; @@ -123,11 +124,19 @@ impl PurgeDroppedTableProcedure { return Ok(Status::done()); }; self.update_dropped_table(dropped_table); + info!( + "Prepared purge dropped table procedure for {}", + self.data.purge_target_info() + ); self.data.state = PurgeDroppedTableState::DropRegions; Ok(Status::executing(true)) } async fn on_drop_regions(&mut self) -> Result { + info!( + "Purging dropped table regions for {}", + self.data.purge_target_info() + ); let expected_generation = self.data.drop_generation.as_deref().unwrap_or_default(); let purge_claim = self .context @@ -178,6 +187,10 @@ impl PurgeDroppedTableProcedure { } async fn on_delete_tombstone(&mut self) -> Result { + info!( + "Deleting dropped table tombstone for {}", + self.data.purge_target_info() + ); if self.data.purging_claimed { let expected_generation = self.data.drop_generation.as_deref().unwrap_or_default(); if self @@ -305,6 +318,16 @@ impl PurgeDroppedTableData { self.table_info.as_ref().unwrap() } + fn purge_target_info(&self) -> String { + match (self.table_name.as_ref(), self.table_id) { + (Some(table_name), Some(table_id)) => format!( + "catalog={}, schema={}, table={}, table_id={}", + table_name.catalog_name, table_name.schema_name, table_name.table_name, table_id + ), + _ => format!("table_id={}", self.task.table_id), + } + } + fn physical_region_routes(&self) -> Option<&[RegionRoute]> { match self.table_route_value() { TableRouteValue::Physical(route) => Some(&route.region_routes), diff --git a/src/common/meta/src/ddl/tests/drop_table.rs b/src/common/meta/src/ddl/tests/drop_table.rs index 9e43fbd823..8a45e66ba3 100644 --- a/src/common/meta/src/ddl/tests/drop_table.rs +++ b/src/common/meta/src/ddl/tests/drop_table.rs @@ -15,8 +15,10 @@ use std::assert_matches; use std::collections::{HashMap, HashSet}; use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; use std::time::Duration; +use api::region::RegionResponse; use api::v1::region::{RegionRequest, region_request}; use async_trait::async_trait; use common_catalog::consts::{DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME, FILE_ENGINE}; @@ -43,7 +45,7 @@ use crate::ddl::test_util::{ }; use crate::ddl::undrop_table::UndropTableProcedure; use crate::ddl::{DdlContext, DetectingRegion, RegionFailureDetectorController, TableMetadata}; -use crate::error::Error; +use crate::error::{self, Error}; use crate::key::MetadataKey; use crate::key::table_name::TableNameKey; use crate::key::table_route::TableRouteValue; @@ -710,6 +712,110 @@ async fn test_soft_drop_closes_regions_and_keeps_tombstone() { ); } +#[tokio::test] +async fn test_soft_drop_keeps_metadata_live_until_regions_close() { + let (tx, mut rx) = mpsc::channel(2); + let fail_close = AtomicBool::new(true); + let datanode_handler = DatanodeWatcher::new(tx).with_handler(move |_, _| { + if fail_close.swap(false, Ordering::SeqCst) { + return error::UnexpectedSnafu { + err_msg: "mock close error".to_string(), + } + .fail(); + } + + Ok(RegionResponse::new(0)) + }); + let node_manager = Arc::new(MockDatanodeManager::new(datanode_handler)); + let mut ddl_context = new_ddl_context(node_manager); + ddl_context.soft_drop_enabled = true; + let table_id = 1024; + let table_name = "foo"; + let task = test_create_table_task(table_name, table_id); + ddl_context + .table_metadata_manager + .create_table_metadata( + task.table_info.clone(), + TableRouteValue::physical(vec![RegionRoute { + region: Region::new_test(RegionId::new(table_id, 1)), + leader_peer: Some(Peer::new(1, "old-leader")), + follower_peers: vec![], + leader_state: None, + leader_down_since: None, + write_route_policy: None, + }]), + HashMap::new(), + ) + .await + .unwrap(); + + let task = new_drop_table_task(table_name, table_id, false); + let mut procedure = DropTableProcedure::new(task, ddl_context.clone()); + let ctx = new_test_procedure_context(); + procedure.execute(&ctx).await.unwrap(); + put_datanode_address(&ddl_context, 1, "new-leader").await; + let error = procedure.execute(&ctx).await.unwrap_err(); + assert!(error.is_retry_later()); + + let (peer, request) = rx.try_recv().unwrap(); + assert_eq!(peer.addr, "new-leader"); + let Some(region_request::Body::Close(request)) = request.body else { + unreachable!(); + }; + assert!(request.flush_on_close); + assert!(rx.try_recv().is_err()); + assert!( + ddl_context + .table_metadata_manager + .table_name_manager() + .get(TableNameKey::new( + DEFAULT_CATALOG_NAME, + DEFAULT_SCHEMA_NAME, + table_name, + )) + .await + .unwrap() + .is_some() + ); + assert!( + ddl_context + .table_metadata_manager + .get_dropped_table_by_id(table_id) + .await + .unwrap() + .is_none() + ); + + put_datanode_address(&ddl_context, 1, "newer-leader").await; + execute_procedure_until_done(&mut procedure).await; + + let (peer, request) = rx.try_recv().unwrap(); + assert_eq!(peer.addr, "newer-leader"); + assert_matches!(request.body, Some(region_request::Body::Close(_))); + assert!(rx.try_recv().is_err()); + assert!( + ddl_context + .table_metadata_manager + .table_name_manager() + .get(TableNameKey::new( + DEFAULT_CATALOG_NAME, + DEFAULT_SCHEMA_NAME, + table_name, + )) + .await + .unwrap() + .is_none() + ); + assert!( + ddl_context + .table_metadata_manager + .get_dropped_table_by_id(table_id) + .await + .unwrap() + .is_some() + ); +} + #[tokio::test] async fn test_soft_drop_timestamp_is_stable_across_retry_and_recovery() { let node_manager = Arc::new(MockDatanodeManager::new(NaiveDatanodeHandler)); @@ -912,6 +1018,10 @@ async fn test_hard_drop_recreated_table_fails_when_soft_tombstone_exists() { ); let err = procedure.on_prepare().await.unwrap_err(); + assert_eq!( + "Cannot drop table 'greptime.public.foo': an older tombstone already uses the same full name", + err.to_string() + ); assert_matches!(err, Error::TableNameTombstoneConflict { .. }); } @@ -1191,7 +1301,7 @@ async fn test_undrop_logical_table_skips_datanode_open() { } #[tokio::test] -async fn test_soft_drop_metric_logical_table_fails() { +async fn test_soft_drop_metric_logical_table_falls_back_to_hard_drop() { let node_manager = Arc::new(MockDatanodeManager::new(NaiveDatanodeHandler)); let mut ddl_context = new_ddl_context(node_manager); ddl_context.soft_drop_enabled = true; @@ -1203,18 +1313,70 @@ async fn test_soft_drop_metric_logical_table_fails() { new_drop_table_task("foo", logical_table_id, false), ddl_context.clone(), ); - let err = procedure.on_prepare().await.unwrap_err(); - assert_eq!(err.status_code(), StatusCode::Unsupported); - let persisted_soft_drop = procedure.dump().unwrap(); + procedure.on_prepare().await.unwrap(); + assert!(!procedure.data.soft_drop_enabled); + let persisted_hard_drop = procedure.dump().unwrap(); ddl_context.soft_drop_enabled = false; - let mut recovered = - DropTableProcedure::from_json(&persisted_soft_drop, ddl_context.clone()).unwrap(); - let err = recovered.on_prepare().await.unwrap_err(); - assert_eq!(err.status_code(), StatusCode::Unsupported); + let recovered = + DropTableProcedure::from_json(&persisted_hard_drop, ddl_context.clone()).unwrap(); + assert!(!recovered.data.soft_drop_enabled); } #[tokio::test] -async fn test_soft_drop_file_engine_table_fails() { +async fn test_soft_drop_metric_physical_table_remains_enabled() { + let (tx, mut rx) = mpsc::channel(8); + let datanode_handler = DatanodeWatcher::new(tx); + let node_manager = Arc::new(MockDatanodeManager::new(datanode_handler)); + let mut ddl_context = new_ddl_context(node_manager); + ddl_context.soft_drop_enabled = true; + ddl_context.soft_drop_retention = Some(Duration::from_millis(100)); + let table_id = 1024; + let table_name = "phy"; + let mut task = test_create_physical_table_task(table_name); + task.set_table_id(table_id); + ddl_context + .table_metadata_manager + .create_table_metadata( + task.table_info, + TableRouteValue::physical(vec![RegionRoute { + region: Region::new_test(RegionId::new(table_id, 1)), + leader_peer: Some(Peer::empty(1)), + follower_peers: vec![], + leader_state: None, + leader_down_since: None, + write_route_policy: None, + }]), + HashMap::new(), + ) + .await + .unwrap(); + + let mut procedure = DropTableProcedure::new( + new_drop_table_task(table_name, table_id, false), + ddl_context.clone(), + ); + execute_procedure_until_done(&mut procedure).await; + + assert!(procedure.data.soft_drop_enabled); + assert!(procedure.data.dropped_at.is_some()); + let (_, request) = rx.try_recv().unwrap(); + let Some(region_request::Body::Close(request)) = request.body else { + unreachable!(); + }; + assert!(request.flush_on_close); + assert!(rx.try_recv().is_err()); + assert!( + ddl_context + .table_metadata_manager + .get_dropped_table_by_id(table_id) + .await + .unwrap() + .is_some() + ); +} + +#[tokio::test] +async fn test_soft_drop_file_engine_table_falls_back_to_hard_drop() { let node_manager = Arc::new(MockDatanodeManager::new(NaiveDatanodeHandler)); let mut ddl_context = new_ddl_context(node_manager); ddl_context.soft_drop_enabled = true; @@ -1236,14 +1398,44 @@ async fn test_soft_drop_file_engine_table_fails() { new_drop_table_task(table_name, table_id, false), ddl_context.clone(), ); - let err = procedure.on_prepare().await.unwrap_err(); - assert_eq!(err.status_code(), StatusCode::Unsupported); - let persisted_soft_drop = procedure.dump().unwrap(); + procedure.on_prepare().await.unwrap(); + assert!(!procedure.data.soft_drop_enabled); + let persisted_hard_drop = procedure.dump().unwrap(); ddl_context.soft_drop_enabled = false; - let mut recovered = - DropTableProcedure::from_json(&persisted_soft_drop, ddl_context.clone()).unwrap(); - let err = recovered.on_prepare().await.unwrap_err(); - assert_eq!(err.status_code(), StatusCode::Unsupported); + let recovered = + DropTableProcedure::from_json(&persisted_hard_drop, ddl_context.clone()).unwrap(); + assert!(!recovered.data.soft_drop_enabled); +} + +#[tokio::test] +async fn test_file_engine_fallback_is_resolved_before_tombstone_conflict() { + let node_manager = Arc::new(MockDatanodeManager::new(NaiveDatanodeHandler)); + let mut ddl_context = new_ddl_context(node_manager); + ddl_context.soft_drop_enabled = true; + let original_table_id = 1024; + let recreated_table_id = 1025; + let table_name = "foo"; + create_dropped_table(&ddl_context, original_table_id, None, None).await; + + let mut task = test_create_table_task(table_name, recreated_table_id); + task.table_info.meta.engine = FILE_ENGINE.to_string(); + ddl_context + .table_metadata_manager + .create_table_metadata( + task.table_info, + TableRouteValue::physical(vec![]), + HashMap::new(), + ) + .await + .unwrap(); + + let mut procedure = DropTableProcedure::new( + new_drop_table_task(table_name, recreated_table_id, false), + ddl_context, + ); + procedure.on_prepare().await.unwrap(); + + assert!(!procedure.data.soft_drop_enabled); } #[tokio::test] @@ -2332,7 +2524,6 @@ async fn test_on_rollback() { let node_manager = Arc::new(MockDatanodeManager::new(NaiveDatanodeHandler)); let kv_backend = Arc::new(MemoryKvBackend::new()); let mut ddl_context = new_ddl_context_with_kv_backend(node_manager, kv_backend.clone()); - ddl_context.soft_drop_enabled = true; // Prepares physical table metadata. let mut create_physical_table_task = test_create_physical_table_task("phy_table"); let TableMetadata { @@ -2373,36 +2564,8 @@ async fn test_on_rollback() { procedure.on_prepare().await.unwrap(); assert!(procedure.rollback_supported()); procedure.on_delete_metadata().await.unwrap(); - assert!( - kv_backend - .get(dropped_at_marker_key(physical_table_id).as_bytes()) - .await - .unwrap() - .is_some() - ); - assert!( - kv_backend - .get(retention_expires_at_marker_key(physical_table_id).as_bytes()) - .await - .unwrap() - .is_some() - ); assert!(procedure.rollback_supported()); procedure.rollback(&ctx).await.unwrap(); - assert!( - kv_backend - .get(dropped_at_marker_key(physical_table_id).as_bytes()) - .await - .unwrap() - .is_none() - ); - assert!( - kv_backend - .get(retention_expires_at_marker_key(physical_table_id).as_bytes()) - .await - .unwrap() - .is_none() - ); // Rollback again assert!(procedure.rollback_supported()); procedure.rollback(&ctx).await.unwrap(); diff --git a/src/common/meta/src/error.rs b/src/common/meta/src/error.rs index 6832940d11..2ef880da2f 100644 --- a/src/common/meta/src/error.rs +++ b/src/common/meta/src/error.rs @@ -380,9 +380,8 @@ pub enum Error { }, #[snafu(display( - "Cannot drop table '{}': tombstoned table id {} already uses the same full name", - table_name, - existing_table_id + "Cannot drop table '{}': an older tombstone already uses the same full name", + table_name ))] /// Raised when a live table is recreated with a name still reserved by an older tombstone. TableNameTombstoneConflict { diff --git a/src/common/meta/src/key.rs b/src/common/meta/src/key.rs index 71c3a72e7d..c9005e20d8 100644 --- a/src/common/meta/src/key.rs +++ b/src/common/meta/src/key.rs @@ -484,19 +484,19 @@ const RETENTION_EXPIRES_AT_KEY_PREFIX: &str = "__retention_expires_at"; const DROP_GENERATION_KEY_PREFIX: &str = "__drop_generation"; const PURGING_KEY_PREFIX: &str = "__purging"; -fn dropped_at_key(table_id: TableId) -> Vec { +pub(crate) fn dropped_at_key(table_id: TableId) -> Vec { format!("{DROPPED_AT_KEY_PREFIX}/{table_id}").into_bytes() } -fn retention_expires_at_key(table_id: TableId) -> Vec { +pub(crate) fn retention_expires_at_key(table_id: TableId) -> Vec { format!("{RETENTION_EXPIRES_AT_KEY_PREFIX}/{table_id}").into_bytes() } -fn drop_generation_key(table_id: TableId) -> Vec { +pub(crate) fn drop_generation_key(table_id: TableId) -> Vec { format!("{DROP_GENERATION_KEY_PREFIX}/{table_id}").into_bytes() } -fn purging_key(table_id: TableId) -> Vec { +pub(crate) fn purging_key(table_id: TableId) -> Vec { format!("{PURGING_KEY_PREFIX}/{table_id}").into_bytes() } @@ -3421,7 +3421,7 @@ mod tests { } #[tokio::test] - async fn test_dropped_table_lookup_ignores_unrelated_malformed_datanode_tombstones() { + async fn test_dropped_table_exact_lookup_ignores_unrelated_malformed_table_name_tombstone() { let table_id = 1025; let table_name = "foo"; let (mem_kv, table_metadata_manager, table_name, table_info, region_routes, options) = @@ -3439,7 +3439,7 @@ mod tests { mem_kv .put( PutRequest::new() - .with_key("__tombstone/__dn_table/not-a-datanode-table-key") + .with_key("__tombstone/__table_name/not-a-table-name-key") .with_value("malformed"), ) .await @@ -3460,6 +3460,63 @@ mod tests { ); } + #[tokio::test] + async fn test_dropped_table_exact_lookup_tracks_reused_name_after_purge() { + let old_table_id = 1025; + let new_table_id = 1026; + let (_, manager, table_name, _, old_routes, old_options) = + create_dropped_physical_table_metadata( + old_table_id, + "foo", + vec![test_physical_region_route(old_table_id, 1, 1, vec![])], + HashMap::new(), + ) + .await; + + manager + .delete_table_metadata_tombstone( + old_table_id, + &table_name, + &TableRouteValue::physical(old_routes), + &old_options, + ) + .await + .unwrap(); + let new_task = test_create_table_task("foo", new_table_id); + let new_info = new_task.table_info.clone(); + let new_route = + TableRouteValue::physical(vec![test_physical_region_route(new_table_id, 1, 1, vec![])]); + manager + .create_table_metadata(new_task.table_info, new_route.clone(), HashMap::new()) + .await + .unwrap(); + manager + .delete_table_metadata(new_table_id, &table_name, &new_route, &HashMap::new(), None) + .await + .unwrap(); + + let dropped = manager + .get_dropped_table(&table_name) + .await + .unwrap() + .unwrap(); + assert_eq!(new_table_id, dropped.table_id); + assert_eq!(new_info, dropped.table_info_value.table_info); + } + + #[tokio::test] + async fn test_dropped_table_exact_lookup_missing() { + let manager = TableMetadataManager::new(Arc::new(MemoryKvBackend::default())); + + assert!( + manager + .get_dropped_table(&TableName::new("greptime", "public", "missing")) + .await + .unwrap() + .is_none() + ); + } + #[tokio::test] async fn test_create_update_view_info() { let mem_kv = Arc::new(MemoryKvBackend::default()); diff --git a/src/common/meta/src/key/tombstone.rs b/src/common/meta/src/key/tombstone.rs index 4e6699d407..51c6e9b923 100644 --- a/src/common/meta/src/key/tombstone.rs +++ b/src/common/meta/src/key/tombstone.rs @@ -62,6 +62,10 @@ const TOMBSTONE_PREFIX: &str = "__tombstone/"; const MOVE_VALUE_TXN_OPS_PER_KEY: usize = 4; const RESTORE_VALUE_TXN_OPS_PER_KEY: usize = 6; +pub(crate) fn to_tombstone_key(key: &[u8]) -> Vec { + [TOMBSTONE_PREFIX.as_bytes(), key].concat() +} + impl TombstoneManager { /// Returns [TombstoneManager]. pub fn new(kv_backend: KvBackendRef) -> Self { @@ -101,7 +105,11 @@ impl TombstoneManager { /// Gets a single tombstoned value by its original key. pub async fn get(&self, key: &[u8]) -> Result> { let tombstone_key = self.to_tombstone(key); - self.kv_backend.get(&tombstone_key).await + let response = self + .kv_backend + .range(RangeRequest::new().with_key(tombstone_key)) + .await?; + Ok(response.kvs.into_iter().next()) } /// Gets tombstoned values by their original keys. @@ -554,6 +562,7 @@ mod tests { use std::any::Any; use std::collections::HashMap; + use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::{Arc, Mutex}; use crate::error::{Error, Result}; @@ -563,6 +572,7 @@ mod tests { use crate::kv_backend::memory::MemoryKvBackend; use crate::kv_backend::txn::{Txn, TxnRequest, TxnResponse}; use crate::kv_backend::{KvBackend, TxnService}; + use crate::rpc::KeyValue; use crate::rpc::store::{ BatchDeleteRequest, BatchDeleteResponse, BatchGetRequest, BatchGetResponse, BatchPutRequest, BatchPutResponse, DeleteRangeRequest, DeleteRangeResponse, PutRequest, @@ -635,6 +645,64 @@ mod tests { } } + struct StalePointGetKvBackend { + inner: Arc>, + get_calls: AtomicUsize, + range_calls: AtomicUsize, + } + + #[async_trait::async_trait] + impl TxnService for StalePointGetKvBackend { + type Error = Error; + + async fn txn(&self, txn: Txn) -> Result { + self.inner.txn(txn).await + } + } + + #[async_trait::async_trait] + impl KvBackend for StalePointGetKvBackend { + fn name(&self) -> &str { + "stale_point_get" + } + + fn as_any(&self) -> &dyn Any { + self + } + + async fn range(&self, req: RangeRequest) -> Result { + assert_eq!(b"__tombstone/foo", req.key.as_slice()); + assert!(req.range_end.is_empty()); + self.range_calls.fetch_add(1, Ordering::SeqCst); + self.inner.range(req).await + } + + async fn get(&self, key: &[u8]) -> Result> { + self.get_calls.fetch_add(1, Ordering::SeqCst); + Ok(Some((key.to_vec(), b"stale".to_vec()).into())) + } + + async fn put(&self, req: PutRequest) -> Result { + self.inner.put(req).await + } + + async fn batch_put(&self, req: BatchPutRequest) -> Result { + self.inner.batch_put(req).await + } + + async fn batch_get(&self, req: BatchGetRequest) -> Result { + self.inner.batch_get(req).await + } + + async fn delete_range(&self, req: DeleteRangeRequest) -> Result { + self.inner.delete_range(req).await + } + + async fn batch_delete(&self, req: BatchDeleteRequest) -> Result { + self.inner.batch_delete(req).await + } + } + #[derive(Debug, Clone)] struct MoveValue { key: Vec, @@ -699,6 +767,31 @@ mod tests { assert_eq!(kv_backend.len(), 2); } + #[tokio::test] + async fn test_get_tombstone_bypasses_stale_point_get() { + let inner = Arc::new(MemoryKvBackend::default()); + let backend = Arc::new(StalePointGetKvBackend { + inner: inner.clone(), + get_calls: AtomicUsize::new(0), + range_calls: AtomicUsize::new(0), + }); + let manager = TombstoneManager::new(backend.clone()); + inner + .put( + PutRequest::new() + .with_key(manager.to_tombstone(b"foo")) + .with_value(b"authoritative"), + ) + .await + .unwrap(); + + let value = manager.get(b"foo").await.unwrap().unwrap(); + + assert_eq!(b"authoritative", value.value.as_slice()); + assert_eq!(0, backend.get_calls.load(Ordering::SeqCst)); + assert_eq!(1, backend.range_calls.load(Ordering::SeqCst)); + } + #[tokio::test] async fn test_create_tombstone_with_non_exist_values() { let kv_backend = Arc::new(MemoryKvBackend::default()); diff --git a/src/frontend/src/instance/builder.rs b/src/frontend/src/instance/builder.rs index a9e7968219..0de3d9b40a 100644 --- a/src/frontend/src/instance/builder.rs +++ b/src/frontend/src/instance/builder.rs @@ -231,9 +231,11 @@ impl FrontendBuilder { requester, )); + let table_metadata_manager = Arc::new(TableMetadataManager::new(kv_backend.clone())); let procedure_service_handler = Arc::new(ProcedureServiceOperator::new( self.procedure_executor.clone(), self.catalog_manager.clone(), + table_metadata_manager.clone(), )); let flow_metadata_manager: Arc = @@ -308,7 +310,7 @@ impl FrontendBuilder { plugins, inserter, deleter, - table_metadata_manager: Arc::new(TableMetadataManager::new(kv_backend)), + table_metadata_manager, event_recorder: Some(event_recorder), process_manager, otlp_metrics_table_legacy_cache: DashMap::new(), diff --git a/src/meta-srv/src/gc.rs b/src/meta-srv/src/gc.rs index f64f52eaeb..5ee2a2e779 100644 --- a/src/meta-srv/src/gc.rs +++ b/src/meta-srv/src/gc.rs @@ -30,7 +30,6 @@ mod tracker; mod util; pub(crate) use ctx::DefaultGcSchedulerCtx; -pub(crate) use options::EXPERIMENTAL_SOFT_DROP_ENABLED; pub use options::GcSchedulerOptions; pub use procedure::BatchGcProcedure; pub use scheduler::{Event, GcJobReport, GcScheduler, GcTickerRef}; diff --git a/src/meta-srv/src/gc/options.rs b/src/meta-srv/src/gc/options.rs index 4d1b93b9d9..826826de99 100644 --- a/src/meta-srv/src/gc/options.rs +++ b/src/meta-srv/src/gc/options.rs @@ -23,21 +23,18 @@ use crate::error::{self, Result}; #[allow(unused)] pub(crate) const TICKER_INTERVAL: Duration = Duration::from_secs(60 * 5); -/// Soft drop remains disconnected from user configuration until it is ready. -pub(crate) const EXPERIMENTAL_SOFT_DROP_ENABLED: bool = false; - -/// Reserved configuration for garbage collecting soft-dropped tables. +/// Configuration for garbage collecting soft-dropped tables. #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] #[serde(default)] -pub struct ExperimentalSoftDropGcOptions { - /// Reserved. This option is currently ignored. +pub struct SoftDropGcOptions { + /// Whether soft drop is enabled. pub enable: bool, - /// Reserved retention duration. This option is currently ignored. + /// How long soft-dropped tables are retained before automatic purge. #[serde(with = "humantime_serde")] pub retention: Duration, } -impl Default for ExperimentalSoftDropGcOptions { +impl Default for SoftDropGcOptions { fn default() -> Self { Self { enable: false, @@ -56,8 +53,8 @@ pub struct GcSchedulerOptions { /// If set to false, no GC will be performed, and potentially some /// files from datanodes will never be deleted. pub enable: bool, - /// Reserved experimental soft-drop garbage collection options. - pub experimental_soft_drop: ExperimentalSoftDropGcOptions, + /// Experimental soft-drop garbage collection options. + pub experimental_soft_drop: SoftDropGcOptions, /// Maximum number of tables to process concurrently. pub max_concurrent_tables: usize, /// Maximum number of retries per region when GC fails. @@ -98,7 +95,7 @@ impl Default for GcSchedulerOptions { fn default() -> Self { Self { enable: false, - experimental_soft_drop: ExperimentalSoftDropGcOptions::default(), + experimental_soft_drop: SoftDropGcOptions::default(), max_concurrent_tables: 10, max_retries_per_region: 3, retry_backoff_duration: Duration::from_secs(5), @@ -120,6 +117,32 @@ impl Default for GcSchedulerOptions { impl GcSchedulerOptions { /// Validates the configuration options. pub fn validate(&self) -> Result<()> { + ensure!( + !self.experimental_soft_drop.enable || self.enable, + error::InvalidArgumentsSnafu { + err_msg: "gc.enable must be true when soft drop is enabled", + } + ); + + if !self.enable { + return Ok(()); + } + + if self.experimental_soft_drop.enable { + ensure!( + self.experimental_soft_drop.retention.as_millis() > 0, + error::InvalidArgumentsSnafu { + err_msg: "soft drop retention must be at least 1ms", + } + ); + ensure!( + self.experimental_soft_drop.retention.as_millis() <= i64::MAX as u128, + error::InvalidArgumentsSnafu { + err_msg: "soft drop retention must fit in an i64 millisecond value", + } + ); + } + ensure!( self.max_concurrent_tables > 0, error::InvalidArgumentsSnafu { @@ -206,7 +229,7 @@ mod tests { use super::*; #[test] - fn test_experimental_soft_drop_defaults() { + fn test_soft_drop_defaults() { let options = GcSchedulerOptions::default(); assert!(!options.experimental_soft_drop.enable); @@ -217,10 +240,10 @@ mod tests { } #[test] - fn test_experimental_soft_drop_valid_when_gc_is_enabled() { + fn test_soft_drop_valid_when_gc_is_enabled() { let options = GcSchedulerOptions { enable: true, - experimental_soft_drop: ExperimentalSoftDropGcOptions { + experimental_soft_drop: SoftDropGcOptions { enable: true, retention: Duration::from_days(1), }, @@ -231,15 +254,57 @@ mod tests { } #[test] - fn test_experimental_soft_drop_options_are_ignored_by_validation() { + fn test_soft_drop_requires_gc() { let options = GcSchedulerOptions { - experimental_soft_drop: ExperimentalSoftDropGcOptions { + experimental_soft_drop: SoftDropGcOptions { + enable: true, + retention: Duration::from_days(1), + }, + ..Default::default() + }; + + assert!(options.validate().is_err()); + } + + #[test] + fn test_soft_drop_retention_must_be_positive() { + let options = GcSchedulerOptions { + enable: true, + experimental_soft_drop: SoftDropGcOptions { enable: true, retention: Duration::ZERO, }, ..Default::default() }; - assert!(options.validate().is_ok()); + assert!(options.validate().is_err()); + } + + #[test] + fn test_soft_drop_retention_must_be_at_least_one_millisecond() { + let options = GcSchedulerOptions { + enable: true, + experimental_soft_drop: SoftDropGcOptions { + enable: true, + retention: Duration::from_micros(999), + }, + ..Default::default() + }; + + assert!(options.validate().is_err()); + } + + #[test] + fn test_soft_drop_retention_must_fit_i64_millis() { + let options = GcSchedulerOptions { + enable: true, + experimental_soft_drop: SoftDropGcOptions { + enable: true, + retention: Duration::from_millis(i64::MAX as u64 + 1), + }, + ..Default::default() + }; + + assert!(options.validate().is_err()); } } diff --git a/src/meta-srv/src/gc/scheduler.rs b/src/meta-srv/src/gc/scheduler.rs index 371ed34e11..ea7dd8e284 100644 --- a/src/meta-srv/src/gc/scheduler.rs +++ b/src/meta-srv/src/gc/scheduler.rs @@ -27,13 +27,13 @@ use tokio::sync::{Mutex, oneshot}; use crate::define_ticker; use crate::error::{self, Error, Result}; +use crate::gc::Region2Peers; #[cfg(test)] use crate::gc::ctx::PurgeReservation; use crate::gc::ctx::{PurgeOutcome, SchedulerCtx}; use crate::gc::dropped::DroppedRegionCollector; use crate::gc::options::{GcSchedulerOptions, TICKER_INTERVAL}; use crate::gc::tracker::RegionGcTracker; -use crate::gc::{EXPERIMENTAL_SOFT_DROP_ENABLED, Region2Peers}; use crate::metrics::{ METRIC_META_GC_SCHEDULER_CYCLES_TOTAL, METRIC_META_GC_SCHEDULER_DURATION_SECONDS, METRIC_META_GC_SOFT_DROP_PURGES_TOTAL, @@ -207,7 +207,7 @@ impl GcScheduler { info!("Skip gc trigger because maintenance mode is enabled"); return Ok(GcJobReport::default()); } - if EXPERIMENTAL_SOFT_DROP_ENABLED && self.config.experimental_soft_drop.enable { + if self.config.experimental_soft_drop.enable { self.purge_expired_soft_dropped_tables(common_time::util::current_time_millis()) .await; } @@ -603,7 +603,7 @@ mod tests { receiver: rx, config: GcSchedulerOptions { enable: true, - experimental_soft_drop: crate::gc::options::ExperimentalSoftDropGcOptions { + experimental_soft_drop: crate::gc::options::SoftDropGcOptions { enable: true, retention: Duration::from_millis(100), }, @@ -721,14 +721,15 @@ mod tests { } #[tokio::test] - async fn test_tick_ignores_experimental_soft_drop_config_and_still_runs_region_gc() { + async fn test_tick_purges_expired_soft_dropped_tables_and_runs_region_gc() { let ctx = Arc::new(SoftDropSchedulerCtx::default()); *ctx.dropped_tables.lock().unwrap() = vec![dropped_table(1, Some(i64::MIN))]; let scheduler = soft_drop_scheduler(ctx.clone()); scheduler.handle_tick().await.unwrap(); + wait_for_purge_attempts(&ctx, 1).await; - assert!(ctx.purge_attempts.lock().unwrap().is_empty()); + assert_eq!(vec![1], *ctx.purge_attempts.lock().unwrap()); assert_eq!(1, ctx.region_gc_calls.load(Ordering::Relaxed)); } diff --git a/src/meta-srv/src/metasrv/builder.rs b/src/meta-srv/src/metasrv/builder.rs index fd7be6ae6d..fe36553172 100644 --- a/src/meta-srv/src/metasrv/builder.rs +++ b/src/meta-srv/src/metasrv/builder.rs @@ -58,7 +58,7 @@ use crate::cache_invalidator::MetasrvCacheInvalidator; use crate::cluster::MetaPeerClientRef; use crate::error::{self, BuildWalProviderSnafu, OtherSnafu, Result}; use crate::events::EventHandlerImpl; -use crate::gc::{DefaultGcSchedulerCtx, EXPERIMENTAL_SOFT_DROP_ENABLED, GcScheduler}; +use crate::gc::{DefaultGcSchedulerCtx, GcScheduler}; use crate::greptimedb_telemetry::get_greptimedb_telemetry_task; use crate::handler::failure_handler::RegionFailureHandler; use crate::handler::flow_state_handler::FlowStateHandler; @@ -219,6 +219,7 @@ impl MetasrvBuilder { } = self; let options = options.unwrap_or_default(); + options.gc.validate()?; let kv_backend = kv_backend.unwrap_or_else(|| Arc::new(MemoryKvBackend::new())); let in_memory = in_memory.unwrap_or_else(|| Arc::new(MemoryKvBackend::new())); @@ -693,7 +694,7 @@ fn build_procedure_manager( /// Resolves if soft-drop is enabled from metasrv options. fn ddl_soft_drop_enabled(options: &MetasrvOptions) -> bool { - EXPERIMENTAL_SOFT_DROP_ENABLED && options.gc.experimental_soft_drop.enable + options.gc.experimental_soft_drop.enable } /// Returns soft-drop retention for recovering persisted procedures. @@ -724,7 +725,7 @@ mod tests { options.gc.experimental_soft_drop.enable = true; options.gc.experimental_soft_drop.retention = Duration::from_secs(123); - assert!(!ddl_soft_drop_enabled(&options)); + assert!(ddl_soft_drop_enabled(&options)); assert_eq!( Some(Duration::from_secs(123)), ddl_soft_drop_retention(&options) @@ -737,12 +738,27 @@ mod tests { } #[test] - fn test_ignored_soft_drop_options_are_not_validated() { + fn test_soft_drop_options_are_validated() { let mut options = MetasrvOptions::default(); + options.gc.enable = true; options.gc.experimental_soft_drop.enable = true; options.gc.experimental_soft_drop.retention = Duration::ZERO; - assert!(options.gc.validate().is_ok()); + assert!(options.gc.validate().is_err()); + } + + #[tokio::test] + async fn test_builder_rejects_soft_drop_when_gc_is_disabled() { + let mut options = MetasrvOptions::default(); + options.gc.experimental_soft_drop.enable = true; + + assert!( + MetasrvBuilder::new() + .options(options) + .build() + .await + .is_err() + ); } #[tokio::test] diff --git a/src/operator/src/procedure.rs b/src/operator/src/procedure.rs index d98c09b77c..cd2a0842f1 100644 --- a/src/operator/src/procedure.rs +++ b/src/operator/src/procedure.rs @@ -17,7 +17,9 @@ use async_trait::async_trait; use catalog::CatalogManagerRef; use common_error::ext::BoxedError; use common_function::handlers::ProcedureServiceHandler; +use common_meta::key::TableMetadataManagerRef; use common_meta::procedure_executor::{ExecutorContext, ProcedureExecutorRef}; +use common_meta::rpc::ddl::{DdlTask, SubmitDdlTaskRequest}; use common_meta::rpc::procedure::{ GcRegionsRequest as MetaGcRegionsRequest, GcResponse as MetaGcResponse, GcTableRequest as MetaGcTableRequest, ManageRegionFollowerRequest, MigrateRegionRequest, @@ -26,28 +28,66 @@ use common_meta::rpc::procedure::{ use common_query::error as query_error; use common_query::error::Result as QueryResult; use snafu::ResultExt; +use table::table_name::TableName; + +use crate::error; +use crate::utils::to_meta_query_context; /// The operator for procedures which implements [`ProcedureServiceHandler`]. #[derive(Clone)] pub struct ProcedureServiceOperator { procedure_executor: ProcedureExecutorRef, catalog_manager: CatalogManagerRef, + table_metadata_manager: TableMetadataManagerRef, } impl ProcedureServiceOperator { pub fn new( procedure_executor: ProcedureExecutorRef, catalog_manager: CatalogManagerRef, + table_metadata_manager: TableMetadataManagerRef, ) -> Self { Self { procedure_executor, catalog_manager, + table_metadata_manager, } } } #[async_trait] impl ProcedureServiceHandler for ProcedureServiceOperator { + async fn purge_table( + &self, + table_name: TableName, + query_ctx: session::context::QueryContextRef, + ) -> QueryResult<()> { + let dropped = self + .table_metadata_manager + .get_dropped_table(&table_name) + .await + .map_err(BoxedError::new) + .context(query_error::ProcedureServiceSnafu)? + .ok_or_else(|| { + error::TableNotFoundSnafu { + table_name: table_name.to_string(), + } + .build() + }) + .map_err(BoxedError::new) + .context(query_error::ProcedureServiceSnafu)?; + let request = SubmitDdlTaskRequest::new( + to_meta_query_context(query_ctx), + DdlTask::new_purge_dropped_table(dropped.table_id), + ); + self.procedure_executor + .submit_ddl_task(&ExecutorContext::default(), request) + .await + .map_err(BoxedError::new) + .context(query_error::ProcedureServiceSnafu)?; + Ok(()) + } + async fn migrate_region(&self, request: MigrateRegionRequest) -> QueryResult> { Ok(self .procedure_executor @@ -109,3 +149,182 @@ impl ProcedureServiceHandler for ProcedureServiceOperator { .context(query_error::ProcedureServiceSnafu) } } + +#[cfg(test)] +mod tests { + use std::collections::HashMap; + use std::sync::{Arc, Mutex}; + + use api::v1::meta::{ProcedureDetailResponse, ReconcileResponse}; + use common_meta::key::TableMetadataManager; + use common_meta::key::table_route::TableRouteValue; + use common_meta::key::test_utils::new_test_table_info_with_name; + use common_meta::kv_backend::memory::MemoryKvBackend; + use common_meta::procedure_executor::{ProcedureExecutor, ProcedureExecutorRef}; + use common_meta::rpc::ddl::{DdlTask, SubmitDdlTaskRequest, SubmitDdlTaskResponse}; + use common_meta::rpc::procedure::{MigrateRegionResponse, ProcedureStateResponse}; + use session::context::QueryContextBuilder; + use table::table_name::TableName; + + use super::*; + + #[derive(Default)] + struct RecordingProcedureExecutor { + requests: Mutex>, + fail: bool, + } + + #[async_trait] + impl ProcedureExecutor for RecordingProcedureExecutor { + async fn submit_ddl_task( + &self, + _: &ExecutorContext, + request: SubmitDdlTaskRequest, + ) -> common_meta::error::Result { + self.requests.lock().unwrap().push(request); + if self.fail { + return common_meta::error::UnsupportedSnafu { + operation: "test purge failure", + } + .fail(); + } + Ok(SubmitDdlTaskResponse::default()) + } + async fn migrate_region( + &self, + _: &ExecutorContext, + _: MigrateRegionRequest, + ) -> common_meta::error::Result { + unimplemented!() + } + async fn reconcile( + &self, + _: &ExecutorContext, + _: ReconcileRequest, + ) -> common_meta::error::Result { + unimplemented!() + } + async fn query_procedure_state( + &self, + _: &ExecutorContext, + _: &str, + ) -> common_meta::error::Result { + unimplemented!() + } + async fn list_procedures( + &self, + _: &ExecutorContext, + ) -> common_meta::error::Result { + unimplemented!() + } + } + + async fn operator_with_tombstone( + table_id: u32, + name: &TableName, + fail: bool, + ) -> (ProcedureServiceOperator, Arc) { + let manager = Arc::new(TableMetadataManager::new(Arc::new( + MemoryKvBackend::default(), + ))); + let mut info = new_test_table_info_with_name(table_id, &name.table_name); + info.catalog_name = name.catalog_name.clone(); + info.schema_name = name.schema_name.clone(); + let route = TableRouteValue::physical(vec![]); + manager + .create_table_metadata(info, route.clone(), HashMap::new()) + .await + .unwrap(); + manager + .delete_table_metadata(table_id, name, &route, &HashMap::new(), None) + .await + .unwrap(); + let executor = Arc::new(RecordingProcedureExecutor { + fail, + ..Default::default() + }); + let operator = ProcedureServiceOperator::new( + executor.clone() as ProcedureExecutorRef, + catalog::memory::MemoryCatalogManager::new(), + manager, + ); + (operator, executor) + } + + #[tokio::test] + async fn test_purge_table_submits_tombstone_id_and_query_context() { + let name = TableName::new("catalog", "schema", "metrics"); + let (operator, executor) = operator_with_tombstone(42, &name, false).await; + let manager = operator.table_metadata_manager.clone(); + let mut live = new_test_table_info_with_name(99, "metrics"); + live.catalog_name = "catalog".to_string(); + live.schema_name = "schema".to_string(); + manager + .create_table_metadata(live, TableRouteValue::physical(vec![]), HashMap::new()) + .await + .unwrap(); + let query_ctx = QueryContextBuilder::default() + .current_catalog("catalog".to_string()) + .current_schema("schema".to_string()) + .build() + .into(); + + operator.purge_table(name.clone(), query_ctx).await.unwrap(); + + let requests = executor.requests.lock().unwrap(); + assert!( + matches!(&requests[0].task, DdlTask::PurgeDroppedTable(task) if task.table_id == 42) + ); + assert_eq!(requests[0].query_context.current_catalog, "catalog"); + assert_eq!(requests[0].query_context.current_schema, "schema"); + } + + #[tokio::test] + async fn test_purge_table_missing_tombstone_is_table_not_found() { + let manager = Arc::new(TableMetadataManager::new(Arc::new( + MemoryKvBackend::default(), + ))); + let executor = Arc::new(RecordingProcedureExecutor::default()); + let operator = ProcedureServiceOperator::new( + executor.clone(), + catalog::memory::MemoryCatalogManager::new(), + manager, + ); + let error = operator + .purge_table( + TableName::new("catalog", "schema", "missing"), + QueryContextBuilder::default() + .current_catalog("catalog".to_string()) + .current_schema("schema".to_string()) + .build() + .into(), + ) + .await + .unwrap_err(); + assert_eq!( + common_error::status_code::StatusCode::TableNotFound, + common_error::ext::ErrorExt::status_code(&error) + ); + assert!(executor.requests.lock().unwrap().is_empty()); + } + + #[tokio::test] + async fn test_purge_table_propagates_submission_failure() { + let name = TableName::new("catalog", "schema", "metrics"); + let (operator, executor) = operator_with_tombstone(42, &name, true).await; + assert!( + operator + .purge_table( + name, + QueryContextBuilder::default() + .current_catalog("catalog".to_string()) + .current_schema("schema".to_string()) + .build() + .into() + ) + .await + .is_err() + ); + assert_eq!(1, executor.requests.lock().unwrap().len()); + } +} diff --git a/src/operator/src/statement/admin.rs b/src/operator/src/statement/admin.rs index d51e739dc2..7494977b21 100644 --- a/src/operator/src/statement/admin.rs +++ b/src/operator/src/statement/admin.rs @@ -15,7 +15,7 @@ use std::sync::Arc; use common_function::function::FunctionContext; -use common_function::function_registry::FUNCTION_REGISTRY; +use common_function::function_registry::{FUNCTION_REGISTRY, get_admin_function}; use common_query::Output; use common_recordbatch::{RecordBatch, RecordBatches}; use common_sql::convert::sql_value_to_value; @@ -49,11 +49,11 @@ impl StatementExecutor { let Admin::Func(func) = &stmt; // the function name should be in lower case. let func_name = func.name.to_string().to_lowercase(); - let factory = FUNCTION_REGISTRY.get_function(&func_name).context( - error::AdminFunctionNotFoundSnafu { + let factory = get_admin_function(&func_name) + .or_else(|| FUNCTION_REGISTRY.get_function(&func_name)) + .context(error::AdminFunctionNotFoundSnafu { name: func_name.clone(), - }, - )?; + })?; let func_ctx = FunctionContext { query_ctx: query_ctx.clone(), diff --git a/src/query/src/datafusion.rs b/src/query/src/datafusion.rs index 08a401e048..2d4f237dd6 100644 --- a/src/query/src/datafusion.rs +++ b/src/query/src/datafusion.rs @@ -1047,6 +1047,22 @@ mod tests { ); } + #[tokio::test] + async fn test_purge_table_is_not_available_to_select() { + let engine = create_test_engine().await; + let stmt = + QueryLanguageParser::parse_sql("select purge_table('numbers')", &QueryContext::arc()) + .unwrap(); + + assert!( + engine + .planner() + .plan(&stmt, QueryContext::arc()) + .await + .is_err() + ); + } + #[tokio::test] async fn test_execute() { let engine = create_test_engine().await; diff --git a/tests/cases-soft-drop/distributed/soft_drop_table.result b/tests/cases-soft-drop/distributed/soft_drop_table.result new file mode 100644 index 0000000000..e9e29caebb --- /dev/null +++ b/tests/cases-soft-drop/distributed/soft_drop_table.result @@ -0,0 +1,245 @@ +CREATE DATABASE soft_drop_lifecycle; + +Affected Rows: 1 + +USE soft_drop_lifecycle; + +Affected Rows: 0 + +CREATE TABLE lifecycle_table ( + host STRING, + reading INT, + ts TIMESTAMP TIME INDEX +); + +Affected Rows: 0 + +INSERT INTO lifecycle_table VALUES ('original', 42, 1000); + +Affected Rows: 1 + +DROP TABLE lifecycle_table; + +Affected Rows: 0 + +SELECT object_id > 0 AS has_object_id, + object_type, + original_catalog_name, + original_schema_name, + original_object_name, + purge_status, + restorable, + dropped_at IS NOT NULL AS has_dropped_at, + dropped_by, + retention_expires_at IS NOT NULL AS has_retention_expires_at, + retention_expires_at > dropped_at AS deadline_after_drop, + restored_at, + restored_by +FROM information_schema.recycle_bin +WHERE original_schema_name = 'soft_drop_lifecycle' + AND original_object_name = 'lifecycle_table'; + ++---------------+-------------+-----------------------+----------------------+----------------------+--------------+------------+----------------+------------+--------------------------+---------------------+-------------+-------------+ +| has_object_id | object_type | original_catalog_name | original_schema_name | original_object_name | purge_status | restorable | has_dropped_at | dropped_by | has_retention_expires_at | deadline_after_drop | restored_at | restored_by | ++---------------+-------------+-----------------------+----------------------+----------------------+--------------+------------+----------------+------------+--------------------------+---------------------+-------------+-------------+ +| true | TABLE | greptime | soft_drop_lifecycle | lifecycle_table | ACTIVE | true | true | | true | true | | | ++---------------+-------------+-----------------------+----------------------+----------------------+--------------+------------+----------------+------------+--------------------------+---------------------+-------------+-------------+ + +SELECT COUNT(*) AS live_table_count +FROM information_schema.tables +WHERE table_catalog = 'greptime' + AND table_schema = 'soft_drop_lifecycle' + AND table_name = 'lifecycle_table'; + ++------------------+ +| live_table_count | ++------------------+ +| 0 | ++------------------+ + +SELECT COUNT(*) AS tombstone_count +FROM information_schema.recycle_bin +WHERE original_schema_name = 'soft_drop_lifecycle' + AND original_object_name = 'lifecycle_table'; + ++-----------------+ +| tombstone_count | ++-----------------+ +| 1 | ++-----------------+ + +UNDROP TABLE greptime.soft_drop_lifecycle.lifecycle_table; + +Affected Rows: 0 + +SELECT host, reading, ts FROM lifecycle_table; + ++----------+---------+---------------------+ +| host | reading | ts | ++----------+---------+---------------------+ +| original | 42 | 1970-01-01T00:00:01 | ++----------+---------+---------------------+ + +DROP TABLE lifecycle_table; + +Affected Rows: 0 + +ADMIN purge_table('lifecycle_table'); + ++--------------------------------------+ +| ADMIN purge_table('lifecycle_table') | ++--------------------------------------+ +| 0 | ++--------------------------------------+ + +SELECT COUNT(*) AS tombstone_count +FROM information_schema.recycle_bin +WHERE original_schema_name = 'soft_drop_lifecycle' + AND original_object_name = 'lifecycle_table'; + ++-----------------+ +| tombstone_count | ++-----------------+ +| 0 | ++-----------------+ + +UNDROP TABLE greptime.soft_drop_lifecycle.lifecycle_table; + +Error: 4001(TableNotFound), Table not found: greptime.soft_drop_lifecycle.lifecycle_table + +CREATE TABLE same_name ( + generation STRING, + reading INT, + ts TIMESTAMP TIME INDEX +); + +Affected Rows: 0 + +INSERT INTO same_name VALUES ('old', 1, 2000); + +Affected Rows: 1 + +DROP TABLE same_name; + +Affected Rows: 0 + +CREATE TABLE same_name ( + generation STRING, + reading INT, + ts TIMESTAMP TIME INDEX +); + +Affected Rows: 0 + +INSERT INTO same_name VALUES ('new', 2, 3000); + +Affected Rows: 1 + +SELECT object_type, + original_catalog_name, + original_schema_name, + original_object_name, + purge_status, + restorable +FROM information_schema.recycle_bin +WHERE original_schema_name = 'soft_drop_lifecycle' + AND original_object_name = 'same_name'; + ++-------------+-----------------------+----------------------+----------------------+--------------+------------+ +| object_type | original_catalog_name | original_schema_name | original_object_name | purge_status | restorable | ++-------------+-----------------------+----------------------+----------------------+--------------+------------+ +| TABLE | greptime | soft_drop_lifecycle | same_name | ACTIVE | true | ++-------------+-----------------------+----------------------+----------------------+--------------+------------+ + +UNDROP TABLE soft_drop_lifecycle.same_name; + +Error: 4000(TableAlreadyExists), Table already exists, table: greptime.soft_drop_lifecycle.same_name + +DROP TABLE same_name; + +Error: 4000(TableAlreadyExists), Cannot drop table 'greptime.soft_drop_lifecycle.same_name': an older tombstone already uses the same full name + +ADMIN purge_table('greptime.soft_drop_lifecycle.same_name'); + ++-------------------------------------------------------------+ +| ADMIN purge_table('greptime.soft_drop_lifecycle.same_name') | ++-------------------------------------------------------------+ +| 0 | ++-------------------------------------------------------------+ + +SELECT COUNT(*) AS tombstone_count +FROM information_schema.recycle_bin +WHERE original_schema_name = 'soft_drop_lifecycle' + AND original_object_name = 'same_name'; + ++-----------------+ +| tombstone_count | ++-----------------+ +| 0 | ++-----------------+ + +SELECT generation, reading, ts FROM same_name; + ++------------+---------+---------------------+ +| generation | reading | ts | ++------------+---------+---------------------+ +| new | 2 | 1970-01-01T00:00:03 | ++------------+---------+---------------------+ + +DROP TABLE same_name; + +Affected Rows: 0 + +SELECT COUNT(*) AS tombstone_count +FROM information_schema.recycle_bin +WHERE original_schema_name = 'soft_drop_lifecycle' + AND original_object_name = 'same_name'; + ++-----------------+ +| tombstone_count | ++-----------------+ +| 1 | ++-----------------+ + +UNDROP TABLE same_name; + +Affected Rows: 0 + +SELECT generation, reading, ts FROM same_name; + ++------------+---------+---------------------+ +| generation | reading | ts | ++------------+---------+---------------------+ +| new | 2 | 1970-01-01T00:00:03 | ++------------+---------+---------------------+ + +DROP TABLE same_name; + +Affected Rows: 0 + +ADMIN purge_table('soft_drop_lifecycle.same_name'); + ++----------------------------------------------------+ +| ADMIN purge_table('soft_drop_lifecycle.same_name') | ++----------------------------------------------------+ +| 0 | ++----------------------------------------------------+ + +SELECT COUNT(*) AS tombstone_count +FROM information_schema.recycle_bin +WHERE original_schema_name = 'soft_drop_lifecycle'; + ++-----------------+ +| tombstone_count | ++-----------------+ +| 0 | ++-----------------+ + +USE public; + +Affected Rows: 0 + +DROP DATABASE soft_drop_lifecycle; + +Affected Rows: 0 + diff --git a/tests/cases-soft-drop/distributed/soft_drop_table.sql b/tests/cases-soft-drop/distributed/soft_drop_table.sql new file mode 100644 index 0000000000..2b3ca86176 --- /dev/null +++ b/tests/cases-soft-drop/distributed/soft_drop_table.sql @@ -0,0 +1,120 @@ +CREATE DATABASE soft_drop_lifecycle; + +USE soft_drop_lifecycle; + +CREATE TABLE lifecycle_table ( + host STRING, + reading INT, + ts TIMESTAMP TIME INDEX +); + +INSERT INTO lifecycle_table VALUES ('original', 42, 1000); + +DROP TABLE lifecycle_table; + +SELECT object_id > 0 AS has_object_id, + object_type, + original_catalog_name, + original_schema_name, + original_object_name, + purge_status, + restorable, + dropped_at IS NOT NULL AS has_dropped_at, + dropped_by, + retention_expires_at IS NOT NULL AS has_retention_expires_at, + retention_expires_at > dropped_at AS deadline_after_drop, + restored_at, + restored_by +FROM information_schema.recycle_bin +WHERE original_schema_name = 'soft_drop_lifecycle' + AND original_object_name = 'lifecycle_table'; + +SELECT COUNT(*) AS live_table_count +FROM information_schema.tables +WHERE table_catalog = 'greptime' + AND table_schema = 'soft_drop_lifecycle' + AND table_name = 'lifecycle_table'; + +SELECT COUNT(*) AS tombstone_count +FROM information_schema.recycle_bin +WHERE original_schema_name = 'soft_drop_lifecycle' + AND original_object_name = 'lifecycle_table'; + +UNDROP TABLE greptime.soft_drop_lifecycle.lifecycle_table; + +SELECT host, reading, ts FROM lifecycle_table; + +DROP TABLE lifecycle_table; + +ADMIN purge_table('lifecycle_table'); + +SELECT COUNT(*) AS tombstone_count +FROM information_schema.recycle_bin +WHERE original_schema_name = 'soft_drop_lifecycle' + AND original_object_name = 'lifecycle_table'; + +UNDROP TABLE greptime.soft_drop_lifecycle.lifecycle_table; + +CREATE TABLE same_name ( + generation STRING, + reading INT, + ts TIMESTAMP TIME INDEX +); + +INSERT INTO same_name VALUES ('old', 1, 2000); + +DROP TABLE same_name; + +CREATE TABLE same_name ( + generation STRING, + reading INT, + ts TIMESTAMP TIME INDEX +); + +INSERT INTO same_name VALUES ('new', 2, 3000); + +SELECT object_type, + original_catalog_name, + original_schema_name, + original_object_name, + purge_status, + restorable +FROM information_schema.recycle_bin +WHERE original_schema_name = 'soft_drop_lifecycle' + AND original_object_name = 'same_name'; + +UNDROP TABLE soft_drop_lifecycle.same_name; + +DROP TABLE same_name; + +ADMIN purge_table('greptime.soft_drop_lifecycle.same_name'); + +SELECT COUNT(*) AS tombstone_count +FROM information_schema.recycle_bin +WHERE original_schema_name = 'soft_drop_lifecycle' + AND original_object_name = 'same_name'; + +SELECT generation, reading, ts FROM same_name; + +DROP TABLE same_name; + +SELECT COUNT(*) AS tombstone_count +FROM information_schema.recycle_bin +WHERE original_schema_name = 'soft_drop_lifecycle' + AND original_object_name = 'same_name'; + +UNDROP TABLE same_name; + +SELECT generation, reading, ts FROM same_name; + +DROP TABLE same_name; + +ADMIN purge_table('soft_drop_lifecycle.same_name'); + +SELECT COUNT(*) AS tombstone_count +FROM information_schema.recycle_bin +WHERE original_schema_name = 'soft_drop_lifecycle'; + +USE public; + +DROP DATABASE soft_drop_lifecycle; diff --git a/tests/conf/datanode-test.toml.template b/tests/conf/datanode-test.toml.template index e68a76cc9a..63d3be2f8c 100644 --- a/tests/conf/datanode-test.toml.template +++ b/tests/conf/datanode-test.toml.template @@ -8,6 +8,10 @@ rpc_runtime_size = 8 {{ if enable_flat_format }} default_flat_format = true {{ endif }} +{{ if enable_gc }} +[region_engine.mito.gc] +enable = true +{{ endif }} [wal] {{ if is_raft_engine }} diff --git a/tests/conf/metasrv-test.toml.template b/tests/conf/metasrv-test.toml.template index cf2c0a20a6..0115a911fa 100644 --- a/tests/conf/metasrv-test.toml.template +++ b/tests/conf/metasrv-test.toml.template @@ -1,5 +1,11 @@ flush_stats_factor = 1 heartbeat_interval = "1s" +{{ if enable_gc }} +[gc] +enable = true +[gc.experimental_soft_drop] +enable = true +{{ endif }} {{ if use_etcd }} ## Store server address default to etcd store. store_addrs = [{store_addrs | unescaped}] diff --git a/tests/runner/src/cmd/bare.rs b/tests/runner/src/cmd/bare.rs index f845d7b2c0..24ba756cee 100644 --- a/tests/runner/src/cmd/bare.rs +++ b/tests/runner/src/cmd/bare.rs @@ -106,6 +106,10 @@ pub struct BareCommand { /// Enable flat format for storage engine (sets default_flat_format = true). #[clap(long, default_value = "false")] enable_flat_format: bool, + + /// Enable garbage collection in metasrv and datanodes. + #[clap(long, default_value = "false")] + enable_gc: bool, } impl BareCommand { @@ -177,6 +181,7 @@ impl BareCommand { setup_pg: self.setup_pg, setup_mysql: self.setup_mysql, enable_flat_format: self.enable_flat_format, + enable_gc: self.enable_gc, }; let runner = Runner::new( diff --git a/tests/runner/src/cmd/compat.rs b/tests/runner/src/cmd/compat.rs index b73f7b23e3..241d7b9eee 100644 --- a/tests/runner/src/cmd/compat.rs +++ b/tests/runner/src/cmd/compat.rs @@ -345,6 +345,7 @@ impl CompatCommand { setup_pg: None, setup_mysql: None, enable_flat_format: false, + enable_gc: false, }; let env = Env::new( diff --git a/tests/runner/src/env/bare.rs b/tests/runner/src/env/bare.rs index 8c43d2416b..7930bec243 100644 --- a/tests/runner/src/env/bare.rs +++ b/tests/runner/src/env/bare.rs @@ -81,6 +81,7 @@ pub struct StoreConfig { pub(crate) setup_pg: Option, pub(crate) setup_mysql: Option, pub enable_flat_format: bool, + pub enable_gc: bool, } #[derive(Clone)] diff --git a/tests/runner/src/server_mode.rs b/tests/runner/src/server_mode.rs index d88a71792a..1f77f5d137 100644 --- a/tests/runner/src/server_mode.rs +++ b/tests/runner/src/server_mode.rs @@ -150,6 +150,8 @@ struct ConfigContext { addrs: HashMap, // enable flat format for storage engine enable_flat_format: bool, + // enable garbage collection in metasrv and datanodes + enable_gc: bool, } impl ServerMode { @@ -357,6 +359,7 @@ impl ServerMode { instance_id: id, addrs, enable_flat_format: db_ctx.store_config().enable_flat_format, + enable_gc: db_ctx.store_config().enable_gc, }; let rendered = tt.render(self.name(), &ctx).unwrap(); @@ -579,6 +582,7 @@ mod tests { setup_pg: None, setup_mysql: None, enable_flat_format: false, + enable_gc: false, }; let env = Env::new( sqlness_home.to_path_buf(), @@ -727,4 +731,41 @@ mod tests { assert_eq!(GrpcArgStyle::for_version(Some(&v1_1_0)), GrpcArgStyle::Grpc); assert_eq!(GrpcArgStyle::for_version(Some(&v1_2_0)), GrpcArgStyle::Grpc); } + + #[test] + fn test_generate_distributed_gc_config_when_enabled() { + let temp_dir = tempfile::tempdir().unwrap(); + let store_config = StoreConfig { + store_addrs: vec![], + setup_etcd: false, + setup_pg: None, + setup_mysql: None, + enable_flat_format: false, + enable_gc: true, + }; + let db_ctx = GreptimeDBContext::new(WalConfig::RaftEngine, store_config); + let metasrv = ServerMode::Metasrv { + rpc_bind_addr: "127.0.0.1:4201".to_string(), + rpc_server_addr: "127.0.0.1:4201".to_string(), + http_addr: "127.0.0.1:4200".to_string(), + }; + let datanode = ServerMode::Datanode { + rpc_bind_addr: "127.0.0.1:4301".to_string(), + rpc_server_addr: "127.0.0.1:4301".to_string(), + http_addr: "127.0.0.1:4300".to_string(), + metasrv_addr: "127.0.0.1:4201".to_string(), + node_id: 0, + }; + + let metasrv_config = + std::fs::read_to_string(metasrv.generate_config_file(temp_dir.path(), &db_ctx, 0)) + .unwrap(); + let datanode_config = + std::fs::read_to_string(datanode.generate_config_file(temp_dir.path(), &db_ctx, 0)) + .unwrap(); + + assert!(metasrv_config.contains("[gc]\nenable = true")); + assert!(metasrv_config.contains("[gc.experimental_soft_drop]\nenable = true")); + assert!(datanode_config.contains("[region_engine.mito.gc]\nenable = true")); + } }