feat: add information_schema.compaction_tasks

Expose live compaction tasks through the datanode inspection path with
node pruning, lazy projected batches, and residual SQL filtering. Reuse
inspection request and plan helpers across compaction and SST tables.

Cover routing, projection, predicates, and lifecycle behavior in datanode,
standalone and distributed integration, and SQL tests. Use the production
information extension in the standalone test fixture.

Signed-off-by: WenyXu <wenymedia@gmail.com>

Cargo.lock

@@ -4518,6 +4518,7 @@ version = "1.4.0-alpha.0"
 dependencies = [
  "api",
  "arrow-flight",
+ "async-stream",
  "async-trait",
  "bytes",
  "cache",

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<SendableRecordBatchStream, crate::error::Error> {
+    ) -> crate::error::Result<SendableRecordBatchStream> {
+        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<NodeInfo>,
+    ) -> crate::error::Result<SendableRecordBatchStream> {
+        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<SendableRecordBatchStream> {
-        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<NodeInfo>,
+    ) -> crate::error::Result<SendableRecordBatchStream> {
+        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<NodeInfo>) -> Vec<NodeInfo> {
+    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<NodeInfo> {
+        [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<Expr> {
+        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::<usize>(),
+                    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);
+            }
         }
     }
 

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<LogicalPlan, DataFusionError> {
         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 {

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<dyn CatalogManager>,
+}
+
+impl InformationSchemaCompactionTasks {
+    pub(crate) fn new(catalog_manager: Weak<dyn CatalogManager>) -> 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<SendableRecordBatchStream> {
+        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<Option<Arc<dyn ExecutionPlan>>> {
+        utils::inspect_plan(
+            &self.catalog_manager,
+            DatanodeInspectKind::CompactionTasks,
+            &self.schema,
+            request,
+        )
+    }
+}

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<SchemaRef> {
-    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<dyn CatalogManager>,
-    kind: DatanodeInspectKind,
-    schema: &SchemaRef,
-    request: ScanRequest,
-) -> Result<Option<Arc<dyn ExecutionPlan>>> {
-    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<DatanodeInspectRequest> {
-    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<SendableRecordBatchStream> {
-        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<Option<Arc<dyn ExecutionPlan>>> {
-        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<SendableRecordBatchStream> {
-        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<Option<Arc<dyn ExecutionPlan>>> {
-        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<SendableRecordBatchStream> {
-        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<Option<Arc<dyn ExecutionPlan>>> {
-        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

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";

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<SchemaRef> {
+    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<dyn CatalogManager>,
+    kind: DatanodeInspectKind,
+    schema: &SchemaRef,
+    request: ScanRequest,
+) -> Result<Option<Arc<dyn ExecutionPlan>>> {
+    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<DatanodeInspectRequest> {
+    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

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 -----
 

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

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<Self> {
+        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<Arc<dyn TableProvider>> {
         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<Arc<dyn TableProvider>> {
+        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<Arc<dyn TableProvider>> {
         let mito = {
@@ -353,6 +375,7 @@ impl TreeNodeRewriter for NameAwareDataSourceInjector {
 
 #[cfg(test)]
 mod tests {
+    mod compaction;
     mod manifest;
 
     use std::sync::Arc;

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<TaskContext>) -> 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<dyn InspectSource>,
@@ -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<Vec<Expr>>,
+            candidates: Arc<AtomicUsize>,
+        }
+
+        impl InspectSource for RecordingSource {
+            fn schema(&self) -> SchemaRef {
+                self.inner.schema()
+            }
+
+            fn scan(
+                &self,
+                request: &InspectScan,
+                ctx: Arc<TaskContext>,
+            ) -> 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::<usize>(),
+                expected_count,
+                "{predicate}"
+            );
+        }
+    }
 }

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<MergeGate>);
+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
+    );
+}

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
 

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(),

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::<Vec<_>>();
     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<Peer>;
+
+        async fn select(
+            &self,
+            _ctx: &Self::Context,
+            _opts: SelectorOptions<'_>,
+        ) -> MetaResult<Self::Output> {
+            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<Instance>,
+    mut region_server: RegionServer,
+    node_id: u64,
+    other_node_id: Option<u64>,
+) {
+    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<CompactionListener>);
+
+    impl Drop for Release {
+        fn drop(&mut self) {
+            self.0.wake();
+        }
+    }
+
+    async fn rows(frontend: &Arc<Instance>, 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::<usize>(),
+        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::<datatypes::arrow::array::TimestampNanosecondArray>()
+            .unwrap();
+        assert_eq!(values.iter().collect::<Vec<_>>(), 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::<usize>(), 1);
+    let snapshot = snapshot.iter().find(|batch| batch.num_rows() > 0).unwrap();
+    assert_eq!(
+        snapshot.column(0).as_string::<i32>().value(0),
+        live_tasks[0].task_id.as_ref()
+    );
+    assert_eq!(snapshot.column(1).as_string::<i32>().value(0), "manual");
+    assert_eq!(snapshot.column(2).as_string::<i32>().value(0), "running");
+    assert_eq!(
+        snapshot
+            .column(3)
+            .as_any()
+            .downcast_ref::<UInt64Array>()
+            .unwrap()
+            .value(0),
+        2
+    );
+    let request: serde_json::Value =
+        serde_json::from_str(snapshot.column(4).as_string::<i32>().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::<i32>().value(0)).unwrap();
+    assert_eq!(effective["time_window_seconds"], 3600);
+    assert_eq!(
+        snapshot
+            .column(6)
+            .as_any()
+            .downcast_ref::<UInt64Array>()
+            .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::<usize>(),
+        0
+    );
+    let nodes = rows(
+        frontend,
+        "SELECT node_id FROM information_schema.compaction_tasks",
+    )
+    .await;
+    assert_eq!(nodes.iter().map(|batch| batch.num_rows()).sum::<usize>(), 1);
+    assert_eq!(
+        nodes
+            .iter()
+            .find(|batch| batch.num_rows() > 0)
+            .unwrap()
+            .column(0)
+            .as_any()
+            .downcast_ref::<UInt64Array>()
+            .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::<usize>(), 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::<usize>(),
+        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<dyn MockInstance>) {
+    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::<usize>(),
+            0,
+            "{sql}"
+        );
+    }
+}

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        |
++----------+
+

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;

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||

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              |                |        |

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|