From 03b304522f41adba6f12af871419279a9ce7b79a Mon Sep 17 00:00:00 2001 From: jeremyhi Date: Fri, 10 Jul 2026 16:53:23 +0800 Subject: [PATCH] refactor: model distributed inspect fan-in as exec (#8447) * refactor: model distributed inspect fan-in as exec Signed-off-by: jeremyhi * fix: harden distributed inspect exec fan-in Signed-off-by: jeremyhi * fix: show distributed inspect scan in plans Signed-off-by: jeremyhi * fix: stabilize distributed inspect sqlness Signed-off-by: jeremyhi * fix: apply sqlness replacement to inspect explain Signed-off-by: jeremyhi * fix: redact inspect repartition plan Signed-off-by: jeremyhi --------- Signed-off-by: jeremyhi --- src/catalog/src/information_extension.rs | 227 +++++++++++++++--- src/catalog/src/system_schema.rs | 16 ++ .../src/system_schema/information_schema.rs | 19 ++ .../information_schema/region_info.rs | 17 ++ .../system_schema/information_schema/ssts.rs | 173 +++++++++++-- src/store-api/src/data_source.rs | 10 + src/table/src/table.rs | 7 + src/table/src/table/adapter.rs | 5 + .../information_schema/ssts_limit.result | 31 +++ .../information_schema/ssts_limit.sql | 9 + 10 files changed, 463 insertions(+), 51 deletions(-) diff --git a/src/catalog/src/information_extension.rs b/src/catalog/src/information_extension.rs index 08d81ac1f03..c41c816c6d7 100644 --- a/src/catalog/src/information_extension.rs +++ b/src/catalog/src/information_extension.rs @@ -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 { + 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, +} + +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 { + &self.properties + } + + fn children(&self) -> Vec<&Arc> { + vec![] + } + + fn with_new_children( + self: Arc, + children: Vec>, + ) -> datafusion::error::Result> { + 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, + ) -> datafusion::error::Result { + 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 { - // 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>, Self::Error> { + Ok(Some(Arc::new(DistributedInspectExec::new( + request, + schema, + self.meta_client.clone(), + self.datanode_manager.clone(), + )))) } } diff --git a/src/catalog/src/system_schema.rs b/src/catalog/src/system_schema.rs index 6c6a5fc2028..fe926d893c9 100644 --- a/src/catalog/src/system_schema.rs +++ b/src/catalog/src/system_schema.rs @@ -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; + fn scan_plan(&self, _request: ScanRequest) -> Result>> { + 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>, BoxedError> { + self.table + .scan_plan(request) + .map_err(BoxedError::new) + .context(TablesRecordBatchSnafu) + .map_err(BoxedError::new) + } } diff --git a/src/catalog/src/system_schema/information_schema.rs b/src/catalog/src/system_schema/information_schema.rs index 01e366da886..0a0864aed85 100644 --- a/src/catalog/src/system_schema/information_schema.rs +++ b/src/catalog/src/system_schema/information_schema.rs @@ -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; + fn scan_plan(&self, _request: ScanRequest) -> Result>> { + Ok(None) + } + fn table_type(&self) -> TableType { TableType::Temporary } @@ -449,6 +454,10 @@ where fn to_stream(&self, request: ScanRequest) -> Result { InformationTable::to_stream(self, request) } + + fn scan_plan(&self, request: ScanRequest) -> Result>> { + InformationTable::scan_plan(self, request) + } } pub type InformationExtensionRef = Arc + Send + Sync>; @@ -475,6 +484,16 @@ pub trait InformationExtension { &self, request: DatanodeInspectRequest, ) -> std::result::Result; + + /// 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>, Self::Error> { + Ok(None) + } } /// The request to inspect the datanode. diff --git a/src/catalog/src/system_schema/information_schema/region_info.rs b/src/catalog/src/system_schema/information_schema/region_info.rs index ffc9dfc7ae7..56dc04b85c8 100644 --- a/src/catalog/src/system_schema/information_schema/region_info.rs +++ b/src/catalog/src/system_schema/information_schema/region_info.rs @@ -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>> { + 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) + } } diff --git a/src/catalog/src/system_schema/information_schema/ssts.rs b/src/catalog/src/system_schema/information_schema/ssts.rs index 520fc6d485d..19c44aee343 100644 --- a/src/catalog/src/system_schema/information_schema/ssts.rs +++ b/src/catalog/src/system_schema/information_schema/ssts.rs @@ -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 { + 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, + kind: DatanodeInspectKind, + schema: &SchemaRef, + request: ScanRequest, +) -> Result>> { + 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 { - 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>> { + 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 { - 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>> { + 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 { - 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>> { + 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, Self::Error> { + Ok(vec![]) + } + + async fn procedures( + &self, + ) -> std::result::Result, Self::Error> { + Ok(vec![]) + } + + async fn region_stats(&self) -> std::result::Result, Self::Error> { + Ok(vec![]) + } + + async fn flow_stats(&self) -> std::result::Result, Self::Error> { + Ok(None) + } + + async fn inspect_datanode( + &self, + _request: DatanodeInspectRequest, + ) -> std::result::Result { + Ok(RecordBatches::empty().as_stream()) + } + + fn inspect_datanode_plan( + &self, + _request: DatanodeInspectRequest, + schema: SchemaRef, + ) -> std::result::Result>, 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::()); + assert_eq!(1, plan.schema().fields().len()); + } } diff --git a/src/store-api/src/data_source.rs b/src/store-api/src/data_source.rs index 3fbef8e08c8..2e047cad6c0 100644 --- a/src/store-api/src/data_source.rs +++ b/src/store-api/src/data_source.rs @@ -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; + + /// 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>, BoxedError> { + Ok(None) + } } pub type DataSourceRef = Arc; diff --git a/src/table/src/table.rs b/src/table/src/table.rs index 0ae7d580d8a..542a5066771 100644 --- a/src/table/src/table.rs +++ b/src/table/src/table.rs @@ -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>> { + 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> { diff --git a/src/table/src/table/adapter.rs b/src/table/src/table/adapter.rs index 339c82dddeb..8907140c312 100644 --- a/src/table/src/table/adapter.rs +++ b/src/table/src/table/adapter.rs @@ -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 diff --git a/tests/cases/distributed/information_schema/ssts_limit.result b/tests/cases/distributed/information_schema/ssts_limit.result index 98a19842995..0b64a3c0fa0 100644 --- a/tests/cases/distributed/information_schema/ssts_limit.result +++ b/tests/cases/distributed/information_schema/ssts_limit.result @@ -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 diff --git a/tests/cases/distributed/information_schema/ssts_limit.sql b/tests/cases/distributed/information_schema/ssts_limit.sql index c02f31d63d7..d548bd283a5 100644 --- a/tests/cases/distributed/information_schema/ssts_limit.sql +++ b/tests/cases/distributed/information_schema/ssts_limit.sql @@ -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;