refactor: model distributed inspect fan-in as exec (#8447)

* refactor: model distributed inspect fan-in as exec Signed-off-by: jeremyhi <fengjiachun@gmail.com> * fix: harden distributed inspect exec fan-in Signed-off-by: jeremyhi <fengjiachun@gmail.com> * fix: show distributed inspect scan in plans Signed-off-by: jeremyhi <fengjiachun@gmail.com> * fix: stabilize distributed inspect sqlness Signed-off-by: jeremyhi <fengjiachun@gmail.com> * fix: apply sqlness replacement to inspect explain Signed-off-by: jeremyhi <fengjiachun@gmail.com> * fix: redact inspect repartition plan Signed-off-by: jeremyhi <fengjiachun@gmail.com> --------- Signed-off-by: jeremyhi <fengjiachun@gmail.com>
2026-10-29 15:06:59 +00:00 · 2026-07-10 08:53:23 +00:00
parent 29423f8ad8
commit 03b304522f
10 changed files with 463 additions and 51 deletions
@@ -12,6 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.

+use std::sync::Arc;
+
 use api::v1::meta::ProcedureStatus;
 use common_error::ext::BoxedError;
 use common_meta::cluster::{ClusterInfo, NodeInfo, Role};
@@ -22,8 +24,16 @@ use common_meta::procedure_executor::{ExecutorContext, ProcedureExecutor};
 use common_meta::rpc::procedure;
 use common_procedure::{ProcedureInfo, ProcedureState};
 use common_query::request::QueryRequest;
-use common_recordbatch::SendableRecordBatchStream;
+use common_recordbatch::adapter::{AsyncRecordBatchStreamAdapter, DfRecordBatchStreamAdapter};
 use common_recordbatch::util::{ChainedRecordBatchStream, LimitedRecordBatchStream};
+use common_recordbatch::{DfSendableRecordBatchStream, SendableRecordBatchStream};
+use datafusion::execution::TaskContext;
+use datafusion::physical_expr::{EquivalenceProperties, Partitioning};
+use datafusion::physical_plan::execution_plan::{Boundedness, EmissionType};
+use datafusion::physical_plan::{DisplayAs, DisplayFormatType, ExecutionPlan, PlanProperties};
+use datatypes::arrow::datatypes::SchemaRef as ArrowSchemaRef;
+use datatypes::schema::SchemaRef;
+use futures_util::future::try_join_all;
 use meta_client::MetaClientRef;
 use snafu::ResultExt;
 use store_api::storage::RegionId;
@@ -43,6 +53,168 @@ impl DistributedInformationExtension {
            datanode_manager,
        }
    }
+
+    async fn inspect_datanode_stream(
+        meta_client: MetaClientRef,
+        datanode_manager: DatanodeManagerRef,
+        request: DatanodeInspectRequest,
+    ) -> std::result::Result<SendableRecordBatchStream, crate::error::Error> {
+        let nodes = meta_client
+            .list_nodes(Some(Role::Datanode))
+            .await
+            .map_err(BoxedError::new)
+            .context(crate::error::ListNodesSnafu)?;
+
+        let limit = request.scan.limit;
+        let plan = request
+            .build_plan()
+            .context(crate::error::DatafusionSnafu)?;
+
+        let streams = try_join_all(nodes.into_iter().map(|node| {
+            let datanode_manager = datanode_manager.clone();
+            let plan = plan.clone();
+            async move {
+                let client = datanode_manager.datanode(&node.peer).await;
+                client
+                    .handle_query(QueryRequest {
+                        plan,
+                        region_id: RegionId::default(),
+                        header: None,
+                    })
+                    .await
+                    .context(crate::error::HandleQuerySnafu)
+            }
+        }))
+        .await?;
+
+        let chained =
+            ChainedRecordBatchStream::new(streams).context(crate::error::CreateRecordBatchSnafu)?;
+        match limit {
+            Some(limit) => Ok(Box::pin(LimitedRecordBatchStream::new(
+                Box::pin(chained),
+                limit,
+            ))),
+            None => Ok(Box::pin(chained)),
+        }
+    }
+}
+
+struct DistributedInspectExec {
+    request: DatanodeInspectRequest,
+    meta_client: MetaClientRef,
+    datanode_manager: DatanodeManagerRef,
+    schema: SchemaRef,
+    arrow_schema: ArrowSchemaRef,
+    properties: Arc<PlanProperties>,
+}
+
+impl std::fmt::Debug for DistributedInspectExec {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("DistributedInspectExec")
+            .field("request", &self.request)
+            .field("schema", &self.schema)
+            .finish_non_exhaustive()
+    }
+}
+
+impl DistributedInspectExec {
+    fn new(
+        request: DatanodeInspectRequest,
+        schema: SchemaRef,
+        meta_client: MetaClientRef,
+        datanode_manager: DatanodeManagerRef,
+    ) -> Self {
+        let arrow_schema = schema.arrow_schema().clone();
+        let properties = Arc::new(PlanProperties::new(
+            EquivalenceProperties::new(arrow_schema.clone()),
+            Partitioning::UnknownPartitioning(1),
+            EmissionType::Incremental,
+            Boundedness::Bounded,
+        ));
+        Self {
+            request,
+            meta_client,
+            datanode_manager,
+            schema,
+            arrow_schema,
+            properties,
+        }
+    }
+}
+
+impl DisplayAs for DistributedInspectExec {
+    fn fmt_as(&self, _t: DisplayFormatType, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(
+            f,
+            "DistributedInspectExec: kind={:?}, scan={}, schema={:?}",
+            self.request.kind, self.request.scan, self.arrow_schema
+        )
+    }
+}
+
+impl ExecutionPlan for DistributedInspectExec {
+    fn name(&self) -> &str {
+        "DistributedInspectExec"
+    }
+
+    fn as_any(&self) -> &dyn std::any::Any {
+        self
+    }
+
+    fn schema(&self) -> ArrowSchemaRef {
+        self.arrow_schema.clone()
+    }
+
+    fn properties(&self) -> &Arc<PlanProperties> {
+        &self.properties
+    }
+
+    fn children(&self) -> Vec<&Arc<dyn ExecutionPlan>> {
+        vec![]
+    }
+
+    fn with_new_children(
+        self: Arc<Self>,
+        children: Vec<Arc<dyn ExecutionPlan>>,
+    ) -> datafusion::error::Result<Arc<dyn ExecutionPlan>> {
+        if children.is_empty() {
+            Ok(self)
+        } else {
+            Err(datafusion::error::DataFusionError::Internal(
+                "DistributedInspectExec is a leaf execution plan and cannot accept children"
+                    .to_string(),
+            ))
+        }
+    }
+
+    fn execute(
+        &self,
+        partition: usize,
+        _context: Arc<TaskContext>,
+    ) -> datafusion::error::Result<DfSendableRecordBatchStream> {
+        if partition != 0 {
+            return Err(datafusion::error::DataFusionError::Execution(format!(
+                "DistributedInspectExec invalid partition. Expected 0, got {partition}"
+            )));
+        }
+
+        let meta_client = self.meta_client.clone();
+        let datanode_manager = self.datanode_manager.clone();
+        let request = self.request.clone();
+        let schema = self.schema.clone();
+        let future = async move {
+            DistributedInformationExtension::inspect_datanode_stream(
+                meta_client,
+                datanode_manager,
+                request,
+            )
+            .await
+            .map_err(BoxedError::new)
+            .context(common_recordbatch::error::ExternalSnafu)
+        };
+        let stream = AsyncRecordBatchStreamAdapter::new(schema, Box::pin(future));
+        Ok(Box::pin(DfRecordBatchStreamAdapter::new(Box::pin(stream))))
+    }
 }

 #[async_trait::async_trait]
@@ -112,41 +284,24 @@ impl InformationExtension for DistributedInformationExtension {
        &self,
        request: DatanodeInspectRequest,
    ) -> std::result::Result<SendableRecordBatchStream, Self::Error> {
-        // Aggregate results from all datanodes
-        let nodes = self
-            .meta_client
-            .list_nodes(Some(Role::Datanode))
-            .await
-            .map_err(BoxedError::new)
-            .context(crate::error::ListNodesSnafu)?;
+        Self::inspect_datanode_stream(
+            self.meta_client.clone(),
+            self.datanode_manager.clone(),
+            request,
+        )
+        .await
+    }

-        let limit = request.scan.limit;
-        let plan = request
-            .build_plan()
-            .context(crate::error::DatafusionSnafu)?;
-
-        let mut streams = Vec::with_capacity(nodes.len());
-        for node in nodes {
-            let client = self.datanode_manager.datanode(&node.peer).await;
-            let stream = client
-                .handle_query(QueryRequest {
-                    plan: plan.clone(),
-                    region_id: RegionId::default(),
-                    header: None,
-                })
-                .await
-                .context(crate::error::HandleQuerySnafu)?;
-            streams.push(stream);
-        }
-
-        let chained =
-            ChainedRecordBatchStream::new(streams).context(crate::error::CreateRecordBatchSnafu)?;
-        match limit {
-            Some(limit) => Ok(Box::pin(LimitedRecordBatchStream::new(
-                Box::pin(chained),
-                limit,
-            ))),
-            None => Ok(Box::pin(chained)),
-        }
+    fn inspect_datanode_plan(
+        &self,
+        request: DatanodeInspectRequest,
+        schema: SchemaRef,
+    ) -> std::result::Result<Option<Arc<dyn ExecutionPlan>>, Self::Error> {
+        Ok(Some(Arc::new(DistributedInspectExec::new(
+            request,
+            schema,
+            self.meta_client.clone(),
+            self.datanode_manager.clone(),
+        ))))
    }
 }
@@ -25,6 +25,7 @@ use std::sync::Arc;
 use common_error::ext::BoxedError;
 use common_recordbatch::{RecordBatchStreamWrapper, SendableRecordBatchStream};
 use common_telemetry::tracing::Span;
+use datafusion::physical_plan::ExecutionPlan;
 use datatypes::schema::SchemaRef;
 use futures_util::StreamExt;
 use snafu::ResultExt;
@@ -107,6 +108,10 @@ pub trait SystemTable {

    fn to_stream(&self, request: ScanRequest) -> Result<SendableRecordBatchStream>;

+    fn scan_plan(&self, _request: ScanRequest) -> Result<Option<Arc<dyn ExecutionPlan>>> {
+        Ok(None)
+    }
+
    fn table_type(&self) -> TableType {
        TableType::Temporary
    }
@@ -173,4 +178,15 @@ impl DataSource for SystemTableDataSource {

        Ok(Box::pin(stream))
    }
+
+    fn get_physical_plan(
+        &self,
+        request: ScanRequest,
+    ) -> std::result::Result<Option<Arc<dyn ExecutionPlan>>, BoxedError> {
+        self.table
+            .scan_plan(request)
+            .map_err(BoxedError::new)
+            .context(TablesRecordBatchSnafu)
+            .map_err(BoxedError::new)
+    }
 }
@@ -46,6 +46,7 @@ use common_procedure::ProcedureInfo;
 use common_recordbatch::SendableRecordBatchStream;
 use datafusion::error::DataFusionError;
 use datafusion::logical_expr::LogicalPlan;
+use datafusion::physical_plan::ExecutionPlan;
 use datatypes::schema::SchemaRef;
 use lazy_static::lazy_static;
 use paste::paste;
@@ -420,6 +421,10 @@ pub trait InformationTable {

    fn to_stream(&self, request: ScanRequest) -> Result<SendableRecordBatchStream>;

+    fn scan_plan(&self, _request: ScanRequest) -> Result<Option<Arc<dyn ExecutionPlan>>> {
+        Ok(None)
+    }
+
    fn table_type(&self) -> TableType {
        TableType::Temporary
    }
@@ -449,6 +454,10 @@ where
    fn to_stream(&self, request: ScanRequest) -> Result<SendableRecordBatchStream> {
        InformationTable::to_stream(self, request)
    }
+
+    fn scan_plan(&self, request: ScanRequest) -> Result<Option<Arc<dyn ExecutionPlan>>> {
+        InformationTable::scan_plan(self, request)
+    }
 }

 pub type InformationExtensionRef = Arc<dyn InformationExtension<Error = Error> + Send + Sync>;
@@ -475,6 +484,16 @@ pub trait InformationExtension {
        &self,
        request: DatanodeInspectRequest,
    ) -> std::result::Result<SendableRecordBatchStream, Self::Error>;
+
+    /// Builds a physical plan for datanode inspect if the extension can expose
+    /// the distributed fan-in semantics to DataFusion.
+    fn inspect_datanode_plan(
+        &self,
+        _request: DatanodeInspectRequest,
+        _schema: SchemaRef,
+    ) -> std::result::Result<Option<Arc<dyn ExecutionPlan>>, Self::Error> {
+        Ok(None)
+    }
 }

 /// The request to inspect the datanode.
@@ -18,6 +18,7 @@ use common_catalog::consts::INFORMATION_SCHEMA_REGION_INFO_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::region_info::RegionInfoEntry;
@@ -83,4 +84,20 @@ impl InformationTable for InformationSchemaRegionInfo {
            Box::pin(future),
        )))
    }
+
+    fn scan_plan(&self, request: ScanRequest) -> Result<Option<Arc<dyn ExecutionPlan>>> {
+        let schema = if let Some(p) = request.projection_indices() {
+            Arc::new(self.schema.try_project(p).context(ProjectSchemaSnafu)?)
+        } else {
+            self.schema.clone()
+        };
+
+        let info_ext = utils::information_extension(&self.catalog_manager)?;
+        let req = DatanodeInspectRequest {
+            kind: DatanodeInspectKind::RegionInfo,
+            scan: request,
+        };
+
+        info_ext.inspect_datanode_plan(req, schema)
+    }
 }
@@ -21,6 +21,7 @@ use common_catalog::consts::{
 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::sst_entry::{ManifestSstEntry, PuffinIndexMetaEntry, StorageSstEntry};
@@ -34,6 +35,30 @@ use crate::information_schema::{
 };
 use crate::system_schema::utils;

+fn projected_schema(schema: &SchemaRef, request: &ScanRequest) -> Result<SchemaRef> {
+    if let Some(p) = request.projection_indices() {
+        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 = DatanodeInspectRequest {
+        kind,
+        scan: request,
+    };
+
+    info_ext.inspect_datanode_plan(req, projected_schema)
+}
+
 /// Information schema table for sst manifest.
 pub struct InformationSchemaSstsManifest {
    schema: SchemaRef,
@@ -63,11 +88,7 @@ impl InformationTable for InformationSchemaSstsManifest {
    }

    fn to_stream(&self, request: ScanRequest) -> Result<SendableRecordBatchStream> {
-        let schema = if let Some(p) = request.projection_indices() {
-            Arc::new(self.schema.try_project(p).context(ProjectSchemaSnafu)?)
-        } else {
-            self.schema.clone()
-        };
+        let schema = projected_schema(&self.schema, &request)?;
        let info_ext = utils::information_extension(&self.catalog_manager)?;
        let req = DatanodeInspectRequest {
            kind: DatanodeInspectKind::SstManifest,
@@ -86,6 +107,15 @@ impl InformationTable for InformationSchemaSstsManifest {
            Box::pin(future),
        )))
    }
+
+    fn scan_plan(&self, request: ScanRequest) -> Result<Option<Arc<dyn ExecutionPlan>>> {
+        inspect_plan(
+            &self.catalog_manager,
+            DatanodeInspectKind::SstManifest,
+            &self.schema,
+            request,
+        )
+    }
 }

 /// Information schema table for sst storage.
@@ -117,11 +147,7 @@ impl InformationTable for InformationSchemaSstsStorage {
    }

    fn to_stream(&self, request: ScanRequest) -> Result<SendableRecordBatchStream> {
-        let schema = if let Some(p) = request.projection_indices() {
-            Arc::new(self.schema.try_project(p).context(ProjectSchemaSnafu)?)
-        } else {
-            self.schema.clone()
-        };
+        let schema = projected_schema(&self.schema, &request)?;

        let info_ext = utils::information_extension(&self.catalog_manager)?;
        let req = DatanodeInspectRequest {
@@ -141,6 +167,15 @@ impl InformationTable for InformationSchemaSstsStorage {
            Box::pin(future),
        )))
    }
+
+    fn scan_plan(&self, request: ScanRequest) -> Result<Option<Arc<dyn ExecutionPlan>>> {
+        inspect_plan(
+            &self.catalog_manager,
+            DatanodeInspectKind::SstStorage,
+            &self.schema,
+            request,
+        )
+    }
 }

 /// Information schema table for index metadata.
@@ -172,11 +207,7 @@ impl InformationTable for InformationSchemaSstsIndexMeta {
    }

    fn to_stream(&self, request: ScanRequest) -> Result<SendableRecordBatchStream> {
-        let schema = if let Some(p) = request.projection_indices() {
-            Arc::new(self.schema.try_project(p).context(ProjectSchemaSnafu)?)
-        } else {
-            self.schema.clone()
-        };
+        let schema = projected_schema(&self.schema, &request)?;

        let info_ext = utils::information_extension(&self.catalog_manager)?;
        let req = DatanodeInspectRequest {
@@ -196,4 +227,116 @@ impl InformationTable for InformationSchemaSstsIndexMeta {
            Box::pin(future),
        )))
    }
+
+    fn scan_plan(&self, request: ScanRequest) -> Result<Option<Arc<dyn ExecutionPlan>>> {
+        inspect_plan(
+            &self.catalog_manager,
+            DatanodeInspectKind::SstIndexMeta,
+            &self.schema,
+            request,
+        )
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    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};
+    use common_meta::cluster::NodeInfo;
+    use common_meta::datanode::RegionStat;
+    use common_meta::key::flow::flow_state::FlowStat;
+    use common_meta::kv_backend::memory::MemoryKvBackend;
+    use common_procedure::ProcedureInfo;
+    use common_recordbatch::{RecordBatches, SendableRecordBatchStream};
+    use datafusion::physical_plan::empty::EmptyExec;
+
+    use super::*;
+    use crate::error::Error;
+    use crate::information_schema::InformationExtension;
+    use crate::kvbackend::KvBackendCatalogManagerBuilder;
+
+    struct PlanInformationExtension;
+
+    #[async_trait::async_trait]
+    impl InformationExtension for PlanInformationExtension {
+        type Error = Error;
+
+        async fn nodes(&self) -> std::result::Result<Vec<NodeInfo>, Self::Error> {
+            Ok(vec![])
+        }
+
+        async fn procedures(
+            &self,
+        ) -> std::result::Result<Vec<(String, ProcedureInfo)>, Self::Error> {
+            Ok(vec![])
+        }
+
+        async fn region_stats(&self) -> std::result::Result<Vec<RegionStat>, Self::Error> {
+            Ok(vec![])
+        }
+
+        async fn flow_stats(&self) -> std::result::Result<Option<FlowStat>, Self::Error> {
+            Ok(None)
+        }
+
+        async fn inspect_datanode(
+            &self,
+            _request: DatanodeInspectRequest,
+        ) -> std::result::Result<SendableRecordBatchStream, Self::Error> {
+            Ok(RecordBatches::empty().as_stream())
+        }
+
+        fn inspect_datanode_plan(
+            &self,
+            _request: DatanodeInspectRequest,
+            schema: SchemaRef,
+        ) -> std::result::Result<Option<Arc<dyn ExecutionPlan>>, Self::Error> {
+            Ok(Some(Arc::new(EmptyExec::new(
+                schema.arrow_schema().clone(),
+            ))))
+        }
+    }
+
+    #[tokio::test]
+    async fn ssts_manifest_uses_information_extension_scan_plan() {
+        let backend = Arc::new(MemoryKvBackend::default());
+        let layered_cache_builder = LayeredCacheRegistryBuilder::default()
+            .add_cache_registry(CacheRegistryBuilder::default().build());
+        let fundamental_cache_registry = build_fundamental_cache_registry(backend.clone());
+        let layered_cache_registry = Arc::new(
+            with_default_composite_cache_registry(
+                layered_cache_builder.add_cache_registry(fundamental_cache_registry),
+            )
+            .unwrap()
+            .build(),
+        );
+
+        let catalog_manager = KvBackendCatalogManagerBuilder::new(
+            Arc::new(PlanInformationExtension),
+            backend,
+            layered_cache_registry,
+        )
+        .build();
+
+        let table = catalog_manager
+            .table(
+                DEFAULT_CATALOG_NAME,
+                INFORMATION_SCHEMA_NAME,
+                SSTS_MANIFEST,
+                None,
+            )
+            .await
+            .unwrap()
+            .unwrap();
+
+        let request = ScanRequest {
+            projection_input: Some(vec![0].into()),
+            ..Default::default()
+        };
+        let plan = table.scan_to_plan(request).unwrap().unwrap();
+
+        assert!(plan.as_any().is::<EmptyExec>());
+        assert_eq!(1, plan.schema().fields().len());
+    }
 }
@@ -16,6 +16,7 @@ use std::sync::Arc;

 use common_error::ext::BoxedError;
 use common_recordbatch::SendableRecordBatchStream;
+use datafusion_physical_plan::ExecutionPlan;

 use crate::storage::ScanRequest;

@@ -24,6 +25,15 @@ use crate::storage::ScanRequest;
 pub trait DataSource {
    /// Retrieves a stream of record batches based on the provided scan request.
    fn get_stream(&self, request: ScanRequest) -> Result<SendableRecordBatchStream, BoxedError>;
+
+    /// Builds a physical plan for the scan request if the data source needs to
+    /// expose its execution semantics to DataFusion.
+    fn get_physical_plan(
+        &self,
+        _request: ScanRequest,
+    ) -> Result<Option<Arc<dyn ExecutionPlan>>, BoxedError> {
+        Ok(None)
+    }
 }

 pub type DataSourceRef = Arc<dyn DataSource + Send + Sync>;
@@ -19,6 +19,7 @@ use common_recordbatch::SendableRecordBatchStream;
 use datafusion::execution::FunctionRegistry;
 use datafusion::logical_expr::Cast;
 use datafusion::logical_expr::expr::ScalarFunction;
+use datafusion::physical_plan::ExecutionPlan;
 use datafusion::prelude::SessionContext;
 use datafusion_expr::expr::Expr;
 use datatypes::data_type::DataType;
@@ -142,6 +143,12 @@ impl Table {
            .context(TablesRecordBatchSnafu)
    }

+    pub fn scan_to_plan(&self, request: ScanRequest) -> Result<Option<Arc<dyn ExecutionPlan>>> {
+        self.data_source
+            .get_physical_plan(request)
+            .context(TablesRecordBatchSnafu)
+    }
+
    /// Tests whether the table provider can make use of any or all filter expressions
    /// to optimise data retrieval.
    pub fn supports_filters_pushdown(&self, filters: &[&Expr]) -> Result<Vec<FilterPushDownType>> {
@@ -113,6 +113,11 @@ impl TableProvider for DfTableProviderAdapter {
            request.limit = limit;
            request.clone()
        };
+
+        if let Some(plan) = self.table.scan_to_plan(request.clone())? {
+            return Ok(plan);
+        }
+
        let stream = self.table.scan_to_stream(request).await?;

        // build sort physical expr
@@ -63,6 +63,37 @@ FROM (
 | 1                     |
 +-----------------------+

+-- SQLNESS REPLACE (RepartitionExec:.*) RepartitionExec: REDACTED
+EXPLAIN SELECT COUNT(*) AS filtered_limited_rows
+FROM (
+  SELECT region_id
+  FROM information_schema.ssts_manifest
+  WHERE table_id > 0
+  LIMIT 1
+);
+
+---------------+-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
+| plan_type     | plan                                                                                                                                                                                                                                                                                            |
+---------------+-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
+| logical_plan  | MergeScan [is_placeholder=false, remote_input=[                                                                                                                                                                                                                                                 |
+|               | Projection: count(Int64(1)) AS count(*) AS filtered_limited_rows                                                                                                                                                                                                                                |
+|               |   Aggregate: groupBy=[[]], aggr=[[count(Int64(1))]]                                                                                                                                                                                                                                             |
+|               |     Limit: skip=0, fetch=1                                                                                                                                                                                                                                                                      |
+|               |       Projection: information_schema.ssts_manifest.region_id                                                                                                                                                                                                                                    |
+|               |         Filter: information_schema.ssts_manifest.table_id > UInt32(0)                                                                                                                                                                                                                           |
+|               |           TableScan: information_schema.ssts_manifest, partial_filters=[information_schema.ssts_manifest.table_id > UInt32(0)]                                                                                                                                                                  |
+|               | ]]                                                                                                                                                                                                                                                                                              |
+| physical_plan | ProjectionExec: expr=[count(Int64(1))@0 as filtered_limited_rows]                                                                                                                                                                                                                               |
+|               |   AggregateExec: mode=Final, gby=[], aggr=[count(Int64(1))]                                                                                                                                                                                                                                     |
+|               |     CoalescePartitionsExec                                                                                                                                                                                                                                                                      |
+|               |       AggregateExec: mode=Partial, gby=[], aggr=[count(Int64(1))]                                                                                                                                                                                                                               |
+|               |         CoalescePartitionsExec: fetch=1                                                                                                                                                                                                                                                         |
+|               |           FilterExec: table_id@0 > 0, projection=[], fetch=1                                                                                                                                                                                                                                    |
+|               |             RepartitionExec: REDACTED
+|               |               DistributedInspectExec: kind=SstManifest, scan=ScanRequest { projection: ProjectionInput { projection: [2], nested_paths: [] }, filters: [table_id > UInt32(0)] }, schema=Schema { fields: [Field { name: "table_id", data_type: UInt32 }], metadata: {"greptime:version": "0"} } |
+|               |                                                                                                                                                                                                                                                                                                 |
+---------------+-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
+
 DROP TABLE ssts_limit_case;

 Affected Rows: 0
@@ -35,4 +35,13 @@ FROM (
  LIMIT 1
 );

+-- SQLNESS REPLACE (RepartitionExec:.*) RepartitionExec: REDACTED
+EXPLAIN SELECT COUNT(*) AS filtered_limited_rows
+FROM (
+  SELECT region_id
+  FROM information_schema.ssts_manifest
+  WHERE table_id > 0
+  LIMIT 1
+);
+
 DROP TABLE ssts_limit_case;