From 78084a9d4485ee61ff03e071f6428fc57c906332 Mon Sep 17 00:00:00 2001 From: Yingwen Date: Mon, 10 Aug 2026 20:19:27 +0800 Subject: [PATCH] feat: add admin function to discard unflushed data (#8768) * feat: add admin function to discard unflushed data Signed-off-by: evenyag * test: cover discarding unflushed data by table Signed-off-by: evenyag * chore: fix license header Signed-off-by: evenyag * fix: reject discarding logical metric table data Signed-off-by: evenyag * refactor: defer table name formatting in error paths Signed-off-by: evenyag * chore(deps): update greptime-proto revision Signed-off-by: evenyag * refactor: rename discard unflushed admin function Signed-off-by: evenyag --------- Signed-off-by: evenyag --- AGENTS.md | 21 +- Cargo.lock | 2 +- Cargo.toml | 2 +- src/common/function/src/admin.rs | 4 +- .../src/admin/discard_unflushed_data.rs | 232 ++++++++++++++++++ src/common/function/src/handlers.rs | 14 ++ src/common/function/src/state.rs | 17 ++ src/operator/src/request.rs | 96 +++++++- src/operator/src/table.rs | 25 ++ src/store-api/src/region_request.rs | 39 +++ tests-integration/src/tests/instance_test.rs | 75 ++++++ .../admin/discard_unflushed_data.result | 133 ++++++++++ .../function/admin/discard_unflushed_data.sql | 62 +++++ 13 files changed, 709 insertions(+), 13 deletions(-) create mode 100644 src/common/function/src/admin/discard_unflushed_data.rs create mode 100644 tests/cases/standalone/common/function/admin/discard_unflushed_data.result create mode 100644 tests/cases/standalone/common/function/admin/discard_unflushed_data.sql diff --git a/AGENTS.md b/AGENTS.md index c99856c294..70f2ae20b4 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -114,22 +114,25 @@ blast radius requires it. ## Before opening a PR -1. `make fmt` -2. `make clippy` -3. `make test` -4. `make check-udeps` (run `make fix-udeps` if it reports unused dependencies). -5. If you added or changed a public configuration option, update the applicable +1. If you added a `.rs`, `.py`, or `.ts` file, apply and verify its license + header with `hawkeye format` followed by `hawkeye check`. Use the inception + year from `licenserc.toml`, not the current year. +2. `make fmt` +3. `make clippy` +4. `make test` +5. `make check-udeps` (run `make fix-udeps` if it reports unused dependencies). +6. If you added or changed a public configuration option, update the applicable example TOMLs, configuration-loading and serialized-config snapshot tests, and related user-facing documentation. Run `make config-docs` (needs Docker) and commit the regenerated `config/config.md`. -6. If you changed a persisted or wire format, add a compatibility test case (see +7. If you changed a persisted or wire format, add a compatibility test case (see `.agents/architecture-invariants.md`). -7. If you added or gated an enterprise-only file, give it the enterprise license +8. If you added or gated an enterprise-only file, give it the enterprise license header, list it in `licenserc-enterprise.toml` (`includes`) and `licenserc.toml` (`excludes`), and run `make check-enterprise-license`. -8. Use a conventional-commit title, sign off commits (`git commit -s`), and sign +9. Use a conventional-commit title, sign off commits (`git commit -s`), and sign the CLA. -9. When creating or updating a pull request, follow +10. When creating or updating a pull request, follow [`.github/pull_request_template.md`](.github/pull_request_template.md): include the CLA statement, fill the change-intention section with enough detail, and update checklist items accurately. diff --git a/Cargo.lock b/Cargo.lock index e2210b80bf..d6385dd765 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6032,7 +6032,7 @@ dependencies = [ [[package]] name = "greptime-proto" version = "0.1.0" -source = "git+https://github.com/GreptimeTeam/greptime-proto.git?rev=032510ded061277b7d1bbb29d4046abb5b1cbf4b#032510ded061277b7d1bbb29d4046abb5b1cbf4b" +source = "git+https://github.com/GreptimeTeam/greptime-proto.git?rev=e7be20ff855b7522efe235383d82238d56e676c3#e7be20ff855b7522efe235383d82238d56e676c3" dependencies = [ "prost 0.14.1", "prost-types 0.14.1", diff --git a/Cargo.toml b/Cargo.toml index 502e652ffc..64e6dba2f7 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -159,7 +159,7 @@ fs2 = "0.4" fst = "0.4.7" futures = "0.3" futures-util = "0.3" -greptime-proto = { git = "https://github.com/GreptimeTeam/greptime-proto.git", rev = "032510ded061277b7d1bbb29d4046abb5b1cbf4b" } +greptime-proto = { git = "https://github.com/GreptimeTeam/greptime-proto.git", rev = "e7be20ff855b7522efe235383d82238d56e676c3" } hex = "0.4" http = "1" humantime = "2.1" diff --git a/src/common/function/src/admin.rs b/src/common/function/src/admin.rs index 73bed86f7a..ff8172ebb5 100644 --- a/src/common/function/src/admin.rs +++ b/src/common/function/src/admin.rs @@ -13,6 +13,7 @@ // limitations under the License. mod build_index_table; +mod discard_unflushed_data; mod flush_compact_region; mod flush_compact_table; mod gc; @@ -34,6 +35,7 @@ use reconcile_database::ReconcileDatabaseFunction; use reconcile_table::ReconcileTableFunction; use crate::admin::build_index_table::BuildIndexFunction; +use crate::admin::discard_unflushed_data::DiscardUnflushedDataFunction; use crate::flush_flow::FlushFlowFunction; use crate::function_registry::FunctionRegistry; @@ -58,8 +60,8 @@ impl AdminFunction { } /// Register functions that must only be resolved by an ADMIN statement. - #[cfg_attr(not(feature = "enterprise"), allow(unused_variables))] pub fn register_admin_only(registry: &FunctionRegistry) { + registry.register(DiscardUnflushedDataFunction::factory()); #[cfg(feature = "enterprise")] registry.register(PurgeTableFunction::factory()); } diff --git a/src/common/function/src/admin/discard_unflushed_data.rs b/src/common/function/src/admin/discard_unflushed_data.rs new file mode 100644 index 0000000000..b9f46f62f9 --- /dev/null +++ b/src/common/function/src/admin/discard_unflushed_data.rs @@ -0,0 +1,232 @@ +// 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, MissingTableMutationHandlerSnafu, Result, TableMutationSnafu, + UnsupportedInputDataTypeSnafu, +}; +use datafusion_expr::{Signature, TypeSignature, Volatility}; +use datatypes::data_type::DataType; +use datatypes::prelude::*; +use session::context::QueryContextRef; +use session::table_name::table_name_to_full_name; +use snafu::{ResultExt, ensure}; +use store_api::storage::RegionId; +use table::table_name::TableName; + +use crate::handlers::TableMutationHandlerRef; +use crate::helper::cast_u64; + +/// Discards all unflushed data from a region. +#[admin_fn( + name = DiscardUnflushedDataFunction, + display_name = discard_unflushed, + sig_fn = signature, + ret = uint64, + single_row +)] +pub(crate) async fn discard_unflushed_data( + table_mutation_handler: &TableMutationHandlerRef, + 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 affected_rows = match params[0] { + ValueRef::String(table_name) => { + let (catalog_name, schema_name, table_name) = + table_name_to_full_name(table_name, query_ctx) + .map_err(BoxedError::new) + .context(TableMutationSnafu)?; + table_mutation_handler + .discard_unflushed_data_by_table( + TableName::new(catalog_name, schema_name, table_name), + query_ctx.clone(), + ) + .await? + } + _ => { + let Some(region_id) = cast_u64(¶ms[0])? else { + return UnsupportedInputDataTypeSnafu { + function: "discard_unflushed", + datatypes: params + .iter() + .map(|value| value.data_type()) + .collect::>(), + } + .fail(); + }; + table_mutation_handler + .discard_unflushed_data(RegionId::from_u64(region_id), query_ctx.clone()) + .await? + } + }; + + Ok(Value::from(affected_rows as u64)) +} + +fn signature() -> Signature { + Signature::one_of( + vec![ + TypeSignature::Uniform( + 1, + ConcreteDataType::numerics() + .into_iter() + .map(|data_type| data_type.as_arrow_type()) + .collect(), + ), + TypeSignature::Exact(vec![ArrowDataType::Utf8]), + ], + Volatility::Immutable, + ) +} + +#[cfg(test)] +mod tests { + use std::sync::Arc; + + use arrow::array::{StringArray, UInt64Array}; + use arrow::datatypes::{DataType, Field}; + use datafusion_expr::{ColumnarValue, TypeSignature}; + + use super::*; + use crate::function::FunctionContext; + use crate::function_factory::ScalarFunctionFactory; + use crate::function_registry::{FUNCTION_REGISTRY, get_admin_function}; + + #[test] + fn test_discard_unflushed_data_is_admin_only() { + assert!(get_admin_function("discard_unflushed").is_some()); + assert!( + FUNCTION_REGISTRY + .get_function("discard_unflushed") + .is_none() + ); + } + + #[test] + fn test_discard_unflushed_data_signature() { + let factory: ScalarFunctionFactory = DiscardUnflushedDataFunction::factory().into(); + let function = factory.provide(FunctionContext::mock()); + + assert_eq!("discard_unflushed", function.name()); + assert_eq!(DataType::UInt64, function.return_type(&[]).unwrap()); + assert!(matches!( + function.signature(), + Signature { + type_signature: TypeSignature::OneOf(valid_types), + volatility: Volatility::Immutable, + .. + } if valid_types == &vec![ + TypeSignature::Uniform( + 1, + ConcreteDataType::numerics() + .into_iter() + .map(|data_type| { + use datatypes::data_type::DataType; + data_type.as_arrow_type() + }) + .collect::>(), + ), + TypeSignature::Exact(vec![DataType::Utf8]), + ] + )); + } + + #[tokio::test] + async fn test_discard_unflushed_data() { + let factory: ScalarFunctionFactory = DiscardUnflushedDataFunction::factory().into(); + let function = factory.provide(FunctionContext::mock()); + let args = datafusion::logical_expr::ScalarFunctionArgs { + args: vec![ColumnarValue::Array(Arc::new(UInt64Array::from(vec![99])))], + arg_fields: vec![Arc::new(Field::new("arg_0", DataType::UInt64, false))], + return_field: Arc::new(Field::new("result", DataType::UInt64, false)), + number_rows: 1, + config_options: Arc::new(datafusion_common::config::ConfigOptions::default()), + }; + + let result = function + .as_async() + .unwrap() + .invoke_async_with_args(args) + .await + .unwrap(); + let ColumnarValue::Array(array) = result else { + panic!("expected array output"); + }; + let array = array.as_any().downcast_ref::().unwrap(); + assert_eq!(42, array.value(0)); + } + + #[tokio::test] + async fn test_discard_unflushed_data_by_table() { + let factory: ScalarFunctionFactory = DiscardUnflushedDataFunction::factory().into(); + let function = factory.provide(FunctionContext::mock()); + let args = datafusion::logical_expr::ScalarFunctionArgs { + args: vec![ColumnarValue::Array(Arc::new(StringArray::from(vec![ + "my_table", + ])))], + arg_fields: vec![Arc::new(Field::new("arg_0", DataType::Utf8, false))], + return_field: Arc::new(Field::new("result", DataType::UInt64, false)), + number_rows: 1, + config_options: Arc::new(datafusion_common::config::ConfigOptions::default()), + }; + + let result = function + .as_async() + .unwrap() + .invoke_async_with_args(args) + .await + .unwrap(); + let ColumnarValue::Array(array) = result else { + panic!("expected array output"); + }; + let array = array.as_any().downcast_ref::().unwrap(); + assert_eq!(42, array.value(0)); + } + + #[tokio::test] + async fn test_discard_unflushed_data_rejects_multiple_rows() { + let factory: ScalarFunctionFactory = DiscardUnflushedDataFunction::factory().into(); + let function = factory.provide(FunctionContext::mock()); + let args = datafusion::logical_expr::ScalarFunctionArgs { + args: vec![ColumnarValue::Array(Arc::new(UInt64Array::from(vec![ + 1, 2, + ])))], + arg_fields: vec![Arc::new(Field::new("arg_0", DataType::UInt64, false))], + return_field: Arc::new(Field::new("result", DataType::UInt64, false)), + number_rows: 2, + config_options: Arc::new(datafusion_common::config::ConfigOptions::default()), + }; + + let error = function + .as_async() + .unwrap() + .invoke_async_with_args(args) + .await + .unwrap_err(); + assert!(error.to_string().contains("received 2")); + } +} diff --git a/src/common/function/src/handlers.rs b/src/common/function/src/handlers.rs index 47de2824b8..24ce115a0d 100644 --- a/src/common/function/src/handlers.rs +++ b/src/common/function/src/handlers.rs @@ -69,6 +69,20 @@ pub trait TableMutationHandler: Send + Sync { region_id: RegionId, ctx: QueryContextRef, ) -> Result; + + /// Discard all unflushed data from a table region. + async fn discard_unflushed_data( + &self, + region_id: RegionId, + ctx: QueryContextRef, + ) -> Result; + + /// Discard all unflushed data from all regions of a table. + async fn discard_unflushed_data_by_table( + &self, + table_name: TableName, + ctx: QueryContextRef, + ) -> Result; } /// A trait for handling procedure service requests in `QueryEngine`. diff --git a/src/common/function/src/state.rs b/src/common/function/src/state.rs index 726785f074..b1802b6257 100644 --- a/src/common/function/src/state.rs +++ b/src/common/function/src/state.rs @@ -48,6 +48,7 @@ impl FunctionState { BuildIndexTableRequest, CompactTableRequest, DeleteRequest, FlushTableRequest, InsertRequest, }; + use table::table_name::TableName; use crate::handlers::{FlowServiceHandler, ProcedureServiceHandler, TableMutationHandler}; struct MockProcedureServiceHandler; @@ -171,6 +172,22 @@ impl FunctionState { ) -> Result { Ok(ROWS) } + + async fn discard_unflushed_data( + &self, + _region_id: RegionId, + _ctx: QueryContextRef, + ) -> Result { + Ok(ROWS) + } + + async fn discard_unflushed_data_by_table( + &self, + _table_name: TableName, + _ctx: QueryContextRef, + ) -> Result { + Ok(ROWS) + } } #[async_trait] diff --git a/src/operator/src/request.rs b/src/operator/src/request.rs index aea395d8ee..b46a6b657a 100644 --- a/src/operator/src/request.rs +++ b/src/operator/src/request.rs @@ -18,9 +18,11 @@ use api::helper::to_pb_time_unit; use api::v1::region::region_request::Body as RegionRequestBody; use api::v1::region::{ BuildIndexRequest, CompactRequest, CompactionTimeRange, FlushRequest, RegionRequestHeader, + TruncateRequest, Unflushed, truncate_request, }; use catalog::CatalogManagerRef; use common_catalog::build_db_string; +use common_catalog::consts::METRIC_ENGINE; use common_meta::node_manager::{AffectedRows, NodeManagerRef}; use common_meta::peer::Peer; use common_telemetry::tracing_context::TracingContext; @@ -34,10 +36,12 @@ use session::context::QueryContextRef; use snafu::prelude::*; use store_api::storage::RegionId; use table::requests::{BuildIndexTableRequest, CompactTableRequest, FlushTableRequest}; +use table::table_name::TableName; use crate::error::{ CatalogSnafu, FindRegionLeaderSnafu, FindTablePartitionRuleSnafu, JoinTaskSnafu, - RequestRegionSnafu, Result, TableNotFoundSnafu, UnsupportedRegionRequestSnafu, + NotSupportedSnafu, RequestRegionSnafu, Result, TableNotFoundSnafu, + UnsupportedRegionRequestSnafu, }; use crate::region_req_factory::RegionRequestFactory; @@ -206,6 +210,76 @@ impl Requester { info!("Handle region manual compaction request: {region_id}"); self.do_request(vec![request], None, &ctx).await } + + /// Discard all unflushed data from the region. + pub async fn handle_discard_unflushed_data( + &self, + region_id: RegionId, + ctx: QueryContextRef, + ) -> Result { + let request = RegionRequestBody::Truncate(TruncateRequest { + region_id: region_id.into(), + kind: Some(truncate_request::Kind::Unflushed(Unflushed {})), + }); + + info!("Handle region discard unflushed data request: {region_id}"); + self.do_request(vec![request], None, &ctx).await + } + + /// Discard all unflushed data from all regions of the table. + pub async fn handle_discard_unflushed_data_by_table( + &self, + table_name: TableName, + ctx: QueryContextRef, + ) -> Result { + let table = self + .catalog_manager + .table( + &table_name.catalog_name, + &table_name.schema_name, + &table_name.table_name, + None, + ) + .await + .context(CatalogSnafu)?; + let table = table.with_context(|| TableNotFoundSnafu { + table_name: table_name.to_string(), + })?; + let table_info = table.table_info(); + ensure_discard_unflushed_supported( + &table_info.meta.engine, + table_info.is_physical_table(), + )?; + + let partitions = &self + .partition_manager + .find_physical_partition_info(table_info.ident.table_id) + .await + .with_context(|_| FindTablePartitionRuleSnafu { + table_name: table_name.to_string(), + })? + .partitions; + let requests = partitions + .iter() + .map(|partition| { + RegionRequestBody::Truncate(TruncateRequest { + region_id: partition.id.into(), + kind: Some(truncate_request::Kind::Unflushed(Unflushed {})), + }) + }) + .collect(); + + info!("Handle table discard unflushed data request: {table_name}"); + self.do_request( + requests, + Some(build_db_string( + &table_name.catalog_name, + &table_name.schema_name, + )), + &ctx, + ) + .await + } } fn to_pb_compaction_time_range(range: TimestampRange) -> Result { @@ -291,6 +365,7 @@ impl Requester { RegionRequestBody::Flush(req) => req.region_id, RegionRequestBody::Compact(req) => req.region_id, RegionRequestBody::BuildIndex(req) => req.region_id, + RegionRequestBody::Truncate(req) => req.region_id, _ => { error!("Unsupported region request: {:?}", req); return UnsupportedRegionRequestSnafu {}.fail(); @@ -329,6 +404,16 @@ impl Requester { } } +fn ensure_discard_unflushed_supported(engine: &str, is_physical_table: bool) -> Result<()> { + ensure!( + engine != METRIC_ENGINE || is_physical_table, + NotSupportedSnafu { + feat: "discarding unflushed data from a Metric Engine logical table" + } + ); + Ok(()) +} + #[cfg(test)] mod tests { use api::v1::TimeUnit; @@ -351,6 +436,15 @@ mod tests { assert_eq!(TimeUnit::Second as i32, pb_range.time_unit); } + #[test] + fn test_discard_unflushed_rejects_metric_logical_table() { + let error = ensure_discard_unflushed_supported(METRIC_ENGINE, false).unwrap_err(); + assert!(matches!(error, crate::error::Error::NotSupported { .. })); + + ensure_discard_unflushed_supported(METRIC_ENGINE, true).unwrap(); + ensure_discard_unflushed_supported("mito", false).unwrap(); + } + #[test] fn test_to_pb_compaction_time_range() { let range = TimestampRange::new( diff --git a/src/operator/src/table.rs b/src/operator/src/table.rs index 13ed57200c..117351280b 100644 --- a/src/operator/src/table.rs +++ b/src/operator/src/table.rs @@ -26,6 +26,7 @@ use table::requests::{ BuildIndexTableRequest, CompactTableRequest, DeleteRequest as TableDeleteRequest, FlushTableRequest, InsertRequest as TableInsertRequest, }; +use table::table_name::TableName; use crate::delete::DeleterRef; use crate::insert::InserterRef; @@ -132,4 +133,28 @@ impl TableMutationHandler for TableMutationOperator { .map_err(BoxedError::new) .context(query_error::TableMutationSnafu) } + + async fn discard_unflushed_data( + &self, + region_id: RegionId, + ctx: QueryContextRef, + ) -> QueryResult { + self.requester + .handle_discard_unflushed_data(region_id, ctx) + .await + .map_err(BoxedError::new) + .context(query_error::TableMutationSnafu) + } + + async fn discard_unflushed_data_by_table( + &self, + table_name: TableName, + ctx: QueryContextRef, + ) -> QueryResult { + self.requester + .handle_discard_unflushed_data_by_table(table_name, ctx) + .await + .map_err(BoxedError::new) + .context(query_error::TableMutationSnafu) + } } diff --git a/src/store-api/src/region_request.rs b/src/store-api/src/region_request.rs index 490e6c4d6e..3bb2bb916c 100644 --- a/src/store-api/src/region_request.rs +++ b/src/store-api/src/region_request.rs @@ -458,6 +458,10 @@ fn make_region_truncate(truncate: TruncateRequest) -> Result Ok(vec![( + region_id, + RegionRequest::Truncate(RegionTruncateRequest::Unflushed), + )]), } } @@ -1811,6 +1815,41 @@ mod tests { ); } + #[test] + fn test_make_region_truncate_unflushed() { + let region_id = RegionId::new(42, 3); + let requests = + RegionRequest::try_from_request_body(region_request::Body::Truncate(TruncateRequest { + region_id: region_id.as_u64(), + kind: Some(truncate_request::Kind::Unflushed( + api::v1::region::Unflushed {}, + )), + })) + .unwrap(); + + assert_eq!(region_id, requests[0].0); + assert!(matches!( + requests[0].1, + RegionRequest::Truncate(RegionTruncateRequest::Unflushed) + )); + } + + #[test] + fn test_make_region_truncate_requires_kind() { + let error = + RegionRequest::try_from_request_body(region_request::Body::Truncate(TruncateRequest { + region_id: RegionId::new(42, 3).as_u64(), + kind: None, + })) + .unwrap_err(); + + assert!( + error + .to_string() + .contains("missing kind in TruncateRequest") + ); + } + #[test] fn test_from_proto_location() { let proto_location = v1::AddColumnLocation { diff --git a/tests-integration/src/tests/instance_test.rs b/tests-integration/src/tests/instance_test.rs index 9c64e84125..f21b459c2a 100644 --- a/tests-integration/src/tests/instance_test.rs +++ b/tests-integration/src/tests/instance_test.rs @@ -90,6 +90,81 @@ async fn test_create_database_and_insert_query(instance: Arc) } } +#[apply(both_instances_cases)] +async fn test_admin_discard_unflushed_data(instance: Arc) { + let instance = instance.frontend(); + + execute_sql( + &instance, + r#"CREATE TABLE discard_unflushed_data_test ( + host STRING PRIMARY KEY, + val DOUBLE, + ts TIMESTAMP TIME INDEX + ) ENGINE = mito"#, + ) + .await; + execute_sql( + &instance, + "INSERT INTO discard_unflushed_data_test VALUES ('persisted', 1, 1)", + ) + .await; + execute_sql( + &instance, + "ADMIN FLUSH_TABLE('discard_unflushed_data_test')", + ) + .await; + execute_sql( + &instance, + "INSERT INTO discard_unflushed_data_test VALUES ('unflushed', 2, 2)", + ) + .await; + + let region_output = execute_sql( + &instance, + "SELECT greptime_partition_id FROM information_schema.partitions \ + WHERE table_name = 'discard_unflushed_data_test' LIMIT 1", + ) + .await; + let OutputData::Stream(stream) = region_output.data else { + panic!("expected region id stream"); + }; + let batches = util::collect(stream).await.unwrap(); + let region_ids = batches[0].column(0); + let region_ids = region_ids.as_any().downcast_ref::().unwrap(); + let region_id = region_ids.value(0); + + let admin_sql = format!("ADMIN discard_unflushed({region_id})"); + for _ in 0..2 { + let output = execute_sql(&instance, &admin_sql).await; + let OutputData::RecordBatches(batches) = output.data else { + panic!("expected ADMIN result batches"); + }; + let result = batches.iter().next().unwrap().column(0); + let result = result.as_any().downcast_ref::().unwrap(); + assert_eq!(0, result.value(0)); + } + + let output = execute_sql( + &instance, + "SELECT host, val FROM discard_unflushed_data_test ORDER BY host", + ) + .await; + assert_eq!( + "+-----------+-----+\n\ + | host | val |\n\ + +-----------+-----+\n\ + | persisted | 1.0 |\n\ + +-----------+-----+", + output.data.pretty_print().await + ); + + assert!( + try_execute_sql(&instance, &format!("SELECT discard_unflushed({region_id})"),) + .await + .is_err() + ); +} + #[tokio::test(flavor = "multi_thread")] async fn test_distributed_scalar_latest_with_query_parallelism_below_regions() { common_telemetry::init_default_ut_logging(); diff --git a/tests/cases/standalone/common/function/admin/discard_unflushed_data.result b/tests/cases/standalone/common/function/admin/discard_unflushed_data.result new file mode 100644 index 0000000000..9723baa863 --- /dev/null +++ b/tests/cases/standalone/common/function/admin/discard_unflushed_data.result @@ -0,0 +1,133 @@ +CREATE TABLE discard_unflushed_data_test ( + host STRING PRIMARY KEY, + val DOUBLE, + ts TIMESTAMP TIME INDEX +) ENGINE = mito; + +Affected Rows: 0 + +INSERT INTO discard_unflushed_data_test VALUES ('persisted', 1, 1); + +Affected Rows: 1 + +ADMIN FLUSH_TABLE('discard_unflushed_data_test'); + ++--------------------------------------------------+ +| ADMIN FLUSH_TABLE('discard_unflushed_data_test') | ++--------------------------------------------------+ +| 0 | ++--------------------------------------------------+ + +INSERT INTO discard_unflushed_data_test VALUES ('unflushed', 2, 2); + +Affected Rows: 1 + +ADMIN discard_unflushed('discard_unflushed_data_test'); + ++--------------------------------------------------------+ +| ADMIN discard_unflushed('discard_unflushed_data_test') | ++--------------------------------------------------------+ +| 0 | ++--------------------------------------------------------+ + +-- Repeating the operation is idempotent. +ADMIN discard_unflushed('discard_unflushed_data_test'); + ++--------------------------------------------------------+ +| ADMIN discard_unflushed('discard_unflushed_data_test') | ++--------------------------------------------------------+ +| 0 | ++--------------------------------------------------------+ + +SELECT host, val FROM discard_unflushed_data_test ORDER BY host; + ++-----------+-----+ +| host | val | ++-----------+-----+ +| persisted | 1.0 | ++-----------+-----+ + +-- The function must only be available through ADMIN. +-- SQLNESS REPLACE \nDid\syou\smean.* +SELECT discard_unflushed(0); + +Error: 3000(PlanQuery), Failed to plan SQL: Error during planning: Invalid function 'discard_unflushed'. + +DROP TABLE discard_unflushed_data_test; + +Affected Rows: 0 + +CREATE TABLE discard_unflushed_data_test ( + host STRING PRIMARY KEY, + val DOUBLE, + ts TIMESTAMP TIME INDEX +) ENGINE = mito; + +Affected Rows: 0 + +INSERT INTO discard_unflushed_data_test VALUES ('unflushed', 1, 1); + +Affected Rows: 1 + +-- The table has no persisted data, so discarding leaves it empty. +ADMIN discard_unflushed('discard_unflushed_data_test'); + ++--------------------------------------------------------+ +| ADMIN discard_unflushed('discard_unflushed_data_test') | ++--------------------------------------------------------+ +| 0 | ++--------------------------------------------------------+ + +SELECT COUNT(*) FROM discard_unflushed_data_test; + ++----------+ +| count(*) | ++----------+ +| 0 | ++----------+ + +DROP TABLE discard_unflushed_data_test; + +Affected Rows: 0 + +CREATE TABLE discard_unflushed_data_physical ( + ts TIMESTAMP TIME INDEX, + val DOUBLE +) ENGINE = metric WITH ("physical_metric_table" = ""); + +Affected Rows: 0 + +CREATE TABLE discard_unflushed_data_logical ( + host STRING PRIMARY KEY, + val DOUBLE, + ts TIMESTAMP TIME INDEX +) ENGINE = metric WITH ("on_physical_table" = "discard_unflushed_data_physical"); + +Affected Rows: 0 + +INSERT INTO discard_unflushed_data_logical VALUES ('unflushed', 1, 1); + +Affected Rows: 1 + +-- Logical Metric Engine tables share their physical regions with other logical tables. +-- Discarding by logical table name must not truncate those shared regions. +ADMIN discard_unflushed('discard_unflushed_data_logical'); + +Error: 1002(Unexpected), Failed to execute admin function discard_unflushed: Execution error: Not supported: discarding unflushed data from a Metric Engine logical table + +SELECT host, val FROM discard_unflushed_data_logical; + ++-----------+-----+ +| host | val | ++-----------+-----+ +| unflushed | 1.0 | ++-----------+-----+ + +DROP TABLE discard_unflushed_data_logical; + +Affected Rows: 0 + +DROP TABLE discard_unflushed_data_physical; + +Affected Rows: 0 + diff --git a/tests/cases/standalone/common/function/admin/discard_unflushed_data.sql b/tests/cases/standalone/common/function/admin/discard_unflushed_data.sql new file mode 100644 index 0000000000..9183489495 --- /dev/null +++ b/tests/cases/standalone/common/function/admin/discard_unflushed_data.sql @@ -0,0 +1,62 @@ +CREATE TABLE discard_unflushed_data_test ( + host STRING PRIMARY KEY, + val DOUBLE, + ts TIMESTAMP TIME INDEX +) ENGINE = mito; + +INSERT INTO discard_unflushed_data_test VALUES ('persisted', 1, 1); + +ADMIN FLUSH_TABLE('discard_unflushed_data_test'); + +INSERT INTO discard_unflushed_data_test VALUES ('unflushed', 2, 2); + +ADMIN discard_unflushed('discard_unflushed_data_test'); + +-- Repeating the operation is idempotent. +ADMIN discard_unflushed('discard_unflushed_data_test'); + +SELECT host, val FROM discard_unflushed_data_test ORDER BY host; + +-- The function must only be available through ADMIN. +-- SQLNESS REPLACE \nDid\syou\smean.* +SELECT discard_unflushed(0); + +DROP TABLE discard_unflushed_data_test; + +CREATE TABLE discard_unflushed_data_test ( + host STRING PRIMARY KEY, + val DOUBLE, + ts TIMESTAMP TIME INDEX +) ENGINE = mito; + +INSERT INTO discard_unflushed_data_test VALUES ('unflushed', 1, 1); + +-- The table has no persisted data, so discarding leaves it empty. +ADMIN discard_unflushed('discard_unflushed_data_test'); + +SELECT COUNT(*) FROM discard_unflushed_data_test; + +DROP TABLE discard_unflushed_data_test; + +CREATE TABLE discard_unflushed_data_physical ( + ts TIMESTAMP TIME INDEX, + val DOUBLE +) ENGINE = metric WITH ("physical_metric_table" = ""); + +CREATE TABLE discard_unflushed_data_logical ( + host STRING PRIMARY KEY, + val DOUBLE, + ts TIMESTAMP TIME INDEX +) ENGINE = metric WITH ("on_physical_table" = "discard_unflushed_data_physical"); + +INSERT INTO discard_unflushed_data_logical VALUES ('unflushed', 1, 1); + +-- Logical Metric Engine tables share their physical regions with other logical tables. +-- Discarding by logical table name must not truncate those shared regions. +ADMIN discard_unflushed('discard_unflushed_data_logical'); + +SELECT host, val FROM discard_unflushed_data_logical; + +DROP TABLE discard_unflushed_data_logical; + +DROP TABLE discard_unflushed_data_physical;