From 32e8467cda0704d79cfae7bca3d0698d24540cee Mon Sep 17 00:00:00 2001 From: discord9 Date: Sat, 10 Oct 2026 11:18:46 +0000 Subject: [PATCH] fix(query): stop reporting region scans as hash partitioned (#9491) * fix(query): stop reporting region scans as hash partitioned Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * test(query): use current DataFusion plan downcasts Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * test(query): match null rendering in join regression Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * test(query): refresh union plan for unknown scan partitioning Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * test(query): refresh shared plans after distribution rule removal Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * test(query): cover semi joins with remote dynamic filtering Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * test(query): strengthen partitioned merge scan regressions Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * test(query): remove unused merge scan test helpers Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * test(perf): measure merge scan hash repartition cost Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * fix(query): adapt merge scan deserialization to two-arg constructor Adapt the production and test serializer callers added by #9206 and remove the obsolete partition_cols assertion. Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * test(query): adapt nested merge scan integration constructor Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> --------- Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> --- src/query/src/dist_plan/analyzer.rs | 9 +- src/query/src/dist_plan/analyzer/fallback.rs | 72 +--- src/query/src/dist_plan/merge_scan.rs | 221 +---------- src/query/src/dist_plan/planner.rs | 1 - src/query/src/metrics.rs | 3 +- src/query/src/optimizer.rs | 1 - .../src/optimizer/json_schema_concretize.rs | 3 +- src/query/src/optimizer/pass_distribution.rs | 362 ------------------ .../src/query_engine/default_serializer.rs | 13 +- src/query/src/query_engine/state.rs | 9 +- tests-integration/src/tests/instance_test.rs | 284 ++++++++++++++ .../nested_merge_scan_capability_test.rs | 2 +- .../distributed/explain/subqueries.result | 3 +- .../common/join/merge_scan_left_join.result | 148 +++++++ .../common/join/merge_scan_left_join.sql | 81 ++++ .../common/join/merge_scan_semi_join.result | 104 +++++ .../common/join/merge_scan_semi_join.sql | 49 +++ .../common/order/order_by_exceptions.result | 2 +- .../common/tql-explain-analyze/explain.result | 3 - .../standalone/common/tql/tql-cte.result | 2 +- .../mergescan_shuffle_9456/case.toml | 152 ++++++++ 21 files changed, 851 insertions(+), 673 deletions(-) delete mode 100644 src/query/src/optimizer/pass_distribution.rs create mode 100644 tests/cases/standalone/common/join/merge_scan_left_join.result create mode 100644 tests/cases/standalone/common/join/merge_scan_left_join.sql create mode 100644 tests/cases/standalone/common/join/merge_scan_semi_join.result create mode 100644 tests/cases/standalone/common/join/merge_scan_semi_join.sql create mode 100644 tests/perf/query_cases/mergescan_shuffle_9456/case.toml diff --git a/src/query/src/dist_plan/analyzer.rs b/src/query/src/dist_plan/analyzer.rs index da26d88b433..38d386cc818 100644 --- a/src/query/src/dist_plan/analyzer.rs +++ b/src/query/src/dist_plan/analyzer.rs @@ -768,14 +768,7 @@ impl PlanRewriter { ); // add merge scan as the new root - let mut node = MergeScanLogicalPlan::new( - on_node.clone(), - false, - // at this stage, the partition cols should be set - // treat it as non-partitioned if None - self.partition_cols.clone().unwrap_or_default(), - ) - .into_logical_plan(); + let mut node = MergeScanLogicalPlan::new(on_node.clone(), false).into_logical_plan(); // expand stages for new_stage in self.stage.drain(..) { diff --git a/src/query/src/dist_plan/analyzer/fallback.rs b/src/query/src/dist_plan/analyzer/fallback.rs index 86c7f50628d..0aabdd6cfd8 100644 --- a/src/query/src/dist_plan/analyzer/fallback.rs +++ b/src/query/src/dist_plan/analyzer/fallback.rs @@ -17,18 +17,11 @@ //! This is a temporary solution, and will be removed once we have a more robust plan rewriter //! -use std::collections::BTreeSet; - -use common_telemetry::debug; -use datafusion::datasource::DefaultTableSource; use datafusion_common::Result as DfResult; use datafusion_common::tree_node::{Transformed, TreeNodeRewriter}; use datafusion_expr::LogicalPlan; -use table::metadata::TableType; -use table::table::adapter::DfTableProviderAdapter; use crate::dist_plan::MergeScanLogicalPlan; -use crate::dist_plan::analyzer::{AliasMapping, OTHER_PHY_PART_COL_PLACEHOLDER}; /// FallbackPlanRewriter is a plan rewriter that will only push down table scan node /// This is used when `PlanRewriter` produce errors when trying to rewrite the plan @@ -44,69 +37,8 @@ impl TreeNodeRewriter for FallbackPlanRewriter { &mut self, plan: Self::Node, ) -> DfResult> { - if let LogicalPlan::TableScan(table_scan) = &plan { - let partition_cols = if let Some(source) = - table_scan.source.downcast_ref::() - { - if let Some(provider) = source - .table_provider - .downcast_ref::() - { - if provider.table().table_type() == TableType::Base { - let info = provider.table().table_info(); - let partition_key_indices = info.meta.partition_key_indices.clone(); - let schema = info.meta.schema.clone(); - let partition_cols = partition_key_indices - .iter() - .map(|index| schema.column_name_by_index(*index).to_string()) - .collect::>(); - debug!( - "FallbackPlanRewriter: loaded table partition metadata, table: {}, table_id: {}, partition_key_indices: {:?}, partition_columns: {:?}", - info.name, - info.ident.table_id, - info.meta.partition_key_indices, - partition_cols, - ); - Some(partition_cols - .into_iter() - .map(|c| { - if c == OTHER_PHY_PART_COL_PLACEHOLDER { - // for placeholder, just return a empty alias - return Ok((c.clone(), BTreeSet::new())); - } - let index = - plan.schema().index_of_column_by_name(None, &c).ok_or_else(|| { - datafusion_common::DataFusionError::Internal( - format!( - "PlanRewriter: maybe_set_partitions: column {c} not found in schema of plan: {plan}" - ), - ) - })?; - let column = plan.schema().columns().get(index).cloned().ok_or_else(|| { - datafusion_common::DataFusionError::Internal(format!( - "PlanRewriter: maybe_set_partitions: column index {index} out of bounds in schema of plan: {plan}" - )) - })?; - Ok((c.clone(), BTreeSet::from([column]))) - }) - .collect::>()?) - } else { - None - } - } else { - None - } - } else { - None - }; - let node = MergeScanLogicalPlan::new( - plan, - false, - // at this stage, the partition cols should be set - // treat it as non-partitioned if None - partition_cols.clone().unwrap_or_default(), - ) - .into_logical_plan(); + if matches!(plan, LogicalPlan::TableScan(_)) { + let node = MergeScanLogicalPlan::new(plan, false).into_logical_plan(); Ok(Transformed::yes(node)) } else { Ok(Transformed::no(plan)) diff --git a/src/query/src/dist_plan/merge_scan.rs b/src/query/src/dist_plan/merge_scan.rs index ee4e9574280..041fea160cf 100644 --- a/src/query/src/dist_plan/merge_scan.rs +++ b/src/query/src/dist_plan/merge_scan.rs @@ -17,7 +17,7 @@ use std::cell::Cell; use std::sync::{Arc, Mutex}; use std::time::Duration; -use ahash::{HashMap, HashSet}; +use ahash::HashMap; use arrow_schema::{ ArrowError, DataType as ArrowDataType, Field, Schema as ArrowSchema, SchemaRef as ArrowSchemaRef, SortOptions, @@ -44,10 +44,9 @@ use datafusion::physical_plan::{ }; use datafusion_common::stats::Precision; use datafusion_common::tree_node::TreeNodeRecursion; -use datafusion_common::{Column as ColumnExpr, DFSchemaRef, DataFusionError, Result, Statistics}; +use datafusion_common::{DFSchemaRef, DataFusionError, Result, Statistics}; use datafusion_expr::{Expr, Extension, FetchType, LogicalPlan, UserDefinedLogicalNodeCore}; -use datafusion_physical_expr::expressions::Column; -use datafusion_physical_expr::{Distribution, EquivalenceProperties, PhysicalSortExpr}; +use datafusion_physical_expr::{EquivalenceProperties, PhysicalSortExpr}; use datatypes::extension::json::is_any_json_extension_type; use futures_util::StreamExt; use greptime_proto::v1::region::RegionRequestHeader; @@ -64,7 +63,6 @@ use tokio::time; use tokio::time::Instant; use tracing::{Instrument, Span}; -use crate::dist_plan::analyzer::AliasMapping; use crate::dist_plan::analyzer::utils::patch_batch_timezone; use crate::dist_plan::dyn_filter_bridge::{ CapturedDynFilter, capture_remote_dyn_filters_for_pushdown, @@ -339,7 +337,6 @@ pub struct MergeScanLogicalPlan { output_schema: DFSchemaRef, /// If this plan is a placeholder is_placeholder: bool, - partition_cols: AliasMapping, /// Assigned after dist-plan rewriting so rewriters only deal with plan shape. remote_dyn_filter_producer_id: Option, } @@ -350,29 +347,20 @@ impl PartialOrd for MergeScanLogicalPlan { input, output_schema, is_placeholder, - partition_cols, remote_dyn_filter_producer_id, } = self; let Self { input: other_input, output_schema: other_output_schema, is_placeholder: other_is_placeholder, - partition_cols: other_partition_cols, remote_dyn_filter_producer_id: other_remote_dyn_filter_producer_id, } = other; - let ordering = ( - input, - is_placeholder, - partition_cols, - remote_dyn_filter_producer_id, - ) - .partial_cmp(&( - other_input, - other_is_placeholder, - other_partition_cols, - other_remote_dyn_filter_producer_id, - )); + let ordering = (input, is_placeholder, remote_dyn_filter_producer_id).partial_cmp(&( + other_input, + other_is_placeholder, + other_remote_dyn_filter_producer_id, + )); match ordering { Some(std::cmp::Ordering::Equal) if output_schema != other_output_schema => None, ordering => ordering, @@ -419,12 +407,11 @@ impl UserDefinedLogicalNodeCore for MergeScanLogicalPlan { impl MergeScanLogicalPlan { /// Creates a merge scan with the input plan's schema. - pub fn new(input: LogicalPlan, is_placeholder: bool, partition_cols: AliasMapping) -> Self { + pub fn new(input: LogicalPlan, is_placeholder: bool) -> Self { Self { output_schema: input.schema().clone(), input, is_placeholder, - partition_cols, remote_dyn_filter_producer_id: None, } } @@ -462,10 +449,6 @@ impl MergeScanLogicalPlan { &self.input } - pub fn partition_cols(&self) -> &AliasMapping { - &self.partition_cols - } - pub fn remote_dyn_filter_producer_id(&self) -> Option { self.remote_dyn_filter_producer_id } @@ -489,7 +472,6 @@ pub struct MergeScanExec { remote_dyn_filter_producer_id: Option, captured_remote_dyn_filters: Arc>>, target_partition: usize, - partition_cols: AliasMapping, enable_per_region_metrics: bool, } @@ -514,7 +496,6 @@ impl MergeScanExec { region_query_handler: RegionQueryHandlerRef, query_ctx: QueryContextRef, target_partition: usize, - partition_cols: AliasMapping, remote_dyn_filter_producer_id: Option, enable_per_region_metrics: bool, ) -> Result { @@ -551,24 +532,8 @@ impl MergeScanExec { EquivalenceProperties::new(arrow_schema.clone()) }; - let partition_exprs = partition_cols - .iter() - .filter_map(|col| { - if let Some(first_alias) = col.1.first() { - session_state - .create_physical_expr( - Expr::Column(ColumnExpr::new_unqualified( - first_alias.name().to_string(), - )), - plan.schema(), - ) - .ok() - } else { - None - } - }) - .collect(); - let partitioning = Partitioning::Hash(partition_exprs, output_partition_count); + // Region stripes are positional, not hash buckets; do not advertise a hash distribution. + let partitioning = Partitioning::UnknownPartitioning(output_partition_count); let properties = Arc::new(PlanProperties::new( eq_properties, @@ -590,7 +555,6 @@ impl MergeScanExec { remote_dyn_filter_producer_id, captured_remote_dyn_filters: Arc::default(), target_partition, - partition_cols, enable_per_region_metrics, }) } @@ -912,78 +876,6 @@ impl MergeScanExec { ))) } - pub fn try_with_new_distribution(&self, distribution: Distribution) -> Option { - let Distribution::KeyPartitioned(hash_exprs) = distribution else { - // not applicable - return None; - }; - - if let Partitioning::Hash(curr_dist, _) = &self.properties.partitioning - && curr_dist == &hash_exprs - { - // No need to change the distribution - return None; - } - - let hash_expr_col_names: HashSet<_> = hash_exprs - .iter() - .filter_map(|expr| { - expr.downcast_ref::() - .map(|col_expr| col_expr.name()) - }) - .collect(); - - let covers_all_partition_cols = self.partition_cols.values().all(|aliases| { - aliases - .iter() - .any(|col| hash_expr_col_names.contains(col.name())) - }); - if !covers_all_partition_cols { - return None; - } - - let all_partition_col_aliases: HashSet<_> = self - .partition_cols - .values() - .flat_map(|aliases| aliases.iter().map(|c| c.name())) - .collect(); - let overlaps: Vec<_> = hash_exprs - .iter() - .filter(|expr| { - expr.downcast_ref::() - .is_some_and(|col_expr| all_partition_col_aliases.contains(col_expr.name())) - }) - .cloned() - .collect(); - - if overlaps.is_empty() { - return None; - } - - Some(Self { - table: self.table.clone(), - regions: self.regions.clone(), - plan: self.plan.clone(), - arrow_schema: self.arrow_schema.clone(), - region_query_handler: self.region_query_handler.clone(), - metric: self.metric.clone(), - properties: Arc::new(PlanProperties::new( - self.properties.eq_properties.clone(), - Partitioning::Hash(overlaps, self.partition_count()), - self.properties.emission_type, - self.properties.boundedness, - )), - sub_stage_metrics: self.sub_stage_metrics.clone(), - partition_metrics: self.partition_metrics.clone(), - query_ctx: self.query_ctx.clone(), - remote_dyn_filter_producer_id: self.remote_dyn_filter_producer_id, - captured_remote_dyn_filters: self.captured_remote_dyn_filters.clone(), - target_partition: self.target_partition, - partition_cols: self.partition_cols.clone(), - enable_per_region_metrics: self.enable_per_region_metrics, - }) - } - fn captured_remote_dyn_filters(&self) -> Vec { self.captured_remote_dyn_filters.lock().unwrap().clone() } @@ -1037,13 +929,6 @@ impl MergeScanExec { } } -#[cfg(test)] -impl MergeScanExec { - fn remote_dyn_filter_producer_id(&self) -> Option { - self.remote_dyn_filter_producer_id - } -} - /// Metrics for a region of a partition. #[derive(Debug, Clone)] struct RegionMetrics { @@ -1420,7 +1305,7 @@ impl MergeScanMetric { #[cfg(test)] mod tests { - use std::collections::{BTreeSet, HashMap as StdHashMap}; + use std::collections::HashMap as StdHashMap; use std::pin::Pin; use std::sync::atomic::{AtomicUsize, Ordering}; use std::task::{Context, Poll}; @@ -1441,15 +1326,15 @@ mod tests { }; use datafusion::config::ConfigOptions; use datafusion::execution::SessionStateBuilder; + use datafusion::physical_plan::Partitioning; use datafusion::physical_plan::filter_pushdown::ChildFilterPushdownResult; use datafusion::physical_plan::repartition::RepartitionExec; use datafusion::physical_plan::{StatisticsArgs, StatisticsContext}; - use datafusion_common::TableReference; use datafusion_expr::{LogicalPlanBuilder, col, lit}; + use datafusion_physical_expr::PhysicalExpr; use datafusion_physical_expr::expressions::{ Column, DynamicFilterPhysicalExpr, lit as physical_lit, }; - use datafusion_physical_expr::{Distribution, PhysicalExpr}; use datatypes::prelude::{ConcreteDataType, VectorRef}; use datatypes::schema::{ColumnSchema, Schema}; use datatypes::vectors::{Int64Vector, StringVector, TimestampMillisecondVector}; @@ -1517,7 +1402,6 @@ mod tests { Arc::new(TestRegionQueryHandler::default()), QueryContext::arc(), target_partition, - AliasMapping::new(), None, false, ) @@ -1560,7 +1444,6 @@ mod tests { handler, query_ctx, 1, - AliasMapping::new(), Some(RemoteDynFilterProducerId::new(42)), false, ) @@ -1661,6 +1544,10 @@ mod tests { exec.properties().output_partitioning().partition_count(), expected ); + assert!(matches!( + exec.properties().output_partitioning(), + Partitioning::UnknownPartitioning(count) if *count == expected + )); } } @@ -1874,7 +1761,6 @@ mod tests { handler.clone(), query_ctx, 1, - AliasMapping::new(), Some(RemoteDynFilterProducerId::new(42)), false, ) @@ -2682,7 +2568,6 @@ mod tests { handler, QueryContext::arc(), target_partition, - AliasMapping::new(), None, false, ) @@ -3288,73 +3173,6 @@ mod tests { assert_int64_batch(&batches[0], (11, 12)); } - #[test] - fn try_with_new_distribution_preserves_remote_dyn_filter_producer_id() { - let remote_dyn_filter_producer_id = RemoteDynFilterProducerId::new(42); - - // Build a plan whose schema contains "col1" - let plan = LogicalPlanBuilder::empty(true) - .project(vec![lit(1i32).alias("col1")]) - .unwrap() - .build() - .unwrap(); - - let schema = plan.schema().as_arrow().clone(); - let table = TableName::new("catalog", "schema", "table"); - let regions = vec![RegionId::new(1024, 1)]; - let query_ctx = QueryContext::arc(); - - // Non-empty partition_cols so try_with_new_distribution can detect an overlap - let mut partition_cols = AliasMapping::new(); - partition_cols.insert( - "col1".to_string(), - BTreeSet::from([ColumnExpr::new(Some(TableReference::bare("table")), "col1")]), - ); - - let session_state = SessionStateBuilder::new().build(); - - let handler = Arc::new(TestRegionQueryHandler::default()); - let target_partition = 2; - - let exec = MergeScanExec::new( - &session_state, - table, - regions, - plan, - &schema, - handler, - query_ctx, - target_partition, - partition_cols, - Some(remote_dyn_filter_producer_id), - false, - ) - .unwrap(); - - assert_eq!( - exec.remote_dyn_filter_producer_id(), - Some(remote_dyn_filter_producer_id) - ); - - // A distribution that differs from the current partitioning but shares a - // column name present in partition_cols, so try_with_new_distribution - // produces a clone instead of returning None. - let new_dist = Distribution::KeyPartitioned(vec![ - Arc::new(Column::new("col1", 0)), - Arc::new(Column::new("col2", 1)), - ]); - - let cloned = exec - .try_with_new_distribution(new_dist) - .expect("expected a cloned exec with overlapping partition col"); - - assert_eq!( - cloned.remote_dyn_filter_producer_id(), - Some(remote_dyn_filter_producer_id), - "try_with_new_distribution must preserve remote dynamic filter producer id" - ); - } - #[test] fn merge_scan_apply_expressions_exposes_remote_dyn_filter_id() { let query_ctx = QueryContext::arc(); @@ -3397,7 +3215,6 @@ mod tests { handler, query_ctx, 1, - AliasMapping::new(), Some(remote_dyn_filter_producer_id), false, ) @@ -3503,7 +3320,6 @@ mod tests { Arc::new(TestRegionQueryHandler::default()), QueryContext::arc(), 1, - AliasMapping::new(), None, true, ) @@ -3572,7 +3388,6 @@ mod tests { Arc::new(TestRegionQueryHandler::default()), QueryContext::arc(), 1, - AliasMapping::new(), None, true, ) diff --git a/src/query/src/dist_plan/planner.rs b/src/query/src/dist_plan/planner.rs index b2a2ade9b4f..88b13caf388 100644 --- a/src/query/src/dist_plan/planner.rs +++ b/src/query/src/dist_plan/planner.rs @@ -228,7 +228,6 @@ impl ExtensionPlanner for DistExtensionPlanner { self.region_query_handler.clone(), query_ctx, session.config().target_partitions(), - merge_scan.partition_cols().clone(), merge_scan.remote_dyn_filter_producer_id(), self.enable_per_region_metrics, )?; diff --git a/src/query/src/metrics.rs b/src/query/src/metrics.rs index d6d62c99c4f..3c0043dcc09 100644 --- a/src/query/src/metrics.rs +++ b/src/query/src/metrics.rs @@ -407,7 +407,7 @@ fn finalize_region_watermarks(merged: BTreeMap) -> Vec>::new(), Some(RemoteDynFilterProducerId::new(0)), false, ) diff --git a/src/query/src/optimizer.rs b/src/query/src/optimizer.rs index fec001e7506..03d6ce3aa46 100644 --- a/src/query/src/optimizer.rs +++ b/src/query/src/optimizer.rs @@ -22,7 +22,6 @@ pub(crate) mod insert_assignment; pub(crate) mod json_schema_concretize; pub(crate) mod json_type_concretize; pub mod parallelize_scan; -pub mod pass_distribution; pub mod promql_tsid_narrow_join; pub mod remove_duplicate; pub mod scan_hint; diff --git a/src/query/src/optimizer/json_schema_concretize.rs b/src/query/src/optimizer/json_schema_concretize.rs index 12f71d44c31..9ea3531853d 100644 --- a/src/query/src/optimizer/json_schema_concretize.rs +++ b/src/query/src/optimizer/json_schema_concretize.rs @@ -192,8 +192,7 @@ mod tests { produce_one_row: false, schema, }); - let merge_scan = - MergeScanLogicalPlan::new(input, false, Default::default()).into_logical_plan(); + let merge_scan = MergeScanLogicalPlan::new(input, false).into_logical_plan(); let plan = LogicalPlanBuilder::from(merge_scan) .project(vec![col("j")])? .build()?; diff --git a/src/query/src/optimizer/pass_distribution.rs b/src/query/src/optimizer/pass_distribution.rs deleted file mode 100644 index 732168e962f..00000000000 --- a/src/query/src/optimizer/pass_distribution.rs +++ /dev/null @@ -1,362 +0,0 @@ -// 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 datafusion::config::ConfigOptions; -use datafusion::physical_optimizer::PhysicalOptimizerRule; -use datafusion::physical_plan::projection::ProjectionExec; -use datafusion::physical_plan::repartition::RepartitionExec; -use datafusion::physical_plan::{ - ChildrenPropertiesMode, ExecutionPlan, Partitioning, ReplaceChildrenOptions, -}; -use datafusion_common::Result as DfResult; -use datafusion_physical_expr::Distribution; -use datafusion_physical_expr::utils::map_columns_before_projection; - -use crate::dist_plan::MergeScanExec; - -/// This is a [`PhysicalOptimizerRule`] to pass distribution requirement to -/// [`MergeScanExec`] to avoid unnecessary shuffling. -/// -/// This rule is expected to be run before [`EnforceDistribution`]. -/// -/// [`EnforceDistribution`]: datafusion::physical_optimizer::enforce_distribution::EnforceDistribution -/// [`MergeScanExec`]: crate::dist_plan::MergeScanExec -#[derive(Debug)] -pub struct PassDistribution; - -impl PhysicalOptimizerRule for PassDistribution { - fn optimize( - &self, - plan: Arc, - config: &ConfigOptions, - ) -> DfResult> { - Self::do_optimize(plan, config) - } - - fn name(&self) -> &str { - "PassDistributionRule" - } - - fn schema_check(&self) -> bool { - false - } -} - -impl PassDistribution { - fn do_optimize( - plan: Arc, - _config: &ConfigOptions, - ) -> DfResult> { - // Start from root with no requirement - Self::rewrite_with_distribution(plan, None) - } - - /// Top-down rewrite that propagates distribution requirements to children. - fn rewrite_with_distribution( - plan: Arc, - current_req: Option, - ) -> DfResult> { - // If this is a MergeScanExec, try to apply the current requirement. - if let Some(merge_scan) = plan.downcast_ref::() - && let Some(Distribution::KeyPartitioned(hash_exprs)) = current_req.as_ref() - { - if let Partitioning::Hash(current_hash_exprs, _) = &merge_scan.properties().partitioning - && *current_hash_exprs == *hash_exprs - { - return Ok(plan); - } - - if let Some(new_plan) = merge_scan - .try_with_new_distribution(Distribution::KeyPartitioned(hash_exprs.clone())) - { - // Leaf node; no children to process - return Ok(Arc::new(new_plan) as _); - } - - let partitioning = Partitioning::Hash( - hash_exprs.clone(), - merge_scan.properties().partitioning.partition_count(), - ); - return Ok(Arc::new(RepartitionExec::try_new(plan, partitioning)?) as _); - } - - // Compute per-child requirements from the current node. - let children = plan.children(); - if children.is_empty() { - return Ok(plan); - } - - let required = plan.input_distribution_requirements(); - let mut new_children = Vec::with_capacity(children.len()); - for (idx, child) in children.into_iter().enumerate() { - let child_req = match required.child_distribution(idx) { - Some(Distribution::UnspecifiedDistribution) if idx == 0 => { - Self::map_hash_requirement_through_projection(plan.as_ref(), ¤t_req) - } - Some(Distribution::UnspecifiedDistribution) => None, - None => current_req.clone(), - Some(req) => Some(req.clone()), - }; - let new_child = Self::rewrite_with_distribution(child.clone(), child_req)?; - new_children.push(new_child); - } - - // Rebuild the node only if any child changed (pointer inequality) - let unchanged = plan - .children() - .into_iter() - .zip(new_children.iter()) - .all(|(old, new)| Arc::ptr_eq(old, new)); - if unchanged { - Ok(plan) - } else { - plan.replace_children( - new_children, - ReplaceChildrenOptions::new(ChildrenPropertiesMode::Recompute), - ) - } - } - - fn map_hash_requirement_through_projection( - plan: &dyn ExecutionPlan, - current_req: &Option, - ) -> Option { - let Some(Distribution::KeyPartitioned(required_exprs)) = current_req else { - return None; - }; - - let projection = plan.downcast_ref::()?; - let proj_exprs = projection - .expr() - .iter() - .map(|expr| (Arc::clone(&expr.expr), expr.alias.clone())) - .collect::>(); - let mapped = map_columns_before_projection(required_exprs, &proj_exprs); - - (mapped.len() == required_exprs.len()).then_some(Distribution::KeyPartitioned(mapped)) - } -} - -#[cfg(test)] -mod tests { - use std::collections::{BTreeMap, BTreeSet}; - - use api::v1::region::{RemoteDynFilterUnregister, RemoteDynFilterUpdate}; - use arrow_schema::{DataType, Field, Schema, SchemaRef, TimeUnit}; - use async_trait::async_trait; - use common_query::request::QueryRequest; - use datafusion::common::NullEquality; - use datafusion::execution::SessionStateBuilder; - use datafusion::physical_optimizer::PhysicalOptimizerRule; - use datafusion::physical_plan::joins::{HashJoinExec, PartitionMode}; - use datafusion::physical_plan::projection::{ProjectionExec, ProjectionExpr}; - use datafusion::physical_plan::{ExecutionPlanProperties, Partitioning}; - use datafusion_expr::{JoinType, LogicalPlanBuilder}; - use datafusion_physical_expr::PhysicalExpr; - use datafusion_physical_expr::expressions::Column as PhysicalColumn; - use session::ReadPreference; - use session::context::QueryContext; - use store_api::metric_engine_consts::DATA_SCHEMA_TSID_COLUMN_NAME; - use store_api::storage::RegionId; - use table::table_name::TableName; - - use super::*; - use crate::dist_plan::RemoteDynFilterProducerId; - use crate::error::Result as QueryResult; - use crate::region_query::RegionQueryHandler; - - struct NoopRegionQueryHandler; - - #[async_trait] - impl RegionQueryHandler for NoopRegionQueryHandler { - async fn select_target( - &self, - _read_preference: ReadPreference, - _region_id: RegionId, - ) -> QueryResult { - unreachable!("pass distribution tests should not execute remote queries") - } - - async fn do_get( - &self, - _target: &crate::region_query::RegionQueryTarget, - _request: QueryRequest, - ) -> QueryResult { - unreachable!("pass distribution tests should not execute remote queries") - } - - async fn handle_remote_dyn_filter_update( - &self, - _target: &crate::region_query::RegionQueryTarget, - _query_id: String, - _update: RemoteDynFilterUpdate, - ) -> QueryResult<()> { - unreachable!("pass distribution tests should not send remote dyn filter updates") - } - - async fn handle_remote_dyn_filter_unregister( - &self, - _target: &crate::region_query::RegionQueryTarget, - _query_id: String, - _unregister: RemoteDynFilterUnregister, - ) -> QueryResult<()> { - unreachable!("pass distribution tests should not send remote dyn filter unregisters") - } - } - - #[test] - fn passes_hash_requirement_through_projection_to_merge_scan() { - let schema = test_schema(); - let left_merge_scan = Arc::new(test_merge_scan_exec(schema.clone())); - let right_merge_scan = Arc::new(test_merge_scan_exec(schema.clone())); - let left_projection = Arc::new( - ProjectionExec::try_new( - vec![ - ProjectionExpr::new(partition_column("greptime_value", 3), "greptime_value"), - ProjectionExpr::new( - partition_column(DATA_SCHEMA_TSID_COLUMN_NAME, 1), - DATA_SCHEMA_TSID_COLUMN_NAME, - ), - ProjectionExpr::new( - partition_column("greptime_timestamp", 2), - "greptime_timestamp", - ), - ], - left_merge_scan, - ) - .unwrap(), - ) as Arc; - let join = Arc::new( - HashJoinExec::try_new( - left_projection, - right_merge_scan, - vec![ - ( - partition_column(DATA_SCHEMA_TSID_COLUMN_NAME, 1), - partition_column(DATA_SCHEMA_TSID_COLUMN_NAME, 1), - ), - ( - partition_column("greptime_timestamp", 2), - partition_column("greptime_timestamp", 2), - ), - ], - None, - &JoinType::Inner, - None, - PartitionMode::Partitioned, - NullEquality::NullEqualsNull, - false, - ) - .unwrap(), - ) as Arc; - - let optimized = PassDistribution - .optimize(join, &ConfigOptions::default()) - .unwrap(); - let hash_join = optimized.downcast_ref::().unwrap(); - let left_projection = hash_join.left().downcast_ref::().unwrap(); - let left_partitioning = left_projection.input().output_partitioning(); - let right_partitioning = hash_join.right().output_partitioning(); - - let Partitioning::Hash(left_exprs, left_count) = left_partitioning else { - panic!("expected left merge scan hash partitioning"); - }; - let Partitioning::Hash(right_exprs, right_count) = right_partitioning else { - panic!("expected right merge scan hash partitioning"); - }; - - assert_eq!(*left_count, 2); - assert_eq!(*right_count, 2); - assert_eq!( - column_names(left_exprs), - vec![DATA_SCHEMA_TSID_COLUMN_NAME, "greptime_timestamp"] - ); - assert_eq!( - column_names(right_exprs), - vec![DATA_SCHEMA_TSID_COLUMN_NAME, "greptime_timestamp"] - ); - } - - #[test] - fn merge_scan_rejects_hash_requirement_on_partition_key_subset() { - let merge_scan = test_merge_scan_exec(test_schema()); - - let new_plan = merge_scan.try_with_new_distribution(Distribution::KeyPartitioned(vec![ - partition_column(DATA_SCHEMA_TSID_COLUMN_NAME, 1), - ])); - - assert!( - new_plan.is_none(), - "partitioning by a subset of multi-column partition keys is not sufficient" - ); - } - - fn test_merge_scan_exec(schema: SchemaRef) -> MergeScanExec { - let session_state = SessionStateBuilder::new().with_default_features().build(); - let partition_cols = BTreeMap::from([ - ( - DATA_SCHEMA_TSID_COLUMN_NAME.to_string(), - BTreeSet::from([datafusion_common::Column::from_name( - DATA_SCHEMA_TSID_COLUMN_NAME, - )]), - ), - ( - "greptime_timestamp".to_string(), - BTreeSet::from([datafusion_common::Column::from_name("greptime_timestamp")]), - ), - ]); - let plan = LogicalPlanBuilder::empty(false).build().unwrap(); - - MergeScanExec::new( - &session_state, - TableName::new("greptime", "public", "test"), - vec![RegionId::new(1, 0), RegionId::new(1, 1)], - plan, - schema.as_ref(), - Arc::new(NoopRegionQueryHandler), - QueryContext::arc(), - 32, - partition_cols, - Some(RemoteDynFilterProducerId::new(1)), - false, - ) - .unwrap() - } - - fn test_schema() -> SchemaRef { - Arc::new(Schema::new(vec![ - Field::new("host", DataType::Utf8, true), - Field::new(DATA_SCHEMA_TSID_COLUMN_NAME, DataType::UInt64, false), - Field::new( - "greptime_timestamp", - DataType::Timestamp(TimeUnit::Millisecond, None), - false, - ), - Field::new("greptime_value", DataType::Float64, true), - ])) - } - - fn partition_column(name: &str, index: usize) -> Arc { - Arc::new(PhysicalColumn::new(name, index)) - } - - fn column_names(exprs: &[Arc]) -> Vec<&str> { - exprs - .iter() - .map(|expr| expr.downcast_ref::().unwrap().name()) - .collect() - } -} diff --git a/src/query/src/query_engine/default_serializer.rs b/src/query/src/query_engine/default_serializer.rs index c924a57a68e..63ee0856aaf 100644 --- a/src/query/src/query_engine/default_serializer.rs +++ b/src/query/src/query_engine/default_serializer.rs @@ -220,11 +220,7 @@ impl SubstraitConsumer for MergeScanSubstraitConsumer<'_> { let input = self.decode_payload(merge_scan.input).await?; - // `PbMergeScan` lacks `partition_cols`; decoded plans lose this optimization and may repartition. - Ok( - MergeScanLogicalPlan::new(input, merge_scan.is_placeholder, Default::default()) - .into_logical_plan(), - ) + Ok(MergeScanLogicalPlan::new(input, merge_scan.is_placeholder).into_logical_plan()) } // Keep child dispatch on this consumer; delegating the parent bypasses async MergeScan decoding. @@ -739,8 +735,6 @@ mod tests { .expect("Expect a MergeScan plan node"); assert!(merge_scan.is_placeholder()); assert_eq!(merge_scan.input().to_string(), numbers_scan().to_string()); - // `PbMergeScan` doesn't carry `partition_cols`, so decoded nodes have an empty mapping. - assert!(merge_scan.partition_cols().is_empty()); } /// Payload table binding uses the engine catalog, not the request catalog. @@ -979,8 +973,7 @@ mod tests { .unwrap() .build() .unwrap(); - let plan = - MergeScanLogicalPlan::new(input, false, Default::default()).into_logical_plan(); + let plan = MergeScanLogicalPlan::new(input, false).into_logical_plan(); let bytes = DFLogicalSubstraitConvertor .encode(&plan, DefaultSerializer) @@ -1041,7 +1034,7 @@ mod tests { } fn merge_scan(input: LogicalPlan, is_placeholder: bool) -> LogicalPlan { - MergeScanLogicalPlan::new(input, is_placeholder, Default::default()).into_logical_plan() + MergeScanLogicalPlan::new(input, is_placeholder).into_logical_plan() } fn encode_plan(plan: &LogicalPlan) -> Bytes { diff --git a/src/query/src/query_engine/state.rs b/src/query/src/query_engine/state.rs index aa3b9417b14..e6db7db10f2 100644 --- a/src/query/src/query_engine/state.rs +++ b/src/query/src/query_engine/state.rs @@ -70,7 +70,6 @@ use crate::optimizer::global_limit::EnsureGlobalLimitForFetch; use crate::optimizer::json_schema_concretize::JsonSchemaConcretizeRule; use crate::optimizer::json_type_concretize::JsonTypeConcretizeRule; use crate::optimizer::parallelize_scan::ParallelizeScan; -use crate::optimizer::pass_distribution::PassDistribution; use crate::optimizer::promql_tsid_narrow_join::PromqlTsidNarrowJoin; use crate::optimizer::remove_duplicate::RemoveDuplicate; use crate::optimizer::scan_hint::ScanHintRule; @@ -232,17 +231,13 @@ impl QueryEngineState { physical_optimizer .rules .insert(5, Arc::new(ParallelizeScan)); - // Pass distribution requirement to MergeScanExec to avoid unnecessary shuffling - physical_optimizer - .rules - .insert(6, Arc::new(PassDistribution)); // Prefer collecting narrow PromQL build sides over repartitioning wide label streams. physical_optimizer .rules - .insert(7, Arc::new(PromqlTsidNarrowJoin)); + .insert(6, Arc::new(PromqlTsidNarrowJoin)); // Re-enforce sorting after custom rules update scan partitioning and distribution. // Keep it immediately before DataFusion's default EnsureRequirements. - physical_optimizer.rules.insert(8, Arc::new(EnforceSorting)); + physical_optimizer.rules.insert(7, Arc::new(EnforceSorting)); // Add rule for windowed sort physical_optimizer .rules diff --git a/tests-integration/src/tests/instance_test.rs b/tests-integration/src/tests/instance_test.rs index a1ccac27328..89706b53764 100644 --- a/tests-integration/src/tests/instance_test.rs +++ b/tests-integration/src/tests/instance_test.rs @@ -29,6 +29,7 @@ use frontend::error::Error; use frontend::instance::Instance; use operator::error::Error as OperatorError; use query::datafusion::QUERY_PARALLELISM_HINT; +use query::options::QUERY_ENABLE_REMOTE_DYNAMIC_FILTER_PUSHDOWN; use rstest::rstest; use rstest_reuse::apply; use servers::error as server_error; @@ -257,6 +258,289 @@ WHERE greptime_timestamp = ( ); } +#[apply(both_instances_cases)] +async fn test_partitioned_merge_scan_left_join_repartitions(instance: Arc) { + let frontend = instance.frontend(); + execute_sql( + &frontend, + r#" +CREATE TABLE join_repro ( + k STRING, + kind STRING, + v DOUBLE, + ts TIMESTAMP NOT NULL, + TIME INDEX (ts) +) +PARTITION ON COLUMNS (k) ( + k < '1', + k >= '1' AND k < '2', + k >= '2' AND k < '3', + k >= '3' AND k < '4', + k >= '4' AND k < '5', + k >= '5' AND k < '6', + k >= '6' AND k < '7', + k >= '7' AND k < '8', + k >= '8' AND k < '9', + k >= '9' AND k < 'a', + k >= 'a' AND k < 'b', + k >= 'b' AND k < 'c', + k >= 'c' AND k < 'd', + k >= 'd' AND k < 'e', + k >= 'e' AND k < 'f', + k >= 'f' +) +ENGINE = mito +WITH (append_mode = 'true') +"#, + ) + .await; + + let mut rows = Vec::new(); + for key in [ + '0', '1', '2', '3', '4', '5', '6', '7', '8', '9', 'a', 'b', 'c', 'd', 'e', 'f', 'z', + ] { + rows.push(format!("('{key}', 'root', 1.0, '2024-01-01 00:00:00')")); + if key != 'z' { + for _ in 0..3 { + rows.push(format!("('{key}', 'child', 1.0, '2024-01-01 00:00:00')")); + } + } + } + execute_sql( + &frontend, + &format!("INSERT INTO join_repro VALUES {}", rows.join(",")), + ) + .await; + + let aggregate_sql = r#"SELECT count(*) roots, count(c.n) roots_with_children, sum(c.n) children +FROM (SELECT k FROM join_repro WHERE kind = 'root') r +LEFT JOIN (SELECT k, count(*) n FROM join_repro WHERE kind = 'child' GROUP BY k) c ON c.k = r.k"#; + for parallelism in [16, 8] { + let output = execute_sql_with_query_parallelism(&frontend, aggregate_sql, parallelism) + .await + .data + .pretty_print() + .await; + assert_eq!( + output, + "+-------+---------------------+----------+\n| roots | roots_with_children | children |\n+-------+---------------------+----------+\n| 17 | 16 | 48 |\n+-------+---------------------+----------+" + ); + + let explain = execute_sql_with_query_parallelism( + &frontend, + &format!("EXPLAIN {aggregate_sql}"), + parallelism, + ) + .await + .data + .pretty_print() + .await; + assert!( + explain.contains("HashJoinExec: mode=Partitioned"), + "query_parallelism={parallelism} should use a partitioned hash join; explain:\n{explain}" + ); + assert!( + explain.contains("RepartitionExec: partitioning=Hash([k@0]"), + "query_parallelism={parallelism} should repartition on the join key; explain:\n{explain}" + ); + + let ordered_sql = r#"SELECT r.k, c.n +FROM (SELECT k FROM join_repro WHERE kind = 'root') r +LEFT JOIN (SELECT k, count(*) n FROM join_repro WHERE kind = 'child' GROUP BY k) c ON c.k = r.k +ORDER BY r.k"#; + let ordered = execute_sql_with_query_parallelism(&frontend, ordered_sql, parallelism) + .await + .data + .pretty_print() + .await; + assert_eq!( + ordered, + r#"+---+---+ +| k | n | ++---+---+ +| 0 | 3 | +| 1 | 3 | +| 2 | 3 | +| 3 | 3 | +| 4 | 3 | +| 5 | 3 | +| 6 | 3 | +| 7 | 3 | +| 8 | 3 | +| 9 | 3 | +| a | 3 | +| b | 3 | +| c | 3 | +| d | 3 | +| e | 3 | +| f | 3 | +| z | | ++---+---+"# + ); + + let pruned_sql = r#"SELECT count(*) roots, count(c.n) roots_with_children, sum(c.n) children +FROM (SELECT k FROM join_repro WHERE kind = 'root' AND k >= '1' AND k < '9') r +LEFT JOIN (SELECT k, count(*) n FROM join_repro WHERE kind = 'child' AND k < '8' GROUP BY k) c ON c.k = r.k"#; + let pruned = execute_sql_with_query_parallelism(&frontend, pruned_sql, parallelism) + .await + .data + .pretty_print() + .await; + assert_eq!( + pruned, + "+-------+---------------------+----------+\n| roots | roots_with_children | children |\n+-------+---------------------+----------+\n| 8 | 7 | 21 |\n+-------+---------------------+----------+" + ); + + let pruned_explain = execute_sql_with_query_parallelism( + &frontend, + &format!("EXPLAIN {pruned_sql}"), + parallelism, + ) + .await + .data + .pretty_print() + .await; + assert!( + pruned_explain.contains("HashJoinExec: mode=Partitioned"), + "query_parallelism={parallelism} pruned join should remain partitioned; explain:\n{pruned_explain}" + ); + let hash_repartition_lines = pruned_explain + .lines() + .filter(|line| line.contains("RepartitionExec: partitioning=Hash([k@0]")) + .count(); + assert_eq!( + hash_repartition_lines, 2, + "query_parallelism={parallelism} pruned join should repartition both inputs on k; explain:\n{pruned_explain}" + ); + } +} + +#[apply(both_instances_cases)] +async fn test_partitioned_merge_scan_semi_join_repartitions(instance: Arc) { + let frontend = instance.frontend(); + execute_sql( + &frontend, + r#" +CREATE TABLE semi_join_repro ( + ts TIMESTAMP TIME INDEX, + key_id INT, + is_match BOOLEAN, + PRIMARY KEY (key_id) +) +PARTITION ON COLUMNS (key_id) ( + key_id < 10, + key_id >= 10 AND key_id < 20, + key_id >= 20 AND key_id < 30, + key_id >= 30 +) +ENGINE = mito +"#, + ) + .await; + execute_sql( + &frontend, + r#"INSERT INTO semi_join_repro VALUES +('2026-10-09 00:00:00', 1, false), +('2026-10-09 00:00:01', 1, true), +('2026-10-09 00:00:00', 11, false), +('2026-10-09 00:00:01', 11, true), +('2026-10-09 00:00:00', 21, false), +('2026-10-09 00:00:01', 21, true), +('2026-10-09 00:00:00', 31, false), +('2026-10-09 00:00:01', 31, true)"#, + ) + .await; + + let query = r#"SELECT t.key_id, MIN(t.ts) AS first_ts +FROM semi_join_repro t +WHERE t.key_id IN ( + SELECT s.key_id FROM semi_join_repro s WHERE s.is_match = true +) +GROUP BY t.key_id +ORDER BY t.key_id"#; + let mut query_ctx = QueryContext::with_db_name(None); + query_ctx.set_extension(QUERY_PARALLELISM_HINT, "2"); + query_ctx.set_extension(QUERY_ENABLE_REMOTE_DYNAMIC_FILTER_PUSHDOWN, "true"); + let query_ctx = Arc::new(query_ctx); + let output = execute_sql_with(&frontend, query, query_ctx.clone()) + .await + .data + .pretty_print() + .await; + let expected = r#"+--------+---------------------+ +| key_id | first_ts | ++--------+---------------------+ +| 1 | 2026-10-09T00:00:00 | +| 11 | 2026-10-09T00:00:00 | +| 21 | 2026-10-09T00:00:00 | +| 31 | 2026-10-09T00:00:00 | ++--------+---------------------+"#; + assert_eq!(output, expected); + + let explain = execute_sql_with(&frontend, &format!("EXPLAIN {query}"), query_ctx.clone()) + .await + .data + .pretty_print() + .await; + assert!( + explain.contains("HashJoinExec: mode=Partitioned, join_type=LeftSemi"), + "parallel IN query should use a partitioned left-semi join; explain:\n{explain}" + ); + assert!( + explain.contains("RepartitionExec: partitioning=Hash([key_id@1], 2)") + && explain.contains("RepartitionExec: partitioning=Hash([key_id@0], 2)"), + "parallel left-semi join should repartition both inputs on key_id; explain:\n{explain}" + ); + + let analyze_explain = execute_sql_with( + &frontend, + &format!("EXPLAIN ANALYZE VERBOSE {query}"), + query_ctx.clone(), + ) + .await + .data + .pretty_print() + .await; + let has_remote_dyn_filter = |explain: &str| { + explain.lines().any(|line| { + line.contains("SeqScan: region=") + && line.contains("\"dyn_filters\"") + && line.contains("DynamicFilter") + }) + }; + assert!( + has_remote_dyn_filter(&analyze_explain), + "expected remote region SeqScan dyn_filters to include DynamicFilter; explain:\n{analyze_explain}" + ); + + let mut query_ctx_without_rdf = QueryContext::with_db_name(None); + query_ctx_without_rdf.set_extension(QUERY_PARALLELISM_HINT, "2"); + query_ctx_without_rdf.set_extension(QUERY_ENABLE_REMOTE_DYNAMIC_FILTER_PUSHDOWN, "false"); + let query_ctx_without_rdf = Arc::new(query_ctx_without_rdf); + let output_without_rdf = execute_sql_with(&frontend, query, query_ctx_without_rdf.clone()) + .await + .data + .pretty_print() + .await; + assert_eq!(output_without_rdf, expected); + + let analyze_explain_without_rdf = execute_sql_with( + &frontend, + &format!("EXPLAIN ANALYZE VERBOSE {query}"), + query_ctx_without_rdf, + ) + .await + .data + .pretty_print() + .await; + assert!( + !analyze_explain_without_rdf + .lines() + .any(|line| line.contains("SeqScan: region=") && line.contains("\"dyn_filters\"")), + "expected no region SeqScan dyn_filters line when remote dynamic filter pushdown is disabled; explain:\n{analyze_explain_without_rdf}" + ); +} + #[apply(both_instances_cases)] async fn test_show_create_table(instance: Arc) { let frontend = instance.frontend(); diff --git a/tests-integration/src/tests/nested_merge_scan_capability_test.rs b/tests-integration/src/tests/nested_merge_scan_capability_test.rs index 2c86589112d..6ad57b7a14c 100644 --- a/tests-integration/src/tests/nested_merge_scan_capability_test.rs +++ b/tests-integration/src/tests/nested_merge_scan_capability_test.rs @@ -547,7 +547,7 @@ fn nested_plan_of(probe: &LogicalPlan, build: LogicalPlan) -> LogicalPlan { info!("probe side of the nested plan:\n{probe}"); info!("build side of the nested plan:\n{build}"); - let inner = MergeScanLogicalPlan::new(build, false, Default::default()).into_logical_plan(); + let inner = MergeScanLogicalPlan::new(build, false).into_logical_plan(); let plan = LogicalPlanBuilder::from(probe.clone()) .join( inner, diff --git a/tests/cases/distributed/explain/subqueries.result b/tests/cases/distributed/explain/subqueries.result index 28fe75df516..3d3f24777db 100644 --- a/tests/cases/distributed/explain/subqueries.result +++ b/tests/cases/distributed/explain/subqueries.result @@ -517,9 +517,10 @@ EXPLAIN SELECT a FROM ((SELECT a FROM t1) UNION ALL (SELECT a FROM t2)) u ORDER |_|_TableScan: t2_| |_| ]]_| | physical_plan | SortPreservingMergeExec: [a@0 ASC NULLS LAST]_| +|_|_UnionExec_| |_|_SortExec: expr=[a@0 ASC NULLS LAST], preserve_partitioning=[true]_| -|_|_InterleaveExec_| |_|_MergeScanExec: REDACTED +|_|_SortExec: expr=[a@0 ASC NULLS LAST], preserve_partitioning=[true]_| |_|_MergeScanExec: REDACTED |_|_| +-+-+ diff --git a/tests/cases/standalone/common/join/merge_scan_left_join.result b/tests/cases/standalone/common/join/merge_scan_left_join.result new file mode 100644 index 00000000000..b50ff0f5bd5 --- /dev/null +++ b/tests/cases/standalone/common/join/merge_scan_left_join.result @@ -0,0 +1,148 @@ +-- Regression coverage for issues 9456/9492 (PR 9491): preserve merge-scan +-- results for a pruned left join over partitioned append data. +CREATE TABLE merge_left_data ( + k STRING, + kind STRING, + ts TIMESTAMP TIME INDEX, + PRIMARY KEY (k, kind) +) +PARTITION ON COLUMNS (k) ( + k < '1', k >= '1' AND k < '2', k >= '2' AND k < '3', + k >= '3' AND k < '4', k >= '4' AND k < '5', k >= '5' AND k < '6', + k >= '6' AND k < '7', k >= '7' AND k < '8', k >= '8' AND k < '9', + k >= '9' AND k < 'a', k >= 'a' AND k < 'b', k >= 'b' AND k < 'c', + k >= 'c' AND k < 'd', k >= 'd' AND k < 'e', k >= 'e' AND k < 'f', + k >= 'f' +) +ENGINE=mito WITH (append_mode = 'true'); + +Affected Rows: 0 + +INSERT INTO merge_left_data VALUES + ('0', 'root', '2026-10-09 00:00:00'), + ('1', 'root', '2026-10-09 00:00:00'), + ('2', 'root', '2026-10-09 00:00:00'), + ('3', 'root', '2026-10-09 00:00:00'), + ('4', 'root', '2026-10-09 00:00:00'), + ('5', 'root', '2026-10-09 00:00:00'), + ('6', 'root', '2026-10-09 00:00:00'), + ('7', 'root', '2026-10-09 00:00:00'), + ('8', 'root', '2026-10-09 00:00:00'), + ('9', 'root', '2026-10-09 00:00:00'), + ('a', 'root', '2026-10-09 00:00:00'), + ('b', 'root', '2026-10-09 00:00:00'), + ('c', 'root', '2026-10-09 00:00:00'), + ('d', 'root', '2026-10-09 00:00:00'), + ('e', 'root', '2026-10-09 00:00:00'), + ('f', 'root', '2026-10-09 00:00:00'), + ('z', 'root', '2026-10-09 00:00:00'), + ('0', 'child', '2026-10-09 00:00:00'), ('0', 'child', '2026-10-09 00:00:00'), ('0', 'child', '2026-10-09 00:00:00'), + ('1', 'child', '2026-10-09 00:00:00'), ('1', 'child', '2026-10-09 00:00:00'), ('1', 'child', '2026-10-09 00:00:00'), + ('2', 'child', '2026-10-09 00:00:00'), ('2', 'child', '2026-10-09 00:00:00'), ('2', 'child', '2026-10-09 00:00:00'), + ('3', 'child', '2026-10-09 00:00:00'), ('3', 'child', '2026-10-09 00:00:00'), ('3', 'child', '2026-10-09 00:00:00'), + ('4', 'child', '2026-10-09 00:00:00'), ('4', 'child', '2026-10-09 00:00:00'), ('4', 'child', '2026-10-09 00:00:00'), + ('5', 'child', '2026-10-09 00:00:00'), ('5', 'child', '2026-10-09 00:00:00'), ('5', 'child', '2026-10-09 00:00:00'), + ('6', 'child', '2026-10-09 00:00:00'), ('6', 'child', '2026-10-09 00:00:00'), ('6', 'child', '2026-10-09 00:00:00'), + ('7', 'child', '2026-10-09 00:00:00'), ('7', 'child', '2026-10-09 00:00:00'), ('7', 'child', '2026-10-09 00:00:00'), + ('8', 'child', '2026-10-09 00:00:00'), ('8', 'child', '2026-10-09 00:00:00'), ('8', 'child', '2026-10-09 00:00:00'), + ('9', 'child', '2026-10-09 00:00:00'), ('9', 'child', '2026-10-09 00:00:00'), ('9', 'child', '2026-10-09 00:00:00'), + ('a', 'child', '2026-10-09 00:00:00'), ('a', 'child', '2026-10-09 00:00:00'), ('a', 'child', '2026-10-09 00:00:00'), + ('b', 'child', '2026-10-09 00:00:00'), ('b', 'child', '2026-10-09 00:00:00'), ('b', 'child', '2026-10-09 00:00:00'), + ('c', 'child', '2026-10-09 00:00:00'), ('c', 'child', '2026-10-09 00:00:00'), ('c', 'child', '2026-10-09 00:00:00'), + ('d', 'child', '2026-10-09 00:00:00'), ('d', 'child', '2026-10-09 00:00:00'), ('d', 'child', '2026-10-09 00:00:00'), + ('e', 'child', '2026-10-09 00:00:00'), ('e', 'child', '2026-10-09 00:00:00'), ('e', 'child', '2026-10-09 00:00:00'), + ('f', 'child', '2026-10-09 00:00:00'), ('f', 'child', '2026-10-09 00:00:00'), ('f', 'child', '2026-10-09 00:00:00'); + +Affected Rows: 65 + +-- Full join: 17 roots, 16 matched roots, 48 child rows. +SELECT r.k, c.n +FROM (SELECT k FROM merge_left_data WHERE kind = 'root') r +LEFT JOIN ( + SELECT k, COUNT(*) AS n FROM merge_left_data WHERE kind = 'child' GROUP BY k +) c ON r.k = c.k +ORDER BY r.k; + ++---+---+ +| k | n | ++---+---+ +| 0 | 3 | +| 1 | 3 | +| 2 | 3 | +| 3 | 3 | +| 4 | 3 | +| 5 | 3 | +| 6 | 3 | +| 7 | 3 | +| 8 | 3 | +| 9 | 3 | +| a | 3 | +| b | 3 | +| c | 3 | +| d | 3 | +| e | 3 | +| f | 3 | +| z | | ++---+---+ + +-- Assert partitioned hash join and both hash repartition keys at SQLness N=4. +-- SQLNESS REPLACE (peers.*) peers=REDACTED +EXPLAIN SELECT r.k, c.n +FROM (SELECT k FROM merge_left_data WHERE kind = 'root' AND k >= '1' AND k < '5') r +LEFT JOIN ( + SELECT k, COUNT(*) AS n FROM merge_left_data + WHERE kind = 'child' AND k < '4' GROUP BY k +) c ON r.k = c.k +ORDER BY r.k; + ++---------------+---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ +| plan_type | plan | ++---------------+---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ +| logical_plan | Sort: r.k ASC NULLS LAST | +| | Projection: r.k, c.n | +| | Left Join: r.k = c.k | +| | MergeScan [is_placeholder=false, remote_input=[ | +| | SubqueryAlias: r | +| | Projection: merge_left_data.k | +| | Filter: merge_left_data.kind = Utf8("root") AND merge_left_data.k >= Utf8("1") AND merge_left_data.k < Utf8("5") | +| | TableScan: merge_left_data, partial_filters=[merge_left_data.kind = Utf8("root"), merge_left_data.k >= Utf8("1"), merge_left_data.k < Utf8("5")] | +| | ]] | +| | MergeScan [is_placeholder=false, remote_input=[ | +| | SubqueryAlias: c | +| | Projection: merge_left_data.k, count(Int64(1)) AS count(*) AS n | +| | Aggregate: groupBy=[[merge_left_data.k]], aggr=[[count(merge_left_data.ts) AS count(Int64(1))]] | +| | Filter: merge_left_data.kind = Utf8("child") AND merge_left_data.k < Utf8("4") AND merge_left_data.k IS NOT NULL | +| | TableScan: merge_left_data, partial_filters=[merge_left_data.kind = Utf8("child"), merge_left_data.k < Utf8("4"), merge_left_data.k IS NOT NULL] | +| | ]] | +| physical_plan | SortPreservingMergeExec: [k@0 ASC NULLS LAST] | +| | SortExec: expr=[k@0 ASC NULLS LAST], preserve_partitioning=[true] | +| | HashJoinExec: mode=Partitioned, join_type=Left, on=[(k@0, k@0)], projection=[k@0, n@2] | +| | RepartitionExec: partitioning=Hash([k@0], 4), input_partitions=4 | +| | MergeScanExec: peers=REDACTED +| | RepartitionExec: partitioning=Hash([k@0], 4), input_partitions=4 | +| | MergeScanExec: peers=REDACTED +| | | ++---------------+---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ + +-- Four roots, three matches, nine child rows; key 4 remains with NULL n. +SELECT r.k, c.n +FROM (SELECT k FROM merge_left_data WHERE kind = 'root' AND k >= '1' AND k < '5') r +LEFT JOIN ( + SELECT k, COUNT(*) AS n FROM merge_left_data + WHERE kind = 'child' AND k < '4' GROUP BY k +) c ON r.k = c.k +ORDER BY r.k; + ++---+---+ +| k | n | ++---+---+ +| 1 | 3 | +| 2 | 3 | +| 3 | 3 | +| 4 | | ++---+---+ + +DROP TABLE merge_left_data; + +Affected Rows: 0 + diff --git a/tests/cases/standalone/common/join/merge_scan_left_join.sql b/tests/cases/standalone/common/join/merge_scan_left_join.sql new file mode 100644 index 00000000000..72b5f75db7b --- /dev/null +++ b/tests/cases/standalone/common/join/merge_scan_left_join.sql @@ -0,0 +1,81 @@ +-- Regression coverage for issues 9456/9492 (PR 9491): preserve merge-scan +-- results for a pruned left join over partitioned append data. +CREATE TABLE merge_left_data ( + k STRING, + kind STRING, + ts TIMESTAMP TIME INDEX, + PRIMARY KEY (k, kind) +) +PARTITION ON COLUMNS (k) ( + k < '1', k >= '1' AND k < '2', k >= '2' AND k < '3', + k >= '3' AND k < '4', k >= '4' AND k < '5', k >= '5' AND k < '6', + k >= '6' AND k < '7', k >= '7' AND k < '8', k >= '8' AND k < '9', + k >= '9' AND k < 'a', k >= 'a' AND k < 'b', k >= 'b' AND k < 'c', + k >= 'c' AND k < 'd', k >= 'd' AND k < 'e', k >= 'e' AND k < 'f', + k >= 'f' +) +ENGINE=mito WITH (append_mode = 'true'); + +INSERT INTO merge_left_data VALUES + ('0', 'root', '2026-10-09 00:00:00'), + ('1', 'root', '2026-10-09 00:00:00'), + ('2', 'root', '2026-10-09 00:00:00'), + ('3', 'root', '2026-10-09 00:00:00'), + ('4', 'root', '2026-10-09 00:00:00'), + ('5', 'root', '2026-10-09 00:00:00'), + ('6', 'root', '2026-10-09 00:00:00'), + ('7', 'root', '2026-10-09 00:00:00'), + ('8', 'root', '2026-10-09 00:00:00'), + ('9', 'root', '2026-10-09 00:00:00'), + ('a', 'root', '2026-10-09 00:00:00'), + ('b', 'root', '2026-10-09 00:00:00'), + ('c', 'root', '2026-10-09 00:00:00'), + ('d', 'root', '2026-10-09 00:00:00'), + ('e', 'root', '2026-10-09 00:00:00'), + ('f', 'root', '2026-10-09 00:00:00'), + ('z', 'root', '2026-10-09 00:00:00'), + ('0', 'child', '2026-10-09 00:00:00'), ('0', 'child', '2026-10-09 00:00:00'), ('0', 'child', '2026-10-09 00:00:00'), + ('1', 'child', '2026-10-09 00:00:00'), ('1', 'child', '2026-10-09 00:00:00'), ('1', 'child', '2026-10-09 00:00:00'), + ('2', 'child', '2026-10-09 00:00:00'), ('2', 'child', '2026-10-09 00:00:00'), ('2', 'child', '2026-10-09 00:00:00'), + ('3', 'child', '2026-10-09 00:00:00'), ('3', 'child', '2026-10-09 00:00:00'), ('3', 'child', '2026-10-09 00:00:00'), + ('4', 'child', '2026-10-09 00:00:00'), ('4', 'child', '2026-10-09 00:00:00'), ('4', 'child', '2026-10-09 00:00:00'), + ('5', 'child', '2026-10-09 00:00:00'), ('5', 'child', '2026-10-09 00:00:00'), ('5', 'child', '2026-10-09 00:00:00'), + ('6', 'child', '2026-10-09 00:00:00'), ('6', 'child', '2026-10-09 00:00:00'), ('6', 'child', '2026-10-09 00:00:00'), + ('7', 'child', '2026-10-09 00:00:00'), ('7', 'child', '2026-10-09 00:00:00'), ('7', 'child', '2026-10-09 00:00:00'), + ('8', 'child', '2026-10-09 00:00:00'), ('8', 'child', '2026-10-09 00:00:00'), ('8', 'child', '2026-10-09 00:00:00'), + ('9', 'child', '2026-10-09 00:00:00'), ('9', 'child', '2026-10-09 00:00:00'), ('9', 'child', '2026-10-09 00:00:00'), + ('a', 'child', '2026-10-09 00:00:00'), ('a', 'child', '2026-10-09 00:00:00'), ('a', 'child', '2026-10-09 00:00:00'), + ('b', 'child', '2026-10-09 00:00:00'), ('b', 'child', '2026-10-09 00:00:00'), ('b', 'child', '2026-10-09 00:00:00'), + ('c', 'child', '2026-10-09 00:00:00'), ('c', 'child', '2026-10-09 00:00:00'), ('c', 'child', '2026-10-09 00:00:00'), + ('d', 'child', '2026-10-09 00:00:00'), ('d', 'child', '2026-10-09 00:00:00'), ('d', 'child', '2026-10-09 00:00:00'), + ('e', 'child', '2026-10-09 00:00:00'), ('e', 'child', '2026-10-09 00:00:00'), ('e', 'child', '2026-10-09 00:00:00'), + ('f', 'child', '2026-10-09 00:00:00'), ('f', 'child', '2026-10-09 00:00:00'), ('f', 'child', '2026-10-09 00:00:00'); + +-- Full join: 17 roots, 16 matched roots, 48 child rows. +SELECT r.k, c.n +FROM (SELECT k FROM merge_left_data WHERE kind = 'root') r +LEFT JOIN ( + SELECT k, COUNT(*) AS n FROM merge_left_data WHERE kind = 'child' GROUP BY k +) c ON r.k = c.k +ORDER BY r.k; + +-- Assert partitioned hash join and both hash repartition keys at SQLness N=4. +-- SQLNESS REPLACE (peers.*) peers=REDACTED +EXPLAIN SELECT r.k, c.n +FROM (SELECT k FROM merge_left_data WHERE kind = 'root' AND k >= '1' AND k < '5') r +LEFT JOIN ( + SELECT k, COUNT(*) AS n FROM merge_left_data + WHERE kind = 'child' AND k < '4' GROUP BY k +) c ON r.k = c.k +ORDER BY r.k; + +-- Four roots, three matches, nine child rows; key 4 remains with NULL n. +SELECT r.k, c.n +FROM (SELECT k FROM merge_left_data WHERE kind = 'root' AND k >= '1' AND k < '5') r +LEFT JOIN ( + SELECT k, COUNT(*) AS n FROM merge_left_data + WHERE kind = 'child' AND k < '4' GROUP BY k +) c ON r.k = c.k +ORDER BY r.k; + +DROP TABLE merge_left_data; diff --git a/tests/cases/standalone/common/join/merge_scan_semi_join.result b/tests/cases/standalone/common/join/merge_scan_semi_join.result new file mode 100644 index 00000000000..5eb70b1241b --- /dev/null +++ b/tests/cases/standalone/common/join/merge_scan_semi_join.result @@ -0,0 +1,104 @@ +-- Regression coverage for issues 9456/9492 (PR 9491): preserve IN/semi-join +-- results when merge-scan inputs are distributed across hash partitions. +CREATE TABLE merge_semi_data ( + ts TIMESTAMP TIME INDEX, + key_id INT, + is_match BOOLEAN, + PRIMARY KEY (key_id) +) +PARTITION ON COLUMNS (key_id) ( + key_id < 10, + key_id >= 10 AND key_id < 20, + key_id >= 20 AND key_id < 30, + key_id >= 30 +) +ENGINE=mito; + +Affected Rows: 0 + +-- Three keys in each of four ranges; each key has a false row then a true row. +INSERT INTO merge_semi_data VALUES + ('2026-10-09 00:00:00', 1, false), ('2026-10-09 00:00:01', 1, true), + ('2026-10-09 00:00:00', 2, false), ('2026-10-09 00:00:01', 2, true), + ('2026-10-09 00:00:00', 3, false), ('2026-10-09 00:00:01', 3, true), + ('2026-10-09 00:00:00', 11, false), ('2026-10-09 00:00:01', 11, true), + ('2026-10-09 00:00:00', 12, false), ('2026-10-09 00:00:01', 12, true), + ('2026-10-09 00:00:00', 13, false), ('2026-10-09 00:00:01', 13, true), + ('2026-10-09 00:00:00', 21, false), ('2026-10-09 00:00:01', 21, true), + ('2026-10-09 00:00:00', 22, false), ('2026-10-09 00:00:01', 22, true), + ('2026-10-09 00:00:00', 23, false), ('2026-10-09 00:00:01', 23, true), + ('2026-10-09 00:00:00', 31, false), ('2026-10-09 00:00:01', 31, true), + ('2026-10-09 00:00:00', 32, false), ('2026-10-09 00:00:01', 32, true), + ('2026-10-09 00:00:00', 33, false), ('2026-10-09 00:00:01', 33, true); + +Affected Rows: 24 + +-- Check the plan retains the partitioned left-semi join and hash keys (N=4). +-- SQLNESS REPLACE (peers.*) peers=REDACTED +EXPLAIN SELECT key_id, MIN(ts) AS first_ts +FROM merge_semi_data +WHERE key_id IN ( + SELECT key_id FROM merge_semi_data WHERE is_match = true +) +GROUP BY key_id ORDER BY key_id; + ++---------------+---------------------------------------------------------------------------------------------------------------------------------------+ +| plan_type | plan | ++---------------+---------------------------------------------------------------------------------------------------------------------------------------+ +| logical_plan | Sort: merge_semi_data.key_id ASC NULLS LAST | +| | Projection: merge_semi_data.key_id, min(merge_semi_data.ts) AS first_ts | +| | Aggregate: groupBy=[[merge_semi_data.key_id]], aggr=[[min(merge_semi_data.ts)]] | +| | LeftSemi Join: merge_semi_data.key_id = __correlated_sq_1.key_id | +| | Projection: merge_semi_data.ts, merge_semi_data.key_id | +| | MergeScan [is_placeholder=false, remote_input=[ | +| | Filter: merge_semi_data.key_id IS NOT NULL | +| | TableScan: merge_semi_data, partial_filters=[merge_semi_data.key_id IS NOT NULL] | +| | ]] | +| | MergeScan [is_placeholder=false, remote_input=[ | +| | SubqueryAlias: __correlated_sq_1 | +| | Projection: merge_semi_data.key_id | +| | Filter: merge_semi_data.is_match AND merge_semi_data.key_id IS NOT NULL | +| | TableScan: merge_semi_data, partial_filters=[merge_semi_data.is_match, merge_semi_data.key_id IS NOT NULL] | +| | ]] | +| physical_plan | SortPreservingMergeExec: [key_id@0 ASC NULLS LAST] | +| | ProjectionExec: expr=[key_id@0 as key_id, min(merge_semi_data.ts)@1 as first_ts] | +| | SortExec: expr=[key_id@0 ASC NULLS LAST], preserve_partitioning=[true] | +| | AggregateExec: mode=SinglePartitioned, gby=[key_id@1 as key_id], aggr=[min(merge_semi_data.ts)] | +| | HashJoinExec: mode=Partitioned, join_type=LeftSemi, on=[(key_id@1, key_id@0)] | +| | RepartitionExec: partitioning=Hash([key_id@1], 4), input_partitions=4 | +| | ProjectionExec: expr=[ts@0 as ts, key_id@1 as key_id] | +| | MergeScanExec: peers=REDACTED +| | RepartitionExec: partitioning=Hash([key_id@0], 4), input_partitions=4 | +| | MergeScanExec: peers=REDACTED +| | | ++---------------+---------------------------------------------------------------------------------------------------------------------------------------+ + +-- Every key matches and reports its earliest (false-row) timestamp. +SELECT key_id, MIN(ts) AS first_ts +FROM merge_semi_data +WHERE key_id IN ( + SELECT key_id FROM merge_semi_data WHERE is_match = true +) +GROUP BY key_id ORDER BY key_id; + ++--------+---------------------+ +| key_id | first_ts | ++--------+---------------------+ +| 1 | 2026-10-09T00:00:00 | +| 2 | 2026-10-09T00:00:00 | +| 3 | 2026-10-09T00:00:00 | +| 11 | 2026-10-09T00:00:00 | +| 12 | 2026-10-09T00:00:00 | +| 13 | 2026-10-09T00:00:00 | +| 21 | 2026-10-09T00:00:00 | +| 22 | 2026-10-09T00:00:00 | +| 23 | 2026-10-09T00:00:00 | +| 31 | 2026-10-09T00:00:00 | +| 32 | 2026-10-09T00:00:00 | +| 33 | 2026-10-09T00:00:00 | ++--------+---------------------+ + +DROP TABLE merge_semi_data; + +Affected Rows: 0 + diff --git a/tests/cases/standalone/common/join/merge_scan_semi_join.sql b/tests/cases/standalone/common/join/merge_scan_semi_join.sql new file mode 100644 index 00000000000..34e7b226592 --- /dev/null +++ b/tests/cases/standalone/common/join/merge_scan_semi_join.sql @@ -0,0 +1,49 @@ +-- Regression coverage for issues 9456/9492 (PR 9491): preserve IN/semi-join +-- results when merge-scan inputs are distributed across hash partitions. +CREATE TABLE merge_semi_data ( + ts TIMESTAMP TIME INDEX, + key_id INT, + is_match BOOLEAN, + PRIMARY KEY (key_id) +) +PARTITION ON COLUMNS (key_id) ( + key_id < 10, + key_id >= 10 AND key_id < 20, + key_id >= 20 AND key_id < 30, + key_id >= 30 +) +ENGINE=mito; + +-- Three keys in each of four ranges; each key has a false row then a true row. +INSERT INTO merge_semi_data VALUES + ('2026-10-09 00:00:00', 1, false), ('2026-10-09 00:00:01', 1, true), + ('2026-10-09 00:00:00', 2, false), ('2026-10-09 00:00:01', 2, true), + ('2026-10-09 00:00:00', 3, false), ('2026-10-09 00:00:01', 3, true), + ('2026-10-09 00:00:00', 11, false), ('2026-10-09 00:00:01', 11, true), + ('2026-10-09 00:00:00', 12, false), ('2026-10-09 00:00:01', 12, true), + ('2026-10-09 00:00:00', 13, false), ('2026-10-09 00:00:01', 13, true), + ('2026-10-09 00:00:00', 21, false), ('2026-10-09 00:00:01', 21, true), + ('2026-10-09 00:00:00', 22, false), ('2026-10-09 00:00:01', 22, true), + ('2026-10-09 00:00:00', 23, false), ('2026-10-09 00:00:01', 23, true), + ('2026-10-09 00:00:00', 31, false), ('2026-10-09 00:00:01', 31, true), + ('2026-10-09 00:00:00', 32, false), ('2026-10-09 00:00:01', 32, true), + ('2026-10-09 00:00:00', 33, false), ('2026-10-09 00:00:01', 33, true); + +-- Check the plan retains the partitioned left-semi join and hash keys (N=4). +-- SQLNESS REPLACE (peers.*) peers=REDACTED +EXPLAIN SELECT key_id, MIN(ts) AS first_ts +FROM merge_semi_data +WHERE key_id IN ( + SELECT key_id FROM merge_semi_data WHERE is_match = true +) +GROUP BY key_id ORDER BY key_id; + +-- Every key matches and reports its earliest (false-row) timestamp. +SELECT key_id, MIN(ts) AS first_ts +FROM merge_semi_data +WHERE key_id IN ( + SELECT key_id FROM merge_semi_data WHERE is_match = true +) +GROUP BY key_id ORDER BY key_id; + +DROP TABLE merge_semi_data; diff --git a/tests/cases/standalone/common/order/order_by_exceptions.result b/tests/cases/standalone/common/order/order_by_exceptions.result index bdf215f8f26..dc964ebfe09 100644 --- a/tests/cases/standalone/common/order/order_by_exceptions.result +++ b/tests/cases/standalone/common/order/order_by_exceptions.result @@ -83,7 +83,7 @@ EXPLAIN SELECT a % 2, b FROM test UNION SELECT a % 2 AS k, b FROM test ORDER BY | | RepartitionExec: REDACTED | | AggregateExec: mode=Partial, gby=[test.a % Int64(2)@0 as test.a % Int64(2), b@1 as b], aggr=[] | | | RepartitionExec: REDACTED -| | InterleaveExec | +| | UnionExec | | | MergeScanExec: REDACTED | | MergeScanExec: REDACTED | | | diff --git a/tests/cases/standalone/common/tql-explain-analyze/explain.result b/tests/cases/standalone/common/tql-explain-analyze/explain.result index fed75f0448e..3e8b0ec2133 100644 --- a/tests/cases/standalone/common/tql-explain-analyze/explain.result +++ b/tests/cases/standalone/common/tql-explain-analyze/explain.result @@ -184,7 +184,6 @@ TQL EXPLAIN VERBOSE (0, 10, '5s') test; | physical_plan after LimitedDistinctAggregation_| SAME TEXT AS ABOVE_| | physical_plan after FilterPushdown_| SAME TEXT AS ABOVE_| | physical_plan after parallelize_scan_| SAME TEXT AS ABOVE_| -| physical_plan after PassDistributionRule_| SAME TEXT AS ABOVE_| | physical_plan after PromqlTsidNarrowJoin_| SAME TEXT AS ABOVE_| | physical_plan after EnforceSorting_| SAME TEXT AS ABOVE_| | physical_plan after WindowTopN_| SAME TEXT AS ABOVE_| @@ -341,7 +340,6 @@ TQL EXPLAIN VERBOSE (0, 10, '5s') test AS series; | physical_plan after LimitedDistinctAggregation_| SAME TEXT AS ABOVE_| | physical_plan after FilterPushdown_| SAME TEXT AS ABOVE_| | physical_plan after parallelize_scan_| SAME TEXT AS ABOVE_| -| physical_plan after PassDistributionRule_| SAME TEXT AS ABOVE_| | physical_plan after PromqlTsidNarrowJoin_| SAME TEXT AS ABOVE_| | physical_plan after EnforceSorting_| SAME TEXT AS ABOVE_| | physical_plan after WindowTopN_| SAME TEXT AS ABOVE_| @@ -528,7 +526,6 @@ TQL EXPLAIN VERBOSE (0, 10, '5s') test_nano; | physical_plan after LimitedDistinctAggregation_| SAME TEXT AS ABOVE_| | physical_plan after FilterPushdown_| SAME TEXT AS ABOVE_| | physical_plan after parallelize_scan_| SAME TEXT AS ABOVE_| -| physical_plan after PassDistributionRule_| SAME TEXT AS ABOVE_| | physical_plan after PromqlTsidNarrowJoin_| SAME TEXT AS ABOVE_| | physical_plan after EnforceSorting_| SAME TEXT AS ABOVE_| | physical_plan after WindowTopN_| SAME TEXT AS ABOVE_| diff --git a/tests/cases/standalone/common/tql/tql-cte.result b/tests/cases/standalone/common/tql/tql-cte.result index 7fbc621319c..c2bc149e66c 100644 --- a/tests/cases/standalone/common/tql/tql-cte.result +++ b/tests/cases/standalone/common/tql/tql-cte.result @@ -596,7 +596,7 @@ SELECT 'host2' as source, ts, cpu FROM host2_data; | | Filter: labels.host = Utf8("host2") AND labels.ts >= TimestampMillisecond(-299999, None) AND labels.ts <= TimestampMillisecond(40000, None) | | | TableScan: labels, partial_filters=[labels.host = Utf8("host2"), labels.ts >= TimestampMillisecond(-299999, None), labels.ts <= TimestampMillisecond(40000, None)] | | | ]] | -| physical_plan | InterleaveExec | +| physical_plan | UnionExec | | | CooperativeExec | | | MergeScanExec: REDACTED | | CooperativeExec | diff --git a/tests/perf/query_cases/mergescan_shuffle_9456/case.toml b/tests/perf/query_cases/mergescan_shuffle_9456/case.toml new file mode 100644 index 00000000000..d96bdc9eb01 --- /dev/null +++ b/tests/perf/query_cases/mergescan_shuffle_9456/case.toml @@ -0,0 +1,152 @@ +# Opt-in comparison of Hash RepartitionExec cost on a multi-region PromQL join. +# ANALYZE records plans and metrics; EVAL preserves complete results for comparison. + +[case] +name = "mergescan_shuffle_9456" +description = "Opt-in multi-region Partitioned PromQL shuffle cost measurement" +issue = "https://github.com/GreptimeTeam/greptimedb/issues/9456" + +[scenario] +kind = "prom_remote_write_then_query" + +[scenario.remote_write] +database = "public" +metric = "mergescan_shuffle_9456" +physical_table = "mergescan_shuffle_phys" +series_count = 4096 +samples_per_series = 120 +start_unix_millis = 1704067200000 +step_millis = 60000 +chunk_series_count = 128 +timeout_seconds = 300 +visibility_timeout_seconds = 300 + +base_setup_sql = [ + """ +CREATE TABLE mergescan_shuffle_phys ( + greptime_timestamp TIMESTAMP(3) TIME INDEX, + host STRING, + instance STRING, + greptime_value DOUBLE, + PRIMARY KEY (host, instance) +) +PARTITION ON COLUMNS (instance) ( + instance < 'instance000128', + instance >= 'instance000128' AND instance < 'instance000256', + instance >= 'instance000256' AND instance < 'instance000384', + instance >= 'instance000384' AND instance < 'instance000512', + instance >= 'instance000512' AND instance < 'instance000640', + instance >= 'instance000640' AND instance < 'instance000768', + instance >= 'instance000768' AND instance < 'instance000896', + instance >= 'instance000896' AND instance < 'instance001024', + instance >= 'instance001024' AND instance < 'instance001152', + instance >= 'instance001152' AND instance < 'instance001280', + instance >= 'instance001280' AND instance < 'instance001408', + instance >= 'instance001408' AND instance < 'instance001536', + instance >= 'instance001536' AND instance < 'instance001664', + instance >= 'instance001664' AND instance < 'instance001792', + instance >= 'instance001792' AND instance < 'instance001920', + instance >= 'instance001920' AND instance < 'instance002048', + instance >= 'instance002048' AND instance < 'instance002176', + instance >= 'instance002176' AND instance < 'instance002304', + instance >= 'instance002304' AND instance < 'instance002432', + instance >= 'instance002432' AND instance < 'instance002560', + instance >= 'instance002560' AND instance < 'instance002688', + instance >= 'instance002688' AND instance < 'instance002816', + instance >= 'instance002816' AND instance < 'instance002944', + instance >= 'instance002944' AND instance < 'instance003072', + instance >= 'instance003072' AND instance < 'instance003200', + instance >= 'instance003200' AND instance < 'instance003328', + instance >= 'instance003328' AND instance < 'instance003456', + instance >= 'instance003456' AND instance < 'instance003584', + instance >= 'instance003584' AND instance < 'instance003712', + instance >= 'instance003712' AND instance < 'instance003840', + instance >= 'instance003840' AND instance < 'instance003968', + instance >= 'instance003968' +) ENGINE=metric WITH ('physical_metric_table'='true') +""", + """ +CREATE TABLE mergescan_shuffle_9456 ( + greptime_timestamp TIMESTAMP(3) TIME INDEX, + host STRING, + instance STRING, + greptime_value DOUBLE, + PRIMARY KEY (host, instance) +) ENGINE=metric WITH ('on_physical_table'='mergescan_shuffle_phys') +""", +] +candidate_setup_sql = [ + """ +CREATE TABLE mergescan_shuffle_phys ( + greptime_timestamp TIMESTAMP(3) TIME INDEX, + host STRING, + instance STRING, + greptime_value DOUBLE, + PRIMARY KEY (host, instance) +) +PARTITION ON COLUMNS (instance) ( + instance < 'instance000128', + instance >= 'instance000128' AND instance < 'instance000256', + instance >= 'instance000256' AND instance < 'instance000384', + instance >= 'instance000384' AND instance < 'instance000512', + instance >= 'instance000512' AND instance < 'instance000640', + instance >= 'instance000640' AND instance < 'instance000768', + instance >= 'instance000768' AND instance < 'instance000896', + instance >= 'instance000896' AND instance < 'instance001024', + instance >= 'instance001024' AND instance < 'instance001152', + instance >= 'instance001152' AND instance < 'instance001280', + instance >= 'instance001280' AND instance < 'instance001408', + instance >= 'instance001408' AND instance < 'instance001536', + instance >= 'instance001536' AND instance < 'instance001664', + instance >= 'instance001664' AND instance < 'instance001792', + instance >= 'instance001792' AND instance < 'instance001920', + instance >= 'instance001920' AND instance < 'instance002048', + instance >= 'instance002048' AND instance < 'instance002176', + instance >= 'instance002176' AND instance < 'instance002304', + instance >= 'instance002304' AND instance < 'instance002432', + instance >= 'instance002432' AND instance < 'instance002560', + instance >= 'instance002560' AND instance < 'instance002688', + instance >= 'instance002688' AND instance < 'instance002816', + instance >= 'instance002816' AND instance < 'instance002944', + instance >= 'instance002944' AND instance < 'instance003072', + instance >= 'instance003072' AND instance < 'instance003200', + instance >= 'instance003200' AND instance < 'instance003328', + instance >= 'instance003328' AND instance < 'instance003456', + instance >= 'instance003456' AND instance < 'instance003584', + instance >= 'instance003584' AND instance < 'instance003712', + instance >= 'instance003712' AND instance < 'instance003840', + instance >= 'instance003840' AND instance < 'instance003968', + instance >= 'instance003968' +) ENGINE=metric WITH ('physical_metric_table'='true') +""", + """ +CREATE TABLE mergescan_shuffle_9456 ( + greptime_timestamp TIMESTAMP(3) TIME INDEX, + host STRING, + instance STRING, + greptime_value DOUBLE, + PRIMARY KEY (host, instance) +) ENGINE=metric WITH ('on_physical_table'='mergescan_shuffle_phys') +""", +] + +[[scenario.queries]] +name = "partitioned_shuffle_cost" +kind = "tql" +query = "TQL ANALYZE VERBOSE (1704067200, 1704074340, '1m') rate(mergescan_shuffle_9456[5m]) / ignoring(host) rate(mergescan_shuffle_9456[5m])" +warmup = 1 +iterations = 7 + +[[scenario.queries]] +name = "collect_left_control" +kind = "tql" +query = "TQL ANALYZE VERBOSE (1704067200, 1704074340, '1m') rate(mergescan_shuffle_9456[5m]) / rate(mergescan_shuffle_9456[5m])" +warmup = 1 +iterations = 7 + +[[scenario.queries]] +name = "partitioned_complete_results" +kind = "tql" +query = "TQL EVAL (1704067200, 1704074340, '1m') rate(mergescan_shuffle_9456[5m]) / ignoring(host) rate(mergescan_shuffle_9456[5m])" +warmup = 0 +iterations = 1