diff --git a/Cargo.lock b/Cargo.lock index 6287647a518..29f129ee445 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4518,6 +4518,7 @@ version = "1.4.0-alpha.0" dependencies = [ "api", "arrow-flight", + "async-stream", "async-trait", "bytes", "cache", diff --git a/src/catalog/src/information_extension.rs b/src/catalog/src/information_extension.rs index 86ca82b7deb..259e98e70e7 100644 --- a/src/catalog/src/information_extension.rs +++ b/src/catalog/src/information_extension.rs @@ -25,12 +25,13 @@ use common_meta::rpc::procedure; use common_procedure::{ProcedureInfo, ProcedureState}; use common_query::request::QueryRequest; use common_recordbatch::adapter::{AsyncRecordBatchStreamAdapter, DfRecordBatchStreamAdapter}; +use common_recordbatch::filter::MetadataFilter; use common_recordbatch::util::{ChainedRecordBatchStream, LimitedRecordBatchStream}; use common_recordbatch::{ DfSendableRecordBatchStream, EmptyRecordBatchStream, SendableRecordBatchStream, }; -use datafusion::common::Result; use datafusion::common::tree_node::TreeNodeRecursion; +use datafusion::common::{Result, ScalarValue}; use datafusion::execution::TaskContext; use datafusion::logical_expr::LogicalPlan; use datafusion::physical_expr::{EquivalenceProperties, Partitioning, PhysicalExpr}; @@ -41,10 +42,13 @@ use datatypes::schema::SchemaRef; use futures_util::future::try_join_all; use meta_client::MetaClientRef; use snafu::ResultExt; +use store_api::compaction_task; use store_api::storage::RegionId; use crate::error; -use crate::information_schema::{DatanodeInspectRequest, InformationExtension}; +use crate::information_schema::{ + DatanodeInspectKind, DatanodeInspectRequest, InformationExtension, +}; pub struct DistributedInformationExtension { meta_client: MetaClientRef, @@ -63,12 +67,27 @@ impl DistributedInformationExtension { meta_client: MetaClientRef, datanode_manager: DatanodeManagerRef, request: DatanodeInspectRequest, - ) -> std::result::Result { + ) -> crate::error::Result { + let nodes = meta_client + .list_nodes(Some(Role::Datanode)) + .await + .map_err(BoxedError::new) + .context(crate::error::ListNodesSnafu)?; + Self::inspect_nodes(datanode_manager, request, nodes).await + } + + /// Executes inspection against discovered nodes after applying safe node pruning. + async fn inspect_nodes( + datanode_manager: DatanodeManagerRef, + request: DatanodeInspectRequest, + nodes: Vec, + ) -> crate::error::Result { + let nodes = prune_compaction_nodes(&request, nodes); let limit = request.scan.limit; let plan = request .build_plan() .context(crate::error::DatafusionSnafu)?; - let stream = Self::inspect_query_stream(meta_client, datanode_manager, plan).await?; + let stream = Self::inspect_query_nodes(datanode_manager, plan, nodes).await?; match limit { Some(limit) => Ok(Box::pin(LimitedRecordBatchStream::new(stream, limit))), None => Ok(stream), @@ -80,13 +99,21 @@ impl DistributedInformationExtension { datanode_manager: DatanodeManagerRef, plan: LogicalPlan, ) -> crate::error::Result { - let schema = inspection_query_schema(&plan)?; let nodes = meta_client .list_nodes(Some(Role::Datanode)) .await .map_err(BoxedError::new) .context(crate::error::ListNodesSnafu)?; + Self::inspect_query_nodes(datanode_manager, plan, nodes).await + } + /// Shares query dispatch and empty-stream schema handling for scans and query plans. + async fn inspect_query_nodes( + datanode_manager: DatanodeManagerRef, + plan: LogicalPlan, + nodes: Vec, + ) -> crate::error::Result { + let schema = inspection_query_schema(&plan)?; let streams = try_join_all(nodes.into_iter().map(|node| { let datanode_manager = datanode_manager.clone(); let plan = plan.clone(); @@ -135,6 +162,24 @@ enum InspectPlan { Query(LogicalPlan), } +/// Prunes reporting nodes only for live compaction inspection. Region routes cannot +/// locate tasks left on a former owner, so only known node IDs can exclude nodes. +fn prune_compaction_nodes(request: &DatanodeInspectRequest, nodes: Vec) -> Vec { + if request.kind != DatanodeInspectKind::CompactionTasks { + return nodes; + } + let filter = MetadataFilter::new(&request.scan.filters); + nodes + .into_iter() + .filter(|node| { + filter.matches(|name| { + (name == compaction_task::NODE_ID) + .then_some(ScalarValue::UInt64(Some(node.peer.id))) + }) + }) + .collect() +} + struct DistributedInspectExec { plan: InspectPlan, meta_client: MetaClientRef, @@ -401,21 +446,198 @@ mod tests { use common_meta::node_manager::{DatanodeManager, DatanodeRef}; use common_meta::peer::Peer; use datafusion::common::tree_node::TreeNode; - use datafusion::logical_expr::LogicalPlan; + use datafusion::datasource::empty::EmptyTable; + use datafusion::logical_expr::{Expr, LogicalPlan}; use datafusion::physical_plan::sorts::sort::SortExec; + use datafusion::prelude::SessionContext; + use datatypes::arrow::datatypes::DataType; use meta_client::client::MetaClientBuilder; use store_api::sst_entry::ManifestSstEntry; use store_api::storage::ScanRequest; + use store_api::compaction_task::CompactionTaskEntry; + use super::*; - use crate::information_schema::DatanodeInspectKind; struct UnusedDatanodeManager; #[async_trait::async_trait] impl DatanodeManager for UnusedDatanodeManager { async fn datanode(&self, _node: &Peer) -> DatanodeRef { - panic!("plan tests must not contact datanodes") + panic!("this test must not contact datanodes") + } + } + + fn inspection_nodes() -> Vec { + [1, 2, 3] + .into_iter() + .map(|id| NodeInfo { + peer: Peer::new(id, format!("datanode-{id}:3001")), + last_activity_ts: 0, + status: common_meta::cluster::NodeStatus::Standalone, + version: String::new(), + git_commit: String::new(), + start_time_ms: 0, + total_cpu_millicores: 0, + total_memory_bytes: 0, + cpu_usage_millicores: 0, + memory_usage_bytes: 0, + hostname: String::new(), + env_vars: Default::default(), + schedulable: None, + }) + .collect() + } + + /// Uses the default optimizer to produce the predicates seen by inspection scans. + async fn compaction_sql_filters(predicate: Option<&str>) -> Vec { + let ctx = SessionContext::new(); + ctx.register_table( + "compaction_tasks", + Arc::new(EmptyTable::new( + CompactionTaskEntry::schema().arrow_schema().clone(), + )), + ) + .unwrap(); + let mut sql = "SELECT * FROM compaction_tasks".to_string(); + if let Some(predicate) = predicate { + sql.push_str(" WHERE "); + sql.push_str(predicate); + } + let plan = ctx.sql(&sql).await.unwrap().into_optimized_plan().unwrap(); + let mut filters = Vec::new(); + plan.apply(|plan| { + match plan { + LogicalPlan::Filter(filter) => filters.push(filter.predicate.clone()), + LogicalPlan::TableScan(scan) => filters.extend(scan.filters.iter().cloned()), + _ => {} + } + Ok(TreeNodeRecursion::Continue) + }) + .unwrap(); + filters + } + + #[tokio::test] + async fn test_compaction_node_pruning() { + for (predicate, expected, expect_cast) in [ + (Some("node_id = 2"), vec![2], false), + (Some("node_id != 2"), vec![1, 3], false), + (Some("node_id > 2"), vec![3], false), + (Some("node_id >= 2"), vec![2, 3], false), + (Some("node_id < 2"), vec![1], false), + (Some("node_id <= 2"), vec![1, 2], false), + (Some("2 = node_id"), vec![2], false), + (Some("2 < node_id"), vec![3], false), + (None, vec![1, 2, 3], false), + (Some("state = 'running'"), vec![1, 2, 3], false), + (Some("node_id = 2 OR table_id = 42"), vec![1, 2, 3], false), + (Some("node_id = 2 AND state = 'running'"), vec![2], false), + ] { + let filters = compaction_sql_filters(predicate).await; + let mut has_cast = false; + for filter in &filters { + filter + .apply(|expr| { + has_cast |= matches!(expr, Expr::Cast(cast) if cast.field.data_type() == &DataType::Decimal128(20, 0) + && matches!(cast.expr.as_ref(), Expr::Column(_))); + Ok(TreeNodeRecursion::Continue) + }) + .unwrap(); + } + assert_eq!(has_cast, expect_cast, "optimized predicate: {predicate:?}"); + let request = DatanodeInspectRequest { + kind: DatanodeInspectKind::CompactionTasks, + scan: ScanRequest { + filters, + ..Default::default() + }, + }; + let actual: Vec<_> = prune_compaction_nodes(&request, inspection_nodes()) + .into_iter() + .map(|node| node.peer.id) + .collect(); + assert_eq!( + actual, expected, + "{predicate:?}: {:?}", + request.scan.filters + ); + } + let filters = compaction_sql_filters(Some("node_id = 99")).await; + for kind in [ + DatanodeInspectKind::SstManifest, + DatanodeInspectKind::SstStorage, + DatanodeInspectKind::SstIndexMeta, + DatanodeInspectKind::RegionInfo, + ] { + let request = DatanodeInspectRequest { + kind, + scan: ScanRequest { + filters: filters.clone(), + ..Default::default() + }, + }; + let actual: Vec<_> = prune_compaction_nodes(&request, inspection_nodes()) + .into_iter() + .map(|node| node.peer.id) + .collect(); + assert_eq!(actual, vec![1, 2, 3]); + } + } + + #[tokio::test] + async fn test_compaction_empty_nodes_preserve_projected_schema() { + let filters = compaction_sql_filters(Some("node_id = 99")).await; + for nodes in [vec![], inspection_nodes()] { + for (projection, expected) in [ + ( + vec![3, 0, 2], + vec![ + ("node_id", DataType::UInt64, true), + ("task_id", DataType::Utf8, false), + ("table_id", DataType::UInt32, false), + ], + ), + (vec![], vec![]), + ] { + let request = DatanodeInspectRequest { + kind: DatanodeInspectKind::CompactionTasks, + scan: ScanRequest { + projection: Some(projection), + filters: filters.clone(), + ..Default::default() + }, + }; + // Panics if the production dispatch path contacts any excluded node. + let stream = DistributedInformationExtension::inspect_nodes( + Arc::new(UnusedDatanodeManager), + request, + nodes.clone(), + ) + .await + .unwrap(); + let batches = common_recordbatch::RecordBatches::try_collect(stream) + .await + .unwrap(); + assert_eq!( + batches.iter().map(|batch| batch.num_rows()).sum::(), + 0 + ); + let schema = batches.schema(); + let actual: Vec<_> = schema + .arrow_schema() + .fields() + .iter() + .map(|field| { + ( + field.name().as_str(), + field.data_type().clone(), + field.is_nullable(), + ) + }) + .collect(); + assert_eq!(actual, expected); + } } } diff --git a/src/catalog/src/system_schema/information_schema.rs b/src/catalog/src/system_schema/information_schema.rs index 1e4f08d31fd..2a82e2ed911 100644 --- a/src/catalog/src/system_schema/information_schema.rs +++ b/src/catalog/src/system_schema/information_schema.rs @@ -14,6 +14,7 @@ mod cluster_info; pub mod columns; +mod compaction_tasks; pub mod flow_statistics; pub mod flows; mod information_memory_table; @@ -58,6 +59,7 @@ use lazy_static::lazy_static; use paste::paste; use process_list::InformationSchemaProcessList; use region_info::InformationSchemaRegionInfo; +use store_api::compaction_task::CompactionTaskEntry; use store_api::metric_engine_consts::{ MEMTABLE_PARTITION_TREE_PRIMARY_KEY_ENCODING, PRIMARY_KEY_ENCODING, }; @@ -74,6 +76,7 @@ use crate::CatalogManager; use crate::error::Error; use crate::process_manager::ProcessManagerRef; use crate::system_schema::information_schema::cluster_info::InformationSchemaClusterInfo; +use crate::system_schema::information_schema::compaction_tasks::InformationSchemaCompactionTasks; use crate::system_schema::information_schema::flow_statistics::InformationSchemaFlowStatistics; use crate::system_schema::information_schema::flows::InformationSchemaFlows; use crate::system_schema::information_schema::information_memory_table::get_schema_columns; @@ -312,6 +315,9 @@ impl InformationSchemaProvider { self.catalog_manager.clone(), ), ) as _), + COMPACTION_TASKS => Some(Arc::new(InformationSchemaCompactionTasks::new( + self.catalog_manager.clone(), + )) as _), REGION_INFO => Some(Arc::new(InformationSchemaRegionInfo::new( self.catalog_manager.clone(), )) as _), @@ -382,6 +388,7 @@ impl InformationSchemaProvider { CLUSTER_INFO, PROCEDURE_INFO, REGION_STATISTICS, + COMPACTION_TASKS, REGION_INFO, SSTS_MANIFEST, SSTS_STORAGE, @@ -506,6 +513,8 @@ pub struct DatanodeInspectRequest { /// The kind of the datanode inspect request. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum DatanodeInspectKind { + /// Live compaction cycles. + CompactionTasks, /// List SST entries recorded in manifest SstManifest, /// List SST entries discovered in storage layer @@ -520,6 +529,7 @@ impl DatanodeInspectRequest { /// Builds a logical plan for the datanode inspect request. pub fn build_plan(self) -> std::result::Result { match self.kind { + DatanodeInspectKind::CompactionTasks => CompactionTaskEntry::build_plan(self.scan), DatanodeInspectKind::SstManifest => ManifestSstEntry::build_plan(self.scan), DatanodeInspectKind::SstStorage => StorageSstEntry::build_plan(self.scan), DatanodeInspectKind::SstIndexMeta => PuffinIndexMetaEntry::build_plan(self.scan), @@ -704,11 +714,11 @@ mod tests { // An extension replacing a built-in supplies the replacement table's scope. let provider = provider.with_extra_table_factories(factories.clone()); let expected_count = - 44 + usize::from(with_process_manager) + usize::from(cfg!(feature = "enterprise")) + 45 + usize::from(with_process_manager) + usize::from(cfg!(feature = "enterprise")) - if catalog == DEFAULT_CATALOG_NAME { 0 } else { - 8 + 9 }; assert_eq!(provider.tables().len(), expected_count); for (name, table) in provider.tables() { @@ -824,6 +834,29 @@ mod tests { ); } + #[test] + fn test_compaction_tasks_build_plan() { + let request = DatanodeInspectRequest { + kind: DatanodeInspectKind::CompactionTasks, + scan: ScanRequest { + projection: Some(vec![0]), + filters: vec![ + datafusion::logical_expr::col("state") + .eq(datafusion::logical_expr::lit("running")), + ], + limit: Some(1), + ..Default::default() + }, + }; + let plan = request.build_plan().unwrap(); + assert_eq!(plan.schema().fields().len(), 1); + assert!( + plan.display_indent() + .to_string() + .contains("__compaction_tasks") + ); + } + #[test] fn test_datanode_inspect_region_info_build_plan() { let plan = DatanodeInspectRequest { diff --git a/src/catalog/src/system_schema/information_schema/compaction_tasks.rs b/src/catalog/src/system_schema/information_schema/compaction_tasks.rs new file mode 100644 index 00000000000..53957fa4625 --- /dev/null +++ b/src/catalog/src/system_schema/information_schema/compaction_tasks.rs @@ -0,0 +1,97 @@ +// 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 std::sync::{Arc, Weak}; + +use common_catalog::consts::INFORMATION_SCHEMA_COMPACTION_TASKS_TABLE_ID; +use common_error::ext::BoxedError; +use common_recordbatch::SendableRecordBatchStream; +use common_recordbatch::adapter::AsyncRecordBatchStreamAdapter; +use datafusion::physical_plan::ExecutionPlan; +use datatypes::schema::SchemaRef; +use snafu::ResultExt; +use store_api::compaction_task::CompactionTaskEntry; +use store_api::storage::{ScanRequest, TableId}; + +use crate::CatalogManager; +use crate::error::Result; +use crate::information_schema::{ + COMPACTION_TASKS, DatanodeInspectKind, InformationSchemaTableScope, InformationTable, +}; +use crate::system_schema::SystemTable; +use crate::system_schema::utils; + +/// Information schema table for live compaction tasks. +pub struct InformationSchemaCompactionTasks { + schema: SchemaRef, + catalog_manager: Weak, +} + +impl InformationSchemaCompactionTasks { + pub(crate) fn new(catalog_manager: Weak) -> Self { + Self { + schema: CompactionTaskEntry::schema(), + catalog_manager, + } + } +} + +impl InformationTable for InformationSchemaCompactionTasks { + fn scope(&self) -> InformationSchemaTableScope { + InformationSchemaTableScope::Global + } +} + +impl SystemTable for InformationSchemaCompactionTasks { + fn table_id(&self) -> TableId { + INFORMATION_SCHEMA_COMPACTION_TASKS_TABLE_ID + } + + fn table_name(&self) -> &'static str { + COMPACTION_TASKS + } + + fn schema(&self) -> SchemaRef { + self.schema.clone() + } + + fn to_stream(&self, request: ScanRequest) -> Result { + let schema = utils::projected_schema(&self.schema, &request)?; + + let info_ext = utils::information_extension(&self.catalog_manager)?; + let req = + utils::inspect_request(DatanodeInspectKind::CompactionTasks, &self.schema, request)?; + + let future = async move { + info_ext + .inspect_datanode(req) + .await + .map_err(BoxedError::new) + .context(common_recordbatch::error::ExternalSnafu) + }; + Ok(Box::pin(AsyncRecordBatchStreamAdapter::new( + schema, + Box::pin(future), + ))) + } + + fn scan_plan(&self, request: ScanRequest) -> Result>> { + utils::inspect_plan( + &self.catalog_manager, + DatanodeInspectKind::CompactionTasks, + &self.schema, + request, + ) + } +} diff --git a/src/catalog/src/system_schema/information_schema/ssts.rs b/src/catalog/src/system_schema/information_schema/ssts.rs index 36921d7113f..eb2bc5efd3f 100644 --- a/src/catalog/src/system_schema/information_schema/ssts.rs +++ b/src/catalog/src/system_schema/information_schema/ssts.rs @@ -19,12 +19,8 @@ use common_catalog::consts::{ INFORMATION_SCHEMA_SSTS_STORAGE_TABLE_ID, }; use common_error::ext::BoxedError; -use common_query::logical_plan::is_lossless_substrait_type; use common_recordbatch::SendableRecordBatchStream; use common_recordbatch::adapter::AsyncRecordBatchStreamAdapter; -use datafusion::common::DFSchema; -use datafusion::common::tree_node::{TreeNode, TreeNodeRecursion}; -use datafusion::logical_expr::ExprSchemable; use datafusion::physical_plan::ExecutionPlan; use datatypes::schema::SchemaRef; use snafu::ResultExt; @@ -32,65 +28,14 @@ use store_api::sst_entry::{ManifestSstEntry, PuffinIndexMetaEntry, StorageSstEnt use store_api::storage::{ScanRequest, TableId}; use crate::CatalogManager; -use crate::error::{ProjectSchemaSnafu, Result}; +use crate::error::Result; use crate::information_schema::{ - DatanodeInspectKind, DatanodeInspectRequest, InformationSchemaTableScope, InformationTable, - SSTS_INDEX_META, SSTS_MANIFEST, SSTS_STORAGE, + DatanodeInspectKind, InformationSchemaTableScope, InformationTable, SSTS_INDEX_META, + SSTS_MANIFEST, SSTS_STORAGE, }; use crate::system_schema::SystemTable; use crate::system_schema::utils; -fn projected_schema(schema: &SchemaRef, request: &ScanRequest) -> Result { - if let Some(p) = request.projection.as_deref() { - Ok(Arc::new(schema.try_project(p).context(ProjectSchemaSnafu)?)) - } else { - Ok(schema.clone()) - } -} - -fn inspect_plan( - catalog_manager: &Weak, - kind: DatanodeInspectKind, - schema: &SchemaRef, - request: ScanRequest, -) -> Result>> { - let projected_schema = projected_schema(schema, &request)?; - let info_ext = utils::information_extension(catalog_manager)?; - let req = inspect_request(kind, schema, request)?; - - info_ext.inspect_datanode_plan(req, projected_schema) -} - -// System table predicates are inexact: the frontend keeps a residual filter. Dropping -// an unsafe remote filter preserves correctness only if no remote limit truncates it. -fn inspect_request( - kind: DatanodeInspectKind, - schema: &SchemaRef, - mut scan: ScanRequest, -) -> Result { - let schema = - DFSchema::try_from(schema.arrow_schema().clone()).context(crate::error::DatafusionSnafu)?; - let filter_count = scan.filters.len(); - scan.filters.retain(|filter| { - let mut supported = true; - let visited = filter.apply(|expr| { - if !expr - .get_type(&schema) - .is_ok_and(|data_type| is_lossless_substrait_type(&data_type)) - { - supported = false; - return Ok(TreeNodeRecursion::Stop); - } - Ok(TreeNodeRecursion::Continue) - }); - visited.is_ok() && supported - }); - if scan.filters.len() != filter_count { - scan.limit = None; - } - Ok(DatanodeInspectRequest { kind, scan }) -} - /// Information schema table for sst manifest. pub struct InformationSchemaSstsManifest { schema: SchemaRef, @@ -126,9 +71,9 @@ impl SystemTable for InformationSchemaSstsManifest { } fn to_stream(&self, request: ScanRequest) -> Result { - let schema = projected_schema(&self.schema, &request)?; + let schema = utils::projected_schema(&self.schema, &request)?; let info_ext = utils::information_extension(&self.catalog_manager)?; - let req = inspect_request(DatanodeInspectKind::SstManifest, &self.schema, request)?; + let req = utils::inspect_request(DatanodeInspectKind::SstManifest, &self.schema, request)?; let future = async move { info_ext @@ -144,7 +89,7 @@ impl SystemTable for InformationSchemaSstsManifest { } fn scan_plan(&self, request: ScanRequest) -> Result>> { - inspect_plan( + utils::inspect_plan( &self.catalog_manager, DatanodeInspectKind::SstManifest, &self.schema, @@ -188,10 +133,10 @@ impl SystemTable for InformationSchemaSstsStorage { } fn to_stream(&self, request: ScanRequest) -> Result { - let schema = projected_schema(&self.schema, &request)?; + let schema = utils::projected_schema(&self.schema, &request)?; let info_ext = utils::information_extension(&self.catalog_manager)?; - let req = inspect_request(DatanodeInspectKind::SstStorage, &self.schema, request)?; + let req = utils::inspect_request(DatanodeInspectKind::SstStorage, &self.schema, request)?; let future = async move { info_ext @@ -207,7 +152,7 @@ impl SystemTable for InformationSchemaSstsStorage { } fn scan_plan(&self, request: ScanRequest) -> Result>> { - inspect_plan( + utils::inspect_plan( &self.catalog_manager, DatanodeInspectKind::SstStorage, &self.schema, @@ -251,10 +196,10 @@ impl SystemTable for InformationSchemaSstsIndexMeta { } fn to_stream(&self, request: ScanRequest) -> Result { - let schema = projected_schema(&self.schema, &request)?; + let schema = utils::projected_schema(&self.schema, &request)?; let info_ext = utils::information_extension(&self.catalog_manager)?; - let req = inspect_request(DatanodeInspectKind::SstIndexMeta, &self.schema, request)?; + let req = utils::inspect_request(DatanodeInspectKind::SstIndexMeta, &self.schema, request)?; let future = async move { info_ext @@ -270,7 +215,7 @@ impl SystemTable for InformationSchemaSstsIndexMeta { } fn scan_plan(&self, request: ScanRequest) -> Result>> { - inspect_plan( + utils::inspect_plan( &self.catalog_manager, DatanodeInspectKind::SstIndexMeta, &self.schema, @@ -281,6 +226,7 @@ impl SystemTable for InformationSchemaSstsIndexMeta { #[cfg(test)] mod tests { + use crate::information_schema::DatanodeInspectRequest; use cache::{build_fundamental_cache_registry, with_default_composite_cache_registry}; use common_catalog::consts::{DEFAULT_CATALOG_NAME, INFORMATION_SCHEMA_NAME}; use common_meta::cache::{CacheRegistryBuilder, LayeredCacheRegistryBuilder}; @@ -346,6 +292,10 @@ mod tests { use datafusion::logical_expr::{Expr, col, lit}; for (kind, schema) in [ + ( + DatanodeInspectKind::CompactionTasks, + store_api::compaction_task::CompactionTaskEntry::schema(), + ), (DatanodeInspectKind::SstManifest, ManifestSstEntry::schema()), (DatanodeInspectKind::SstStorage, StorageSstEntry::schema()), ( @@ -361,7 +311,7 @@ mod tests { ..Default::default() }; assert_eq!( - inspect_request(kind, &schema, scan.clone()).unwrap(), + utils::inspect_request(kind, &schema, scan.clone()).unwrap(), DatanodeInspectRequest { kind, scan: scan.clone() @@ -382,7 +332,7 @@ mod tests { let mut expected = scan.clone(); expected.limit = None; assert_eq!( - inspect_request(kind, &schema, input).unwrap(), + utils::inspect_request(kind, &schema, input).unwrap(), DatanodeInspectRequest { kind, scan: expected diff --git a/src/catalog/src/system_schema/information_schema/table_names.rs b/src/catalog/src/system_schema/information_schema/table_names.rs index 3e9b2e2d6de..7ae58ead59c 100644 --- a/src/catalog/src/system_schema/information_schema/table_names.rs +++ b/src/catalog/src/system_schema/information_schema/table_names.rs @@ -60,3 +60,6 @@ pub const USER_PRIVILEGES: &str = "user_privileges"; /// MySQL's session list. GreptimeDB reports its own sessions through [`PROCESS_LIST`]; /// this table only exists so MySQL tooling finds the name and column shape it expects. pub const PROCESSLIST: &str = "processlist"; + +/// Live compaction tasks. +pub const COMPACTION_TASKS: &str = "compaction_tasks"; diff --git a/src/catalog/src/system_schema/utils.rs b/src/catalog/src/system_schema/utils.rs index b989bf91228..03752796664 100644 --- a/src/catalog/src/system_schema/utils.rs +++ b/src/catalog/src/system_schema/utils.rs @@ -12,18 +12,82 @@ // See the License for the specific language governing permissions and // limitations under the License. -use std::sync::Weak; +pub mod tables; + +use std::sync::{Arc, Weak}; use common_meta::key::TableMetadataManagerRef; -use snafu::OptionExt; +use common_query::logical_plan::is_lossless_substrait_type; +use datafusion::common::DFSchema; +use datafusion::common::tree_node::{TreeNode, TreeNodeRecursion}; +use datafusion::logical_expr::ExprSchemable; +use datafusion::physical_plan::ExecutionPlan; +use datatypes::schema::SchemaRef; +use snafu::{OptionExt, ResultExt}; +use store_api::storage::ScanRequest; use crate::CatalogManager; -use crate::error::{GetInformationExtensionSnafu, Result, UpgradeWeakCatalogManagerRefSnafu}; -use crate::information_schema::InformationExtensionRef; +use crate::error::{ + GetInformationExtensionSnafu, ProjectSchemaSnafu, Result, UpgradeWeakCatalogManagerRefSnafu, +}; +use crate::information_schema::{ + DatanodeInspectKind, DatanodeInspectRequest, InformationExtensionRef, +}; use crate::kvbackend::KvBackendCatalogManager; use crate::system_schema::semantic_graph::EntityGraphProviderRef; -pub mod tables; +/// Projects an inspection table schema according to the scan request. +pub(crate) fn projected_schema(schema: &SchemaRef, request: &ScanRequest) -> Result { + if let Some(p) = request.projection.as_deref() { + Ok(Arc::new(schema.try_project(p).context(ProjectSchemaSnafu)?)) + } else { + Ok(schema.clone()) + } +} + +/// Builds a datanode inspection plan with the requested output schema. +pub(crate) fn inspect_plan( + catalog_manager: &Weak, + kind: DatanodeInspectKind, + schema: &SchemaRef, + request: ScanRequest, +) -> Result>> { + let projected_schema = projected_schema(schema, &request)?; + let info_ext = information_extension(catalog_manager)?; + let req = inspect_request(kind, schema, request)?; + + info_ext.inspect_datanode_plan(req, projected_schema) +} + +/// System table predicates are inexact: the frontend keeps a residual filter. Dropping +/// an unsafe remote filter preserves correctness only if no remote limit truncates it. +pub(crate) fn inspect_request( + kind: DatanodeInspectKind, + schema: &SchemaRef, + mut scan: ScanRequest, +) -> Result { + let schema = + DFSchema::try_from(schema.arrow_schema().clone()).context(crate::error::DatafusionSnafu)?; + let filter_count = scan.filters.len(); + scan.filters.retain(|filter| { + let mut supported = true; + let visited = filter.apply(|expr| { + if !expr + .get_type(&schema) + .is_ok_and(|data_type| is_lossless_substrait_type(&data_type)) + { + supported = false; + return Ok(TreeNodeRecursion::Stop); + } + Ok(TreeNodeRecursion::Continue) + }); + visited.is_ok() && supported + }); + if scan.filters.len() != filter_count { + scan.limit = None; + } + Ok(DatanodeInspectRequest { kind, scan }) +} /// Try to get the entity-graph provider from a `[CatalogManager]` weak reference. /// Returns `None` when the manager is not the kv-backed one or the provider has diff --git a/src/common/catalog/src/consts.rs b/src/common/catalog/src/consts.rs index cf2b9363ea8..30e523533a0 100644 --- a/src/common/catalog/src/consts.rs +++ b/src/common/catalog/src/consts.rs @@ -128,6 +128,8 @@ pub const INFORMATION_SCHEMA_PLUGINS_TABLE_ID: u32 = 47; pub const INFORMATION_SCHEMA_USER_PRIVILEGES_TABLE_ID: u32 = 48; /// id for information_schema.PROCESSLIST (for mysql) pub const INFORMATION_SCHEMA_PROCESSLIST_TABLE_ID: u32 = 49; +/// Id for information_schema.compaction_tasks. +pub const INFORMATION_SCHEMA_COMPACTION_TASKS_TABLE_ID: u32 = 50; // ----- End of information_schema tables ----- diff --git a/src/datanode/Cargo.toml b/src/datanode/Cargo.toml index 93bb6e89ed8..5bae783df1e 100644 --- a/src/datanode/Cargo.toml +++ b/src/datanode/Cargo.toml @@ -14,6 +14,7 @@ workspace = true [dependencies] api.workspace = true arrow-flight.workspace = true +async-stream.workspace = true async-trait.workspace = true bytes.workspace = true catalog.workspace = true diff --git a/src/datanode/src/region_server/catalog.rs b/src/datanode/src/region_server/catalog.rs index 789aa514785..9b26c35a660 100644 --- a/src/datanode/src/region_server/catalog.rs +++ b/src/datanode/src/region_server/catalog.rs @@ -29,19 +29,23 @@ use datafusion_expr::{LogicalPlan, TableSource}; use futures::TryStreamExt; use session::context::QueryContextRef; use snafu::{OptionExt, ResultExt}; +use store_api::compaction_task::CompactionTaskEntry; use store_api::region_info::RegionInfoEntry; use store_api::sst_entry::{ManifestSstEntry, PuffinIndexMetaEntry, StorageSstEntry}; use store_api::storage::RegionId; use crate::error::{DataFusionSnafu, ListStorageSstsSnafu, Result, UnexpectedSnafu}; use crate::region_server::RegionServer; -use crate::region_server::catalog::inspect::{InspectTable, ManifestSstSource}; +use crate::region_server::catalog::inspect::{ + CompactionTasksSource, InspectTable, ManifestSstSource, +}; /// Reserved internal table kinds used. /// These are recognized by reserved table names and mapped to providers. #[allow(clippy::enum_variant_names)] #[derive(Clone, Debug, PartialEq, Eq, Hash, Copy)] enum InternalTableKind { + InspectCompactionTasks, InspectSstManifest, InspectSstStorage, InspectSstIndexMeta, @@ -51,6 +55,9 @@ enum InternalTableKind { impl InternalTableKind { /// Determine if the name is a reserved internal table (case-insensitive). pub fn from_table_name(name: &str) -> Option { + if name.eq_ignore_ascii_case(CompactionTaskEntry::reserved_table_name_for_inspection()) { + return Some(Self::InspectCompactionTasks); + } if name.eq_ignore_ascii_case(ManifestSstEntry::reserved_table_name_for_inspection()) { return Some(Self::InspectSstManifest); } @@ -69,6 +76,7 @@ impl InternalTableKind { /// Return the `TableProvider` for the internal table. pub async fn table_provider(&self, server: &RegionServer) -> Result> { match self { + Self::InspectCompactionTasks => server.inspect_compaction_tasks_provider().await, Self::InspectSstManifest => server.inspect_sst_manifest_provider().await, Self::InspectSstStorage => server.inspect_sst_storage_provider().await, Self::InspectSstIndexMeta => server.inspect_sst_index_meta_provider().await, @@ -78,6 +86,20 @@ impl InternalTableKind { } impl RegionServer { + /// Exposes current scheduler state through the shared lazy inspection provider. + pub async fn inspect_compaction_tasks_provider(&self) -> Result> { + let mito = { + let guard = self.inner.mito_engine.read().unwrap(); + guard.as_ref().cloned().context(UnexpectedSnafu { + violated: "mito engine not available", + })? + }; + + Ok(Arc::new(InspectTable::new(Arc::new( + CompactionTasksSource::new(mito), + )))) + } + /// Exposes manifest SSTs through a lazy table with metadata predicate pruning. pub async fn inspect_sst_manifest_provider(&self) -> Result> { let mito = { @@ -353,6 +375,7 @@ impl TreeNodeRewriter for NameAwareDataSourceInjector { #[cfg(test)] mod tests { + mod compaction; mod manifest; use std::sync::Arc; diff --git a/src/datanode/src/region_server/catalog/inspect.rs b/src/datanode/src/region_server/catalog/inspect.rs index 307c8b42883..da3ebdb130b 100644 --- a/src/datanode/src/region_server/catalog/inspect.rs +++ b/src/datanode/src/region_server/catalog/inspect.rs @@ -27,6 +27,7 @@ use datafusion_common::DataFusionError; use datafusion_expr::{Expr, TableProviderFilterPushDown, TableType}; use futures::StreamExt; use mito2::engine::MitoEngine; +use store_api::compaction_task::{CompactionTaskBatchBuilder, CompactionTaskEntry}; use store_api::sst_entry::{ManifestSstBatchBuilder, ManifestSstEntry}; /// Supplies batches for an internal inspection table. Implementations may prune @@ -156,6 +157,48 @@ impl InspectSource for ManifestSstSource { } } +/// Reads live compaction state only when a scan is executed. +pub(crate) struct CompactionTasksSource { + engine: MitoEngine, +} + +impl CompactionTasksSource { + /// Creates a source backed by the engine's live compaction cycles. + pub(crate) fn new(engine: MitoEngine) -> Self { + Self { engine } + } +} + +impl std::fmt::Debug for CompactionTasksSource { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("CompactionTasksSource") + .finish_non_exhaustive() + } +} + +impl InspectSource for CompactionTasksSource { + fn schema(&self) -> SchemaRef { + CompactionTaskEntry::schema().arrow_schema().clone() + } + + fn scan(&self, request: &InspectScan, _ctx: Arc) -> SendableRecordBatchStream { + let engine = self.engine.clone(); + let request = request.clone(); + let schema = request.schema.clone(); + let builder = CompactionTaskBatchBuilder::new(schema.clone()); + let stream = async_stream::try_stream! { + let entries = engine + .inspect_compaction_tasks(&request.filters) + .await + .map_err(|error| DataFusionError::External(Box::new(error)))?; + for entries in entries.chunks(request.batch_size) { + yield builder.build(entries)?; + } + }; + Box::pin(RecordBatchStreamAdapter::new(schema, stream)) + } +} + #[derive(Debug)] struct InspectPartition { source: Arc, @@ -308,4 +351,139 @@ mod tests { assert_eq!(source.scans.load(Ordering::SeqCst), 1); assert_eq!(source.drops.load(Ordering::SeqCst), 1); } + #[tokio::test] + async fn test_manifest_optimized_sql_pruning() { + use std::sync::Mutex; + + use datafusion_expr::{col, lit}; + use mito2::config::MitoConfig; + use mito2::manifest::action::RegionEdit; + use mito2::sst::file::FileMeta; + use mito2::test_util::{CreateRequestBuilder, TestEnv}; + use store_api::region_engine::RegionEngine; + use store_api::region_request::RegionRequest; + use store_api::storage::{FileId, RegionId}; + + #[derive(Debug)] + struct RecordingSource { + inner: ManifestSstSource, + filters: Mutex>, + candidates: Arc, + } + + impl InspectSource for RecordingSource { + fn schema(&self) -> SchemaRef { + self.inner.schema() + } + + fn scan( + &self, + request: &InspectScan, + ctx: Arc, + ) -> SendableRecordBatchStream { + *self.filters.lock().unwrap() = request.filters.clone(); + let candidates = self.candidates.clone(); + let stream = self.inner.scan(request, ctx).map(move |batch| { + if let Ok(batch) = &batch { + candidates.fetch_add(batch.num_rows(), Ordering::SeqCst); + } + batch + }); + Box::pin(RecordBatchStreamAdapter::new( + request.schema.clone(), + stream, + )) + } + } + + let mut env = TestEnv::new().await.with_node_id(42); + let engine = env + .create_engine(MitoConfig { + schedule_compaction_after_edit: false, + ..Default::default() + }) + .await; + let region_id = RegionId::new(11, 1); + engine + .handle_request( + region_id, + RegionRequest::Create(CreateRequestBuilder::new().build()), + ) + .await + .unwrap(); + engine + .edit_region( + region_id, + RegionEdit { + files_to_add: vec![FileMeta { + region_id, + file_id: FileId::from_bytes(1u128.to_be_bytes()), + ..Default::default() + }], + files_to_remove: vec![], + timestamp_ms: None, + compaction_time_window: None, + flushed_entry_id: None, + flushed_sequence: None, + committed_sequence: None, + }, + ) + .await + .unwrap(); + let source = Arc::new(RecordingSource { + inner: ManifestSstSource::new(engine), + filters: Mutex::new(Vec::new()), + candidates: Arc::new(AtomicUsize::new(0)), + }); + let ctx = SessionContext::new(); + ctx.register_table("manifest", Arc::new(InspectTable::new(source.clone()))) + .unwrap(); + for (predicate, expected_filter, expected_count) in [ + ("node_id = 99".to_string(), col("node_id").eq(lit(99u64)), 0), + ( + "node_id != 42".to_string(), + col("node_id").not_eq(lit(42u64)), + 0, + ), + ("node_id = 42".to_string(), col("node_id").eq(lit(42u64)), 1), + ( + "region_id = 0".to_string(), + col("region_id").eq(lit(0u64)), + 0, + ), + ( + format!("region_id = {}", region_id.as_u64()), + col("region_id").eq(lit(region_id.as_u64())), + 1, + ), + ] { + source.candidates.store(0, Ordering::SeqCst); + source.filters.lock().unwrap().clear(); + // collect() runs the default optimizer and the actual inspection provider. + let batches = ctx + .sql(&format!("SELECT file_id FROM manifest WHERE {predicate}")) + .await + .unwrap() + .collect() + .await + .unwrap(); + // The scanner receives a UInt64 comparison, with no Decimal cast. + assert_eq!( + *source.filters.lock().unwrap(), + vec![expected_filter], + "{predicate}" + ); + // Count the manifest source's candidates before DataFusion's residual filter. + assert_eq!( + source.candidates.load(Ordering::SeqCst), + expected_count, + "{predicate}" + ); + assert_eq!( + batches.iter().map(|batch| batch.num_rows()).sum::(), + expected_count, + "{predicate}" + ); + } + } } diff --git a/src/datanode/src/region_server/catalog/tests/compaction.rs b/src/datanode/src/region_server/catalog/tests/compaction.rs new file mode 100644 index 00000000000..49ba1270abc --- /dev/null +++ b/src/datanode/src/region_server/catalog/tests/compaction.rs @@ -0,0 +1,206 @@ +// 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 std::sync::Arc; +use std::time::Duration; + +use api::v1::Rows; +use datafusion::arrow::compute::concat_batches; +use datafusion::arrow::record_batch::RecordBatch; +use datafusion::catalog::MemTable; +use datafusion::prelude::SessionContext; +use mito2::config::MitoConfig; +use mito2::engine::listener::EventListener; +use mito2::test_util::{ + CreateRequestBuilder, TestEnv, build_rows_for_key, flush_region, put_rows, rows_schema, +}; +use store_api::compaction_task::CompactionTaskEntry; +use store_api::region_engine::RegionEngine; +use store_api::region_request::{RegionCompactRequest, RegionRequest}; +use store_api::storage::RegionId; +use tokio::sync::Semaphore; + +use crate::region_server::catalog::inspect::{CompactionTasksSource, InspectTable}; + +struct MergeGate { + entered: Semaphore, + resume: Semaphore, +} + +#[async_trait::async_trait] +impl EventListener for MergeGate { + async fn on_compaction_unit_merge_begin(&self, _: RegionId, _: u64) { + self.entered.add_permits(1); + if let Ok(permit) = self.resume.acquire().await { + permit.forget(); + } + } +} + +struct ReleaseGate(Arc); +impl Drop for ReleaseGate { + fn drop(&mut self) { + self.0.resume.close(); + } +} + +async fn query(context: &SessionContext, sql: &str) -> RecordBatch { + let frame = context.sql(sql).await.unwrap(); + let schema = Arc::new(frame.schema().as_arrow().clone()); + let batches = frame.collect().await.unwrap(); + concat_batches(&schema, &batches).unwrap() +} + +/// Ablates lazy inspection against eager snapshots under the same gated workload. +#[tokio::test] +async fn test_compaction_tasks_live_scan_and_pushdown() { + let gate = Arc::new(MergeGate { + entered: Semaphore::new(0), + resume: Semaphore::new(0), + }); + let guard = ReleaseGate(gate.clone()); + let mut env = TestEnv::new().await; + let engine = env + .create_engine_with( + MitoConfig { + max_background_compactions: 2, + min_compaction_interval: Duration::from_secs(3600), + ..Default::default() + }, + None, + Some(gate.clone()), + None, + ) + .await; + let mut jobs = Vec::new(); + for table_id in [11, 12] { + let region_id = RegionId::new(table_id, 1); + let create = CreateRequestBuilder::new() + .insert_option("compaction.type", "twcs") + .insert_option("compaction.twcs.trigger_file_num", "100") + .build(); + let schema = rows_schema(&create); + engine + .handle_request(region_id, RegionRequest::Create(create)) + .await + .unwrap(); + for _ in 0..2 { + put_rows( + &engine, + region_id, + Rows { + schema: schema.clone(), + rows: build_rows_for_key("a", 0, 10, 0), + }, + ) + .await; + flush_region(&engine, region_id, None).await; + } + let engine = engine.clone(); + jobs.push(tokio::spawn(async move { + engine + .handle_request( + region_id, + RegionRequest::Compact(RegionCompactRequest { + options: api::v1::region::compact_request::Options::StrictWindow( + api::v1::region::StrictWindow { + window_seconds: 3600, + }, + ), + ..Default::default() + }), + ) + .await + .unwrap(); + })); + } + tokio::time::timeout(Duration::from_secs(20), gate.entered.acquire_many(2)) + .await + .unwrap() + .unwrap() + .forget(); + let entries = engine.inspect_compaction_tasks(&[]).await.unwrap(); + assert_eq!(entries.len(), 2); + for entry in &entries { + assert_eq!(entry.state.as_ref(), "running"); + assert_eq!(entry.trigger_type.as_ref(), "manual"); + assert_eq!(entry.input_files, Some(2)); + let options: serde_json::Value = serde_json::from_str(&entry.request_options).unwrap(); + assert_eq!(options["window_seconds"], 3600); + assert!(entry.effective_options.is_some()); + } + let lazy = SessionContext::new(); + lazy.register_table( + "tasks", + Arc::new(InspectTable::new(Arc::new(CompactionTasksSource::new( + engine.clone(), + )))), + ) + .unwrap(); + let eager = SessionContext::new(); + let batch = CompactionTaskEntry::to_record_batch(&entries).unwrap(); + eager + .register_table( + "tasks", + Arc::new(MemTable::try_new(batch.schema(), vec![vec![batch]]).unwrap()), + ) + .unwrap(); + // Repeat with identical state and workload; only the scan implementation differs. + for _ in 0..3 { + for sql in [ + "SELECT task_id, request_options, effective_options FROM tasks ORDER BY task_id", + "SELECT task_id, input_time_start, input_time_end, output_time_start, output_time_end FROM tasks ORDER BY task_id", + "SELECT task_id FROM tasks WHERE input_time_start <= CAST('1970-01-01T00:00:01Z' AS TIMESTAMP(9)) AND output_time_end > CAST('1970-01-01T00:00:01Z' AS TIMESTAMP(9)) ORDER BY task_id", + "SELECT task_id FROM tasks WHERE output_time_end IS NULL ORDER BY task_id", + "SELECT task_id FROM tasks WHERE table_id = 12 AND state = 'running' LIMIT 1", + "SELECT count(*) FROM tasks WHERE table_id = 12", + "SELECT task_id FROM tasks WHERE table_id = 99 LIMIT 1", + "SELECT task_id FROM tasks WHERE table_id = 99 OR trigger_type = 'manual' ORDER BY task_id", + "SELECT task_id FROM tasks WHERE length(request_options) > 2 ORDER BY task_id LIMIT 1", + "SELECT task_id FROM tasks WHERE table_id = 12 AND length(request_options) < 2 LIMIT 1", + "SELECT task_id FROM tasks WHERE table_id = 99 OR length(request_options) > 2 ORDER BY task_id LIMIT 1", + "SELECT table_id FROM tasks ORDER BY table_id DESC LIMIT 1", + "SELECT task_id FROM tasks WHERE node_id IS NULL ORDER BY task_id", + "SELECT task_id FROM tasks LIMIT 0", + ] { + assert_eq!(query(&lazy, sql).await, query(&eager, sql).await, "{sql}"); + } + } + drop(guard); + for job in jobs { + tokio::time::timeout(Duration::from_secs(20), job) + .await + .unwrap() + .unwrap(); + } + tokio::time::timeout(Duration::from_secs(10), async { + loop { + if engine + .inspect_compaction_tasks(&[]) + .await + .unwrap() + .is_empty() + { + break; + } + tokio::task::yield_now().await; + } + }) + .await + .unwrap(); + assert_eq!( + query(&lazy, "SELECT task_id FROM tasks").await.num_rows(), + 0 + ); +} diff --git a/src/mito2/AGENTS.md b/src/mito2/AGENTS.md index eb99b9b31d9..50c5abf311f 100644 --- a/src/mito2/AGENTS.md +++ b/src/mito2/AGENTS.md @@ -63,7 +63,8 @@ without sending a worker request. The scheduler publishes active and pending tas state together; readers clone task Arcs under a short read lock, then filter and build rows outside the lock. Execution handles provide live background phases. There is no second task registry or completed-task history. See -`compaction/scheduler/state.rs`. +`compaction/scheduler/state.rs`. The datanode provider reuses the lazy inspection scan +path and retains residual SQL filters. ## Public surface diff --git a/tests-integration/src/standalone.rs b/tests-integration/src/standalone.rs index 5bd0d970de5..142ef72b000 100644 --- a/tests-integration/src/standalone.rs +++ b/tests-integration/src/standalone.rs @@ -19,7 +19,6 @@ use cache::{ build_datanode_layered_cache_registry, build_fundamental_cache_registry, with_default_composite_cache_registry, }; -use catalog::information_schema::NoopInformationExtension; use catalog::kvbackend::KvBackendCatalogManagerBuilder; use catalog::process_manager::ProcessManager; use common_base::Plugins; @@ -55,7 +54,10 @@ use frontend::service_config::{BatcherOptions, PendingRowsBatcherOptions}; use meta_srv::metasrv::{FLOW_ID_SEQ, TABLE_ID_SEQ}; use servers::grpc::GrpcOptions; use standalone::options::StandaloneOptions; -use standalone::{StandaloneDatanodeManager, StandaloneRepartitionProcedureFactory}; +use standalone::{ + StandaloneDatanodeManager, StandaloneInformationExtension, + StandaloneRepartitionProcedureFactory, +}; use crate::test_util::{self, StorageType, TestGuard, create_tmp_dir_and_datanode_opts}; @@ -234,8 +236,12 @@ impl GreptimeDbStandaloneBuilder { .build(), ); + let information_extension = Arc::new(StandaloneInformationExtension::new( + datanode.region_server(), + procedure_manager.clone(), + )); let catalog_manager = KvBackendCatalogManagerBuilder::new( - Arc::new(NoopInformationExtension), + information_extension.clone(), kv_backend.clone(), cache_registry.clone(), ) @@ -253,6 +259,9 @@ impl GreptimeDbStandaloneBuilder { Arc::new(frontend_client), ); let flownode = Arc::new(flow_builder.build().await.unwrap()); + information_extension + .set_flow_engine(flownode.flow_engine()) + .await; let node_manager = Arc::new(StandaloneDatanodeManager { region_server: datanode.region_server(), diff --git a/tests-integration/src/tests/instance_test.rs b/tests-integration/src/tests/instance_test.rs index 0615b684f6b..84689fe0613 100644 --- a/tests-integration/src/tests/instance_test.rs +++ b/tests-integration/src/tests/instance_test.rs @@ -22,6 +22,7 @@ use common_error::status_code::StatusCode; use common_query::Output; use common_recordbatch::util; use common_test_util::recordbatch::check_output_stream; +use datanode::region_server::RegionServer; use datatypes::arrow::array::{ ArrayRef, AsArray, StringArray, TimestampMillisecondArray, UInt64Array, }; @@ -4979,3 +4980,308 @@ async fn test_exemplar_source_identity_policy() { .collect::>(); assert_eq!(ids, vec![source.table_id(), replacement.table_id()]); } + +#[tokio::test] +async fn test_compaction_tasks_sql_lifecycle() { + use crate::standalone::GreptimeDbStandaloneBuilder; + + let instance = GreptimeDbStandaloneBuilder::new("compaction_tasks_sql_lifecycle") + .build() + .await; + check_compaction_tasks_sql_lifecycle( + instance.fe_instance(), + instance.datanode.region_server(), + 0, + None, + ) + .await; +} + +#[tokio::test] +async fn test_compaction_tasks_distributed_sql_lifecycle() { + use common_meta::peer::Peer; + use meta_srv::error::Result as MetaResult; + use meta_srv::metasrv::SelectorContext; + use meta_srv::selector::{Selector, SelectorOptions}; + + use crate::cluster::GreptimeDbClusterBuilder; + use crate::test_util::PEER_PLACEHOLDER_ADDR; + + struct CompactionNodeSelector; + + #[async_trait::async_trait] + impl Selector for CompactionNodeSelector { + type Context = SelectorContext; + type Output = Vec; + + async fn select( + &self, + _ctx: &Self::Context, + _opts: SelectorOptions<'_>, + ) -> MetaResult { + Ok(vec![Peer { + id: 1, + addr: PEER_PLACEHOLDER_ADDR.to_string(), + }]) + } + } + + let cluster = GreptimeDbClusterBuilder::new("compaction_tasks_distributed_sql_lifecycle") + .await + .with_datanodes(2) + // Only node 1 gets the gated engine; place the table there deterministically. + .with_meta_selector(Arc::new(CompactionNodeSelector)) + .build(false) + .await; + assert_eq!(cluster.datanode_instances.len(), 2); + check_compaction_tasks_sql_lifecycle( + cluster.fe_instance(), + cluster.datanode_instances[&1].region_server(), + 1, + Some(2), + ) + .await; +} + +/// Exercises SQL -> catalog -> datanode inspection -> live scheduler state. +async fn check_compaction_tasks_sql_lifecycle( + frontend: &Arc, + mut region_server: RegionServer, + node_id: u64, + other_node_id: Option, +) { + use std::time::Duration; + + use common_recordbatch::RecordBatches; + use mito2::config::MitoConfig; + use mito2::engine::listener::CompactionListener; + use mito2::test_util::TestEnv; + + struct Release(Arc); + + impl Drop for Release { + fn drop(&mut self) { + self.0.wake(); + } + } + + async fn rows(frontend: &Arc, sql: &str) -> RecordBatches { + let output = execute_sql(frontend, sql).await; + match output.data { + OutputData::Stream(stream) => RecordBatches::try_collect(stream).await.unwrap(), + OutputData::RecordBatches(batches) => batches, + _ => panic!("expected query output"), + } + } + + let listener = Arc::new(CompactionListener::default()); + let release = Release(listener.clone()); + let mut env = TestEnv::new().await.with_node_id(node_id); + let engine = env + .create_engine_with( + MitoConfig { + min_compaction_interval: Duration::from_secs(3600), + ..Default::default() + }, + None, + Some(listener.clone()), + None, + ) + .await; + // Install a real engine with a deterministic merge gate before creating any user table. + region_server.register_engine(Arc::new(engine.clone())); + execute_sql(frontend, "CREATE TABLE compact_inspect (host STRING PRIMARY KEY, ts TIMESTAMP TIME INDEX, val DOUBLE) WITH ('compaction.type' = 'twcs', 'compaction.twcs.trigger_file_num' = '100')").await; + for value in [1, 2] { + execute_sql( + frontend, + &format!("INSERT INTO compact_inspect VALUES ('a', 0, {value}), ('a', 1000, {value})"), + ) + .await; + execute_sql(frontend, "ADMIN FLUSH_TABLE('compact_inspect')").await; + } + let before = rows(frontend, "SELECT * FROM compact_inspect ORDER BY ts") + .await + .pretty_print() + .unwrap(); + let compact = { + let frontend = frontend.clone(); + tokio::spawn(async move { + execute_sql( + &frontend, + "ADMIN COMPACT_TABLE('compact_inspect', 'strict_window', 'window=3600')", + ) + .await + }) + }; + tokio::time::timeout(Duration::from_secs(20), listener.wait_handle_finished()) + .await + .unwrap(); + let live_tasks = engine.inspect_compaction_tasks(&[]).await.unwrap(); + assert_eq!(live_tasks.len(), 1); + assert_eq!(live_tasks[0].state.as_ref(), "running"); + assert_eq!(live_tasks[0].node_id, Some(node_id)); + // Input SST bounds are inclusive; strict-window output bounds are [start, end). + let ranges = rows(frontend, &format!("SELECT input_time_start, input_time_end, output_time_start, output_time_end FROM information_schema.compaction_tasks WHERE node_id = {node_id} AND input_time_end = CAST('1970-01-01T00:00:01Z' AS TIMESTAMP(9))")).await; + assert_eq!( + ranges.iter().map(|batch| batch.num_rows()).sum::(), + 1 + ); + let range_batch = ranges.iter().find(|batch| batch.num_rows() > 0).unwrap(); + for (index, expected) in [0, 1_000_000_000, 0, 3_600_000_000_000i64] + .into_iter() + .enumerate() + { + let values = range_batch + .column(index) + .as_any() + .downcast_ref::() + .unwrap(); + assert_eq!(values.iter().collect::>(), vec![Some(expected)]); + } + let snapshot = rows(frontend, &format!("SELECT task_id, trigger_type, state, input_files, request_options, effective_options, node_id FROM information_schema.compaction_tasks WHERE trigger_type = 'manual' AND node_id = {node_id} LIMIT 1")).await; + assert_eq!(snapshot.iter().map(|b| b.num_rows()).sum::(), 1); + let snapshot = snapshot.iter().find(|batch| batch.num_rows() > 0).unwrap(); + assert_eq!( + snapshot.column(0).as_string::().value(0), + live_tasks[0].task_id.as_ref() + ); + assert_eq!(snapshot.column(1).as_string::().value(0), "manual"); + assert_eq!(snapshot.column(2).as_string::().value(0), "running"); + assert_eq!( + snapshot + .column(3) + .as_any() + .downcast_ref::() + .unwrap() + .value(0), + 2 + ); + let request: serde_json::Value = + serde_json::from_str(snapshot.column(4).as_string::().value(0)).unwrap(); + assert_eq!(request["type"], "strict_window"); + assert_eq!(request["window_seconds"], 3600); + let effective: serde_json::Value = + serde_json::from_str(snapshot.column(5).as_string::().value(0)).unwrap(); + assert_eq!(effective["time_window_seconds"], 3600); + assert_eq!( + snapshot + .column(6) + .as_any() + .downcast_ref::() + .unwrap() + .value(0), + node_id + ); + assert_eq!( + rows( + frontend, + "SELECT task_id FROM information_schema.compaction_tasks WHERE node_id IS NULL" + ) + .await + .iter() + .map(|batch| batch.num_rows()) + .sum::(), + 0 + ); + let nodes = rows( + frontend, + "SELECT node_id FROM information_schema.compaction_tasks", + ) + .await; + assert_eq!(nodes.iter().map(|batch| batch.num_rows()).sum::(), 1); + assert_eq!( + nodes + .iter() + .find(|batch| batch.num_rows() > 0) + .unwrap() + .column(0) + .as_any() + .downcast_ref::() + .unwrap() + .value(0), + node_id + ); + if let Some(other_node_id) = other_node_id { + let count = rows(frontend, &format!( + "SELECT count(*) FROM information_schema.compaction_tasks WHERE node_id = {other_node_id}" + )).await.pretty_print().unwrap(); + assert_eq!( + count, + "+----------+\n| count(*) |\n+----------+\n| 0 |\n+----------+" + ); + } + for sql in [ + "SELECT count(*) FROM information_schema.compaction_tasks", + "SELECT count(*) FROM information_schema.compaction_tasks WHERE state = 'running'", + ] { + let count = rows(frontend, sql).await.pretty_print().unwrap(); + assert_eq!( + count, + "+----------+\n| count(*) |\n+----------+\n| 1 |\n+----------+" + ); + } + assert_eq!(rows(frontend, "SELECT task_id FROM information_schema.compaction_tasks WHERE node_id = 18446744073709551615").await.iter().map(|b| b.num_rows()).sum::(), 0); + drop(release); + tokio::time::timeout(Duration::from_secs(20), compact) + .await + .unwrap() + .unwrap(); + tokio::time::timeout(Duration::from_secs(10), async { + loop { + if engine + .inspect_compaction_tasks(&[]) + .await + .unwrap() + .is_empty() + { + break; + } + tokio::task::yield_now().await; + } + }) + .await + .unwrap(); + assert_eq!( + rows( + frontend, + "SELECT task_id FROM information_schema.compaction_tasks" + ) + .await + .iter() + .map(|b| b.num_rows()) + .sum::(), + 0 + ); + assert_eq!( + before, + rows(frontend, "SELECT * FROM compact_inspect ORDER BY ts") + .await + .pretty_print() + .unwrap() + ); + execute_sql(frontend, "DROP TABLE compact_inspect").await; +} + +#[apply(both_instances_cases)] +async fn test_compaction_tasks_empty_and_node_pruning(instance: Arc) { + let frontend = instance.frontend(); + for sql in [ + "SELECT task_id FROM information_schema.compaction_tasks", + "SELECT task_id FROM information_schema.compaction_tasks WHERE node_id = 18446744073709551615 LIMIT 1", + "SELECT task_id FROM information_schema.compaction_tasks WHERE state = 'running' AND region_id = 0", + ] { + let output = execute_sql(&frontend, sql).await; + let batches = match output.data { + OutputData::Stream(stream) => common_recordbatch::RecordBatches::try_collect(stream) + .await + .unwrap(), + OutputData::RecordBatches(batches) => batches, + _ => panic!("expected query output"), + }; + assert_eq!( + batches.iter().map(|b| b.num_rows()).sum::(), + 0, + "{sql}" + ); + } +} diff --git a/tests/cases/standalone/common/information_schema/compaction_tasks.result b/tests/cases/standalone/common/information_schema/compaction_tasks.result new file mode 100644 index 00000000000..778b15dc997 --- /dev/null +++ b/tests/cases/standalone/common/information_schema/compaction_tasks.result @@ -0,0 +1,49 @@ +DESC TABLE information_schema.compaction_tasks; + ++-------------------+----------------------+-----+------+---------+---------------+ +| Column | Type | Key | Null | Default | Semantic Type | ++-------------------+----------------------+-----+------+---------+---------------+ +| task_id | String | | NO | | FIELD | +| region_id | UInt64 | | NO | | FIELD | +| table_id | UInt32 | | NO | | FIELD | +| node_id | UInt64 | | YES | | FIELD | +| trigger_type | String | | NO | | FIELD | +| state | String | | NO | | FIELD | +| phase | String | | NO | | FIELD | +| submitted_at | TimestampMillisecond | | NO | | FIELD | +| elapsed_millis | UInt64 | | NO | | FIELD | +| input_files | UInt64 | | YES | | FIELD | +| input_bytes | UInt64 | | YES | | FIELD | +| total_units | UInt64 | | YES | | FIELD | +| completed_units | UInt64 | | YES | | FIELD | +| request_options | String | | NO | | FIELD | +| effective_options | String | | YES | | FIELD | +| input_time_start | TimestampNanosecond | | YES | | FIELD | +| input_time_end | TimestampNanosecond | | YES | | FIELD | +| output_time_start | TimestampNanosecond | | YES | | FIELD | +| output_time_end | TimestampNanosecond | | YES | | FIELD | ++-------------------+----------------------+-----+------+---------+---------------+ + +SELECT * FROM information_schema.compaction_tasks WHERE node_id = 18446744073709551615; + ++---------+-----------+----------+---------+--------------+-------+-------+--------------+----------------+-------------+-------------+-------------+-----------------+-----------------+-------------------+------------------+----------------+-------------------+-----------------+ +| task_id | region_id | table_id | node_id | trigger_type | state | phase | submitted_at | elapsed_millis | input_files | input_bytes | total_units | completed_units | request_options | effective_options | input_time_start | input_time_end | output_time_start | output_time_end | ++---------+-----------+----------+---------+--------------+-------+-------+--------------+----------------+-------------+-------------+-------------+-----------------+-----------------+-------------------+------------------+----------------+-------------------+-----------------+ ++---------+-----------+----------+---------+--------------+-------+-------+--------------+----------------+-------------+-------------+-------------+-----------------+-----------------+-------------------+------------------+----------------+-------------------+-----------------+ + +-- Constant projections and aggregates must preserve the empty row count. +SELECT 1 FROM information_schema.compaction_tasks WHERE node_id = 18446744073709551615; + ++----------+ +| Int64(1) | ++----------+ ++----------+ + +SELECT count(*) FROM information_schema.compaction_tasks WHERE node_id = 18446744073709551615; + ++----------+ +| count(*) | ++----------+ +| 0 | ++----------+ + diff --git a/tests/cases/standalone/common/information_schema/compaction_tasks.sql b/tests/cases/standalone/common/information_schema/compaction_tasks.sql new file mode 100644 index 00000000000..854f9c02772 --- /dev/null +++ b/tests/cases/standalone/common/information_schema/compaction_tasks.sql @@ -0,0 +1,8 @@ +DESC TABLE information_schema.compaction_tasks; + +SELECT * FROM information_schema.compaction_tasks WHERE node_id = 18446744073709551615; + +-- Constant projections and aggregates must preserve the empty row count. +SELECT 1 FROM information_schema.compaction_tasks WHERE node_id = 18446744073709551615; + +SELECT count(*) FROM information_schema.compaction_tasks WHERE node_id = 18446744073709551615; diff --git a/tests/cases/standalone/common/show/show_databases_tables.result b/tests/cases/standalone/common/show/show_databases_tables.result index d9eaaaa3cfe..05481af85af 100644 --- a/tests/cases/standalone/common/show/show_databases_tables.result +++ b/tests/cases/standalone/common/show/show_databases_tables.result @@ -36,6 +36,7 @@ SHOW TABLES; | column_privileges | | column_statistics | | columns | +| compaction_tasks | | engines | | events | | files | @@ -93,6 +94,7 @@ SHOW FULL TABLES; | column_privileges | LOCAL TEMPORARY | | column_statistics | LOCAL TEMPORARY | | columns | LOCAL TEMPORARY | +| compaction_tasks | LOCAL TEMPORARY | | engines | LOCAL TEMPORARY | | events | LOCAL TEMPORARY | | files | LOCAL TEMPORARY | @@ -144,6 +146,7 @@ SHOW TABLE STATUS; |column_privileges||11|Fixed|0|0|0|0|0|0|0|DATETIME|DATETIME||utf8_bin|0|greptime.semantic.table_scope=global|| |column_statistics||11|Fixed|0|0|0|0|0|0|0|DATETIME|DATETIME||utf8_bin|0|greptime.semantic.table_scope=global|| |columns||11|Fixed|0|0|0|0|0|0|0|DATETIME|DATETIME||utf8_bin|0|greptime.semantic.table_scope=catalog|| +|compaction_tasks||11|Fixed|0|0|0|0|0|0|0|DATETIME|DATETIME||utf8_bin|0|greptime.semantic.table_scope=global|| |engines||11|Fixed|0|0|0|0|0|0|0|DATETIME|DATETIME||utf8_bin|0|greptime.semantic.table_scope=public_global|| |events||11|Fixed|0|0|0|0|0|0|0|DATETIME|DATETIME||utf8_bin|0|greptime.semantic.table_scope=global|| |files||11|Fixed|0|0|0|0|0|0|0|DATETIME|DATETIME||utf8_bin|0|greptime.semantic.table_scope=global|| diff --git a/tests/cases/standalone/common/system/information_schema.result b/tests/cases/standalone/common/system/information_schema.result index 8fb02828fea..f39b4d53cbd 100644 --- a/tests/cases/standalone/common/system/information_schema.result +++ b/tests/cases/standalone/common/system/information_schema.result @@ -24,6 +24,7 @@ order by table_schema, table_name; |greptime|information_schema|column_privileges|LOCALTEMPORARY|6|0|0|0|0|0||11|Fixed|0|0|0|DATETIME|DATETIME||utf8_bin|0|greptime.semantic.table_scope=global||Y| |greptime|information_schema|column_statistics|LOCALTEMPORARY|7|0|0|0|0|0||11|Fixed|0|0|0|DATETIME|DATETIME||utf8_bin|0|greptime.semantic.table_scope=global||Y| |greptime|information_schema|columns|LOCALTEMPORARY|4|0|0|0|0|0||11|Fixed|0|0|0|DATETIME|DATETIME||utf8_bin|0|greptime.semantic.table_scope=catalog||Y| +|greptime|information_schema|compaction_tasks|LOCALTEMPORARY|50|0|0|0|0|0||11|Fixed|0|0|0|DATETIME|DATETIME||utf8_bin|0|greptime.semantic.table_scope=global||Y| |greptime|information_schema|engines|LOCALTEMPORARY|5|0|0|0|0|0||11|Fixed|0|0|0|DATETIME|DATETIME||utf8_bin|0|greptime.semantic.table_scope=public_global||Y| |greptime|information_schema|events|LOCALTEMPORARY|13|0|0|0|0|0||11|Fixed|0|0|0|DATETIME|DATETIME||utf8_bin|0|greptime.semantic.table_scope=global||Y| |greptime|information_schema|files|LOCALTEMPORARY|14|0|0|0|0|0||11|Fixed|0|0|0|DATETIME|DATETIME||utf8_bin|0|greptime.semantic.table_scope=global||Y| @@ -138,6 +139,25 @@ order by table_schema, table_name, column_name; | greptime | information_schema | columns | table_catalog | 1 | 2147483647 | 2147483647 | | | | utf8 | utf8_bin | | | select,insert | | String | string | FIELD | | NO | string | | | | greptime | information_schema | columns | table_name | 3 | 2147483647 | 2147483647 | | | | utf8 | utf8_bin | | | select,insert | | String | string | FIELD | | NO | string | | | | greptime | information_schema | columns | table_schema | 2 | 2147483647 | 2147483647 | | | | utf8 | utf8_bin | | | select,insert | | String | string | FIELD | | NO | string | | | +| greptime | information_schema | compaction_tasks | completed_units | 13 | | | 20 | 0 | | | | | | select,insert | | UInt64 | bigint unsigned | FIELD | | YES | bigint unsigned | | | +| greptime | information_schema | compaction_tasks | effective_options | 15 | 2147483647 | 2147483647 | | | | utf8 | utf8_bin | | | select,insert | | String | string | FIELD | | YES | string | | | +| greptime | information_schema | compaction_tasks | elapsed_millis | 9 | | | 20 | 0 | | | | | | select,insert | | UInt64 | bigint unsigned | FIELD | | NO | bigint unsigned | | | +| greptime | information_schema | compaction_tasks | input_bytes | 11 | | | 20 | 0 | | | | | | select,insert | | UInt64 | bigint unsigned | FIELD | | YES | bigint unsigned | | | +| greptime | information_schema | compaction_tasks | input_files | 10 | | | 20 | 0 | | | | | | select,insert | | UInt64 | bigint unsigned | FIELD | | YES | bigint unsigned | | | +| greptime | information_schema | compaction_tasks | input_time_end | 17 | | | | | 9 | | | | | select,insert | | TimestampNanosecond | timestamp(9) | FIELD | | YES | timestamp(9) | | | +| greptime | information_schema | compaction_tasks | input_time_start | 16 | | | | | 9 | | | | | select,insert | | TimestampNanosecond | timestamp(9) | FIELD | | YES | timestamp(9) | | | +| greptime | information_schema | compaction_tasks | node_id | 4 | | | 20 | 0 | | | | | | select,insert | | UInt64 | bigint unsigned | FIELD | | YES | bigint unsigned | | | +| greptime | information_schema | compaction_tasks | output_time_end | 19 | | | | | 9 | | | | | select,insert | | TimestampNanosecond | timestamp(9) | FIELD | | YES | timestamp(9) | | | +| greptime | information_schema | compaction_tasks | output_time_start | 18 | | | | | 9 | | | | | select,insert | | TimestampNanosecond | timestamp(9) | FIELD | | YES | timestamp(9) | | | +| greptime | information_schema | compaction_tasks | phase | 7 | 2147483647 | 2147483647 | | | | utf8 | utf8_bin | | | select,insert | | String | string | FIELD | | NO | string | | | +| greptime | information_schema | compaction_tasks | region_id | 2 | | | 20 | 0 | | | | | | select,insert | | UInt64 | bigint unsigned | FIELD | | NO | bigint unsigned | | | +| greptime | information_schema | compaction_tasks | request_options | 14 | 2147483647 | 2147483647 | | | | utf8 | utf8_bin | | | select,insert | | String | string | FIELD | | NO | string | | | +| greptime | information_schema | compaction_tasks | state | 6 | 2147483647 | 2147483647 | | | | utf8 | utf8_bin | | | select,insert | | String | string | FIELD | | NO | string | | | +| greptime | information_schema | compaction_tasks | submitted_at | 8 | | | | | 3 | | | | | select,insert | | TimestampMillisecond | timestamp(3) | FIELD | | NO | timestamp(3) | | | +| greptime | information_schema | compaction_tasks | table_id | 3 | | | 10 | 0 | | | | | | select,insert | | UInt32 | int unsigned | FIELD | | NO | int unsigned | | | +| greptime | information_schema | compaction_tasks | task_id | 1 | 2147483647 | 2147483647 | | | | utf8 | utf8_bin | | | select,insert | | String | string | FIELD | | NO | string | | | +| greptime | information_schema | compaction_tasks | total_units | 12 | | | 20 | 0 | | | | | | select,insert | | UInt64 | bigint unsigned | FIELD | | YES | bigint unsigned | | | +| greptime | information_schema | compaction_tasks | trigger_type | 5 | 2147483647 | 2147483647 | | | | utf8 | utf8_bin | | | select,insert | | String | string | FIELD | | NO | string | | | | greptime | information_schema | engines | comment | 3 | 2147483647 | 2147483647 | | | | utf8 | utf8_bin | | | select,insert | | String | string | FIELD | | NO | string | | | | greptime | information_schema | engines | engine | 1 | 2147483647 | 2147483647 | | | | utf8 | utf8_bin | | | select,insert | | String | string | FIELD | | NO | string | | | | greptime | information_schema | engines | savepoints | 6 | 2147483647 | 2147483647 | | | | utf8 | utf8_bin | | | select,insert | | String | string | FIELD | | NO | string | | | diff --git a/tests/cases/standalone/common/view/create.result b/tests/cases/standalone/common/view/create.result index e8c53070189..ee2e984bb27 100644 --- a/tests/cases/standalone/common/view/create.result +++ b/tests/cases/standalone/common/view/create.result @@ -105,6 +105,7 @@ ORDER BY TABLE_NAME, TABLE_TYPE; |greptime|information_schema|column_privileges|LOCALTEMPORARY|ID|ID|ID|ID|ID|ID||ID|Fixed|ID|ID|ID|DATETIME|DATETIME||utf8_bin|ID|greptime.semantic.table_scope=global||Y| |greptime|information_schema|column_statistics|LOCALTEMPORARY|ID|ID|ID|ID|ID|ID||ID|Fixed|ID|ID|ID|DATETIME|DATETIME||utf8_bin|ID|greptime.semantic.table_scope=global||Y| |greptime|information_schema|columns|LOCALTEMPORARY|ID|ID|ID|ID|ID|ID||ID|Fixed|ID|ID|ID|DATETIME|DATETIME||utf8_bin|ID|greptime.semantic.table_scope=catalog||Y| +|greptime|information_schema|compaction_tasks|LOCALTEMPORARY|ID|ID|ID|ID|ID|ID||ID|Fixed|ID|ID|ID|DATETIME|DATETIME||utf8_bin|ID|greptime.semantic.table_scope=global||Y| |greptime|information_schema|engines|LOCALTEMPORARY|ID|ID|ID|ID|ID|ID||ID|Fixed|ID|ID|ID|DATETIME|DATETIME||utf8_bin|ID|greptime.semantic.table_scope=public_global||Y| |greptime|information_schema|events|LOCALTEMPORARY|ID|ID|ID|ID|ID|ID||ID|Fixed|ID|ID|ID|DATETIME|DATETIME||utf8_bin|ID|greptime.semantic.table_scope=global||Y| |greptime|information_schema|files|LOCALTEMPORARY|ID|ID|ID|ID|ID|ID||ID|Fixed|ID|ID|ID|DATETIME|DATETIME||utf8_bin|ID|greptime.semantic.table_scope=global||Y|