diff --git a/datafusion/core/tests/physical_optimizer/enforce_distribution.rs b/datafusion/core/tests/physical_optimizer/enforce_distribution.rs index 189650fe4afca..ac7e7a75a2c56 100644 --- a/datafusion/core/tests/physical_optimizer/enforce_distribution.rs +++ b/datafusion/core/tests/physical_optimizer/enforce_distribution.rs @@ -20,10 +20,10 @@ use std::ops::Deref; use std::sync::Arc; use crate::physical_optimizer::test_utils::{ - bounded_window_exec_with_can_repartition, check_integrity, coalesce_partitions_exec, - parquet_exec_with_sort, parquet_exec_with_stats, repartition_exec, schema, sort_exec, - sort_exec_with_preserve_partitioning, sort_merge_join_exec, - sort_preserving_merge_exec, union_exec, + RequirementsTestExec, bounded_window_exec_with_can_repartition, check_integrity, + coalesce_partitions_exec, parquet_exec_with_sort, parquet_exec_with_stats, + repartition_exec, schema, sort_exec, sort_exec_with_preserve_partitioning, + sort_merge_join_exec, sort_preserving_merge_exec, union_exec, }; use arrow::array::{RecordBatch, UInt8Array, UInt64Array}; @@ -747,6 +747,93 @@ impl TestConfig { } } +#[derive(Debug, Clone, Copy)] +enum ExpectedPlan { + Reuse, + Hash, +} + +#[test] +fn range_satisfaction_config_matrix() -> Result<()> { + const INPUT_PARTITIONS: usize = 4; + const MET: usize = INPUT_PARTITIONS; + const NOT_MET: usize = INPUT_PARTITIONS + 1; + const DISABLED: usize = 0; + const EQUAL: usize = INPUT_PARTITIONS; + const GREATER: usize = INPUT_PARTITIONS + 1; + use ExpectedPlan::{Hash, Reuse}; + + let config_cases = [ + // subset preserve target exact subset incompatible + (NOT_MET, DISABLED, EQUAL, [Reuse, Hash, Hash]), + (NOT_MET, DISABLED, GREATER, [Hash, Hash, Hash]), + (NOT_MET, NOT_MET, EQUAL, [Reuse, Hash, Hash]), + (NOT_MET, NOT_MET, GREATER, [Hash, Hash, Hash]), + (NOT_MET, MET, EQUAL, [Reuse, Hash, Hash]), + (NOT_MET, MET, GREATER, [Reuse, Reuse, Hash]), + (MET, DISABLED, EQUAL, [Reuse, Reuse, Hash]), + (MET, DISABLED, GREATER, [Reuse, Reuse, Hash]), + (MET, NOT_MET, EQUAL, [Reuse, Reuse, Hash]), + (MET, NOT_MET, GREATER, [Reuse, Reuse, Hash]), + (MET, MET, EQUAL, [Reuse, Reuse, Hash]), + (MET, MET, GREATER, [Reuse, Reuse, Hash]), + ]; + for (subset_threshold, preserve_file_partitions, target_partitions, expected) in + config_cases + { + let key_cases = [ + ("exact", vec![col("a", &schema())?], expected[0]), + ( + "subset", + vec![col("a", &schema())?, col("b", &schema())?], + expected[1], + ), + ("incompatible", vec![col("b", &schema())?], expected[2]), + ]; + for (key_match, partition_keys, expected_plan) in key_cases { + let input = parquet_exec_with_output_partitioning(range_partitioning( + "a", + [10, 20, 30], + SortOptions::default(), + )?); + let requirement = RequirementsTestExec::new(input) + .with_required_input_distribution(Distribution::KeyPartitioned( + partition_keys, + )) + .into_arc(); + + let mut config = + TestConfig::default().with_query_execution_partitions(target_partitions); + config.config.optimizer.subset_repartition_threshold = subset_threshold; + config.config.optimizer.preserve_file_partitions = preserve_file_partitions; + + let plan = config.to_plan(requirement, &DISTRIB_DISTRIB_SORT); + let plan = displayable(plan.as_ref()).indent(true).to_string(); + let repartitions = plan + .lines() + .filter(|line| line.contains("RepartitionExec:")) + .collect::>(); + + let matches_expected = match expected_plan { + Reuse => repartitions.is_empty(), + Hash => matches!( + repartitions.as_slice(), + [repartition] if repartition.contains("partitioning=Hash") + ), + }; + assert!( + matches_expected, + "unexpected optimized plan for key_match={key_match}, \ + subset_threshold={subset_threshold}, \ + preserve_file_partitions={preserve_file_partitions}, \ + target_partitions={target_partitions}:\n{plan}" + ); + } + } + + Ok(()) +} + #[test] fn range_aggregate_reuses_range_partitioning() -> Result<()> { let input = parquet_exec_with_output_partitioning(range_partitioning( diff --git a/datafusion/core/tests/physical_optimizer/test_utils.rs b/datafusion/core/tests/physical_optimizer/test_utils.rs index 74230b24e2ab5..3235ea25fdb3b 100644 --- a/datafusion/core/tests/physical_optimizer/test_utils.rs +++ b/datafusion/core/tests/physical_optimizer/test_utils.rs @@ -40,10 +40,10 @@ use datafusion_execution::object_store::ObjectStoreUrl; use datafusion_execution::{SendableRecordBatchStream, TaskContext}; use datafusion_expr::{WindowFrame, WindowFunctionDefinition}; use datafusion_functions_aggregate::count::count_udaf; -use datafusion_physical_expr::EquivalenceProperties; use datafusion_physical_expr::aggregate::{AggregateExprBuilder, AggregateFunctionExpr}; use datafusion_physical_expr::expressions::{self, col}; use datafusion_physical_expr::projection::ProjectionExprs; +use datafusion_physical_expr::{Distribution, EquivalenceProperties}; use datafusion_physical_expr_common::physical_expr::PhysicalExpr; use datafusion_physical_expr_common::sort_expr::{ LexOrdering, OrderingRequirements, PhysicalSortExpr, @@ -68,8 +68,9 @@ use datafusion_physical_plan::tree_node::PlanContext; use datafusion_physical_plan::union::UnionExec; use datafusion_physical_plan::windows::{BoundedWindowAggExec, create_window_expr}; use datafusion_physical_plan::{ - DisplayAs, DisplayFormatType, ExecutionPlan, InputOrderMode, Partitioning, - PlanProperties, SortOrderPushdownResult, StatisticsArgs, displayable, + DisplayAs, DisplayFormatType, ExecutionPlan, InputDistributionRequirements, + InputOrderMode, Partitioning, PlanProperties, SortOrderPushdownResult, + StatisticsArgs, displayable, }; /// Create a non sorted parquet exec @@ -435,6 +436,7 @@ pub fn projection_exec( #[derive(Debug)] pub struct RequirementsTestExec { required_input_ordering: Option, + required_input_distribution: Distribution, maintains_input_order: bool, input: Arc, } @@ -443,6 +445,7 @@ impl RequirementsTestExec { pub fn new(input: Arc) -> Self { Self { required_input_ordering: None, + required_input_distribution: Distribution::UnspecifiedDistribution, maintains_input_order: true, input, } @@ -457,6 +460,15 @@ impl RequirementsTestExec { self } + /// sets the required input distribution + pub fn with_required_input_distribution( + mut self, + required_input_distribution: Distribution, + ) -> Self { + self.required_input_distribution = required_input_distribution; + self + } + /// set the maintains_input_order flag pub fn with_maintains_input_order(mut self, maintains_input_order: bool) -> Self { self.maintains_input_order = maintains_input_order; @@ -500,6 +512,10 @@ impl ExecutionPlan for RequirementsTestExec { ] } + fn input_distribution_requirements(&self) -> InputDistributionRequirements { + InputDistributionRequirements::new(vec![self.required_input_distribution.clone()]) + } + fn maintains_input_order(&self) -> Vec { vec![self.maintains_input_order] } @@ -515,6 +531,7 @@ impl ExecutionPlan for RequirementsTestExec { assert_eq!(children.len(), 1); Ok(RequirementsTestExec::new(Arc::clone(&children[0])) .with_required_input_ordering(self.required_input_ordering.clone()) + .with_required_input_distribution(self.required_input_distribution.clone()) .with_maintains_input_order(self.maintains_input_order) .into_arc()) } diff --git a/datafusion/sqllogictest/test_files/range_partitioning.slt b/datafusion/sqllogictest/test_files/range_partitioning.slt index 1e0a1582eac65..9701c41377ef3 100644 --- a/datafusion/sqllogictest/test_files/range_partitioning.slt +++ b/datafusion/sqllogictest/test_files/range_partitioning.slt @@ -119,89 +119,7 @@ SELECT range_key, non_range_key, SUM(value) FROM range_partitioned GROUP BY rang ########## -# TEST 4: Exact Range Aggregate Below Subset Threshold -# Even when subset satisfaction is disabled, exact Range([range_key]) -# satisfies GROUP BY range_key when repartitioning would not increase -# partition count. -########## - -statement ok -set datafusion.execution.target_partitions = 4; - -statement ok -set datafusion.optimizer.subset_repartition_threshold = 5; - -statement ok -set datafusion.optimizer.preserve_file_partitions = 0; - -query TT -EXPLAIN SELECT range_key, SUM(value) FROM range_partitioned GROUP BY range_key; ----- -physical_plan -01)AggregateExec: mode=SinglePartitioned, gby=[range_key@0 as range_key], aggr=[sum(range_partitioned.value)] -02)--DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false - - -########## -# TEST 5: Range Subset Aggregate Rehashes Below Subset Threshold -# Range([range_key]) is only a subset of GROUP BY (range_key, non_range_key), -# so it should not satisfy the aggregate key when subset satisfaction is -# disabled. -########## - -statement ok -set datafusion.execution.target_partitions = 4; - -statement ok -set datafusion.optimizer.subset_repartition_threshold = 5; - -statement ok -set datafusion.optimizer.preserve_file_partitions = 0; - -query TT -EXPLAIN SELECT range_key, non_range_key, SUM(value) FROM range_partitioned GROUP BY range_key, non_range_key; ----- -physical_plan -01)AggregateExec: mode=FinalPartitioned, gby=[range_key@0 as range_key, non_range_key@1 as non_range_key], aggr=[sum(range_partitioned.value)] -02)--RepartitionExec: partitioning=Hash([range_key@0, non_range_key@1], 4), input_partitions=4 -03)----AggregateExec: mode=Partial, gby=[range_key@0 as range_key, non_range_key@1 as non_range_key], aggr=[sum(range_partitioned.value)] -04)------DataSourceExec: file_groups=, projection=[range_key, non_range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false - - -########## -# TEST 6: Aggregate Rehashes Below Subset Threshold -# With subset threshold 5 and only 4 input partitions, planning repartitions -# to increase parallelism instead of reusing Range partitioning. -########## - -statement ok -set datafusion.execution.target_partitions = 5; - -statement ok -set datafusion.optimizer.subset_repartition_threshold = 5; - -statement ok -set datafusion.optimizer.preserve_file_partitions = 0; - -query TT -EXPLAIN SELECT range_key, SUM(value) FROM range_partitioned GROUP BY range_key; ----- -physical_plan -01)AggregateExec: mode=FinalPartitioned, gby=[range_key@0 as range_key], aggr=[sum(range_partitioned.value)] -02)--RepartitionExec: partitioning=Hash([range_key@0], 5), input_partitions=5 -03)----AggregateExec: mode=Partial, gby=[range_key@0 as range_key], aggr=[sum(range_partitioned.value)] -04)------RepartitionExec: partitioning=RoundRobinBatch(5), input_partitions=4 -05)--------DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false - -statement ok -set datafusion.execution.target_partitions = 4; - -statement ok -reset datafusion.optimizer.subset_repartition_threshold; - - -########## -# TEST 7: Aggregate Preserves Range When Preserve File Threshold Met +# TEST 4: Aggregate Preserves Range When Preserve File Threshold Met # With preserve-file threshold 1 and 4 input partitions, Range is preserved # even though target_partitions is 5. ########## @@ -230,7 +148,7 @@ reset datafusion.optimizer.preserve_file_partitions; ########## -# TEST 8: Aggregate Rehashes When Preserve File Threshold Not Met +# TEST 5: Aggregate Rehashes When Preserve File Threshold Not Met # With preserve-file threshold 5 and only 4 input partitions, planning can # repartition to increase parallelism. ########## @@ -271,7 +189,7 @@ set datafusion.optimizer.preserve_file_partitions = 0; ########## -# TEST 9: Join on Range Partition Column +# TEST 6: Join on Range Partition Column # A partitioned inner hash join requires co-partitioned KeyPartitioned inputs. # Compatible Range layouts satisfy both the per-child key requirements and the # cross-child layout requirement, so no Hash repartitioning is inserted. @@ -303,7 +221,7 @@ ORDER BY l.range_key; 35 350 350 ########## -# TEST 10: Incompatible Range Join Repartitions +# TEST 7: Incompatible Range Join Repartitions # Both inputs are independently range partitioned on range_key, but their split # points differ. The per-child key requirements can be satisfied by Range, but # the co-partitioned layout requirement cannot, so Hash repartitioning repairs @@ -338,7 +256,7 @@ ORDER BY l.range_key; 35 350 350 ########## -# TEST 11: Non-Range Join Repartitions +# TEST 8: Non-Range Join Repartitions # Range([range_key]) does not satisfy KeyPartitioned([non_range_key]), so # planning inserts Hash repartitioning on the actual join key. ########## @@ -395,7 +313,7 @@ ORDER BY l.non_range_key, l.value, r.value; 2 350 350 ########## -# TEST 12: Left-Side Range Hash Joins +# TEST 9: Left-Side Range Hash Joins # Compatible Range layouts satisfy left-side partitioned hash join # requirements without Hash repartitioning. ########## @@ -477,7 +395,7 @@ ORDER BY l.range_key; 35 350 ########## -# TEST 13: Left-Side Range Hash Joins With Incomplete Range Keys +# TEST 10: Left-Side Range Hash Joins With Incomplete Range Keys # Range partitioning covers only range_key, so joins requiring additional # or different keys are repaired with Hash repartitioning. ########## @@ -512,7 +430,7 @@ physical_plan 05)----DataSourceExec: file_groups=, projection=[non_range_key, value], output_partitioning=UnknownPartitioning(4), file_type=csv, has_header=false ########## -# TEST 14: Left-Side Range Hash Joins With Incompatible Range Layouts +# TEST 11: Left-Side Range Hash Joins With Incompatible Range Layouts # Different split points or partition counts do not satisfy the # co-partitioned layout requirement. ########## @@ -560,7 +478,7 @@ physical_plan 05)----DataSourceExec: file_groups=, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20)], 3), file_type=csv, has_header=false ########## -# TEST 15: LeftMark Subqueries Over Range Hash Joins +# TEST 12: LeftMark Subqueries Over Range Hash Joins # SQL IN subqueries decorrelate to LeftMark joins. These queries pin matched, # unmatched, and NULL marker behavior over compatible Range inputs. ########## @@ -608,7 +526,7 @@ ORDER BY l.range_key; 35 350 ########## -# TEST 16: Compatible Range Join Repartitions to Increase Parallelism +# TEST 13: Compatible Range Join Repartitions to Increase Parallelism # Co-partitioning satisfaction does not prevent a repartition that increases # parallelism. With target_partitions larger than the Range partition count, # both sides are hash repartitioned. @@ -645,7 +563,7 @@ ORDER BY l.range_key; 35 350 350 ########## -# TEST 17: Preserve File Partitions Preserves Range Join Inputs +# TEST 14: Preserve File Partitions Preserves Range Join Inputs # preserve_file_partitions preserves compatible Range inputs for partitioned # joins even when target_partitions is higher than the input partition count. ########## @@ -685,7 +603,7 @@ statement ok set datafusion.optimizer.preserve_file_partitions = 0; ########## -# TEST 18: Nested Range Joins +# TEST 15: Nested Range Joins # Compatible Range partitioning is preserved through the lower join, allowing # the upper join to consume it without Hash repartitioning either input. ########## @@ -720,7 +638,7 @@ ORDER BY l.range_key; 35 350 350 350 ########## -# TEST 19: Range Aggregates Feed Range Join +# TEST 16: Range Aggregates Feed Range Join # Aggregates on range_key preserve reusable partitioning for the downstream # partitioned join. ########## @@ -775,7 +693,7 @@ ORDER BY l.range_key; 35 350 350 ########## -# TEST 20: Range Join Feeds Aggregate +# TEST 17: Range Join Feeds Aggregate # The join preserves compatible Range partitioning on range_key, allowing the # aggregate above it to avoid Hash repartitioning. ########## @@ -809,7 +727,7 @@ ORDER BY l.range_key; 35 700 ########## -# TEST 21: Right Join on Range Partition Column +# TEST 18: Right Join on Range Partition Column # Compatible Range inputs satisfy the join's partitioning requirements, so no # Hash repartitioning is inserted. The left filter keeps its Range partitioning # and the unmatched right rows above 150 are preserved. @@ -842,7 +760,7 @@ NULL 30 300 NULL 35 350 ########## -# TEST 22: Right Semi Join on Range Partition Column +# TEST 19: Right Semi Join on Range Partition Column # Compatible Range inputs avoid Hash repartitioning for RightSemi joins. # Only right rows with a match on the filtered left side are returned. ########## @@ -870,7 +788,7 @@ ORDER BY r.range_key; 15 150 ########## -# TEST 23: Right Anti Join on Range Partition Column +# TEST 20: Right Anti Join on Range Partition Column # Compatible Range inputs avoid Hash repartitioning for RightAnti joins. # Only right rows without a match on the filtered left side are returned. ########## @@ -898,7 +816,7 @@ ORDER BY r.range_key; 35 350 ########## -# TEST 24: Incompatible Range Right Join Repartitions +# TEST 21: Incompatible Range Right Join Repartitions # The split points of the two inputs differ, so the co-partitioned layout # requirement cannot be satisfied and Hash repartitioning repairs both sides # of the right join. Results stay correct on the repartitioned path. @@ -933,7 +851,7 @@ NULL 30 300 NULL 35 350 ########## -# TEST 25: Composite-Key Right Join Repartitions +# TEST 22: Composite-Key Right Join Repartitions # Range([range_key]) does not satisfy a partitioned join on # (range_key, non_range_key), so both sides repartition on the full key. ########## @@ -972,7 +890,7 @@ statement ok reset datafusion.optimizer.subset_repartition_threshold; ########## -# TEST 26: Right Join with Mismatched Range Partition Counts Repartitions +# TEST 23: Right Join with Mismatched Range Partition Counts Repartitions # Both inputs are range partitioned on range_key, but declare a different number # of partitions (four vs three). The per-child key requirements can be satisfied # by Range, but the co-partitioned layout requirement cannot, so Hash @@ -1007,7 +925,7 @@ ORDER BY r.range_key; 350 35 350 ########## -# TEST 27: Right Join on Non-Range Key Repartitions +# TEST 24: Right Join on Non-Range Key Repartitions # Both inputs expose Range([range_key]), but the join key is non_range_key. # Range([range_key]) does not satisfy KeyPartitioned([non_range_key]), so # planning inserts Hash repartitioning on the actual join key for the right join. @@ -1042,7 +960,7 @@ ORDER BY r.range_key; 50 35 350 ########## -# TEST 28: Mark Join Marker Semantics +# TEST 25: Mark Join Marker Semantics # Mark joins preserve matched, unmatched, and NULL-key marker behavior over # range-partitioned inputs. ########## @@ -1094,7 +1012,7 @@ ORDER BY r.range_key; 35 350 ########## -# TEST 29: Sort Merge Join Avoids Repartition for Compatible Range Inputs +# TEST 26: Sort Merge Join Avoids Repartition for Compatible Range Inputs # Compatible Range inputs satisfy SortMergeJoinExec's co-partitioned # KeyPartitioned requirements. ########## @@ -1131,7 +1049,7 @@ ORDER BY l.range_key; 35 350 350 ########## -# TEST 30: Sort Merge Join Repartitions Incompatible Range Inputs +# TEST 27: Sort Merge Join Repartitions Incompatible Range Inputs # Different Range split points do not satisfy SortMergeJoinExec's # co-partitioned KeyPartitioned requirements. ########## @@ -1170,7 +1088,7 @@ statement ok reset datafusion.optimizer.prefer_hash_join; ########## -# TEST 31: Symmetric Hash Join Avoids Repartition for Compatible Range Inputs +# TEST 28: Symmetric Hash Join Avoids Repartition for Compatible Range Inputs # Compatible Range streams satisfy SymmetricHashJoinExec's co-partitioned # KeyPartitioned requirements. ########## @@ -1204,7 +1122,7 @@ FULL JOIN unbounded_range_like r ON l.range_key = r.range_key; 5 50 50 ########## -# TEST 32: Symmetric Hash Join Repartitions Incompatible Range Inputs +# TEST 29: Symmetric Hash Join Repartitions Incompatible Range Inputs # Different Range split points do not satisfy SymmetricHashJoinExec's # co-partitioned KeyPartitioned requirements. ########## @@ -1237,7 +1155,7 @@ FULL JOIN unbounded_range_like_shifted r ON l.range_key = r.range_key; 5 50 50 ########## -# TEST 33: Full Outer Join on Range Partition Column +# TEST 30: Full Outer Join on Range Partition Column # Full partitioned hash joins also opt in to Range satisfying KeyPartitioned # requirements, so compatible Range layouts avoid Hash repartitioning here too. ########## @@ -1268,9 +1186,9 @@ ORDER BY l.range_key; 35 350 350 ########## -# TEST 34: Full Outer Join Incompatible Range Repartitions -# Same as TEST 10, but for Full: differing split points between the two -# Range-partitioned inputs still require Hash repartitioning to co-partition. +# TEST 31: Full Outer Join Incompatible Range Repartitions +# For Full joins, differing split points between the two Range-partitioned +# inputs still require Hash repartitioning to co-partition. ########## query TT @@ -1301,7 +1219,7 @@ ORDER BY l.range_key; 35 350 350 ########## -# TEST 35: Full Outer Join Produces Matched and Unmatched Rows +# TEST 32: Full Outer Join Produces Matched and Unmatched Rows # `range_partitioned` and `range_partitioned_sparse` share the same Range # split points/partition count but only partially overlapping range_key # values, so this exercises matched rows, left-only unmatched rows (NULLs on @@ -1346,7 +1264,7 @@ statement ok reset datafusion.optimizer.preserve_file_partitions; ########## -# TEST 36: Union of Range Partitioned Inputs +# TEST 33: Union of Range Partitioned Inputs # Each input exposes the same Range partitioning on range_key, so the optimizer # converts UnionExec to InterleaveExec to avoid redundant repartitioning. ########## @@ -1395,7 +1313,7 @@ set datafusion.optimizer.preserve_file_partitions = 0; ########## -# TEST 37: Window on Range Partition Column +# TEST 34: Window on Range Partition Column # Range([range_key]) colocates equal range_key values, so # PARTITION BY range_key is satisfied without a hash repartition. ########## @@ -1423,7 +1341,7 @@ SELECT range_key, SUM(value) OVER (PARTITION BY range_key ORDER BY value) FROM r ########## -# TEST 38: Unbounded-Frame Window on Range Partition Column +# TEST 35: Unbounded-Frame Window on Range Partition Column # The unbounded frame makes DataFusion use WindowAggExec instead of # BoundedWindowAggExec, which likewise reuses Range partitioning without a # hash repartition. @@ -1452,7 +1370,7 @@ SELECT range_key, SUM(value) OVER (PARTITION BY range_key ORDER BY value ROWS BE ########## -# TEST 39: Window on Non-Range Column Rehashes +# TEST 36: Window on Non-Range Column Rehashes # Range([range_key]) does not colocate non_range_key values, so # PARTITION BY non_range_key still requires a hash repartition. ########## @@ -1481,7 +1399,7 @@ SELECT non_range_key, value, SUM(value) OVER (PARTITION BY non_range_key ORDER B ########## -# TEST 40: Unbounded-Frame Window on Non-Range Column Rehashes +# TEST 37: Unbounded-Frame Window on Non-Range Column Rehashes # The unbounded frame makes DataFusion use WindowAggExec; Range([range_key]) # does not colocate non_range_key values, so PARTITION BY non_range_key # still requires a hash repartition. @@ -1511,7 +1429,7 @@ SELECT non_range_key, value, SUM(value) OVER (PARTITION BY non_range_key ORDER B ########## -# TEST 41: Window Subset Satisfaction on Range Partition Column +# TEST 38: Window Subset Satisfaction on Range Partition Column # With the subset threshold met, Range([range_key]) satisfies # PARTITION BY (range_key, non_range_key): equal composite keys share the # same range_key, so they are already colocated. @@ -1543,7 +1461,7 @@ SELECT range_key, SUM(value) OVER (PARTITION BY range_key, non_range_key ORDER B ########## -# TEST 42: Window Subset Rehashes Below Subset Threshold +# TEST 39: Window Subset Rehashes Below Subset Threshold # Range([range_key]) is only a subset of PARTITION BY # (range_key, non_range_key), so it should not satisfy the window key when # subset satisfaction is disabled. @@ -1582,7 +1500,7 @@ reset datafusion.optimizer.preserve_file_partitions; ########## -# TEST 43: Window Without Partition Keys Uses a Single Partition +# TEST 40: Window Without Partition Keys Uses a Single Partition # A window with no PARTITION BY requires a single partition; range # partitioning is not applicable. ########## @@ -1612,7 +1530,7 @@ SELECT range_key, SUM(value) OVER (ORDER BY value) FROM range_partitioned ORDER ########## -# TEST 44: PartitionedTopK on Range Partition Column +# TEST 41: PartitionedTopK on Range Partition Column # Exact Range([range_key]) satisfies the TopK partition key and avoids repartitioning. ########## @@ -1655,7 +1573,7 @@ ORDER BY range_key; ########## -# TEST 45: PartitionedTopK on Non-Range Column +# TEST 42: PartitionedTopK on Non-Range Column # Partitioning on a non-range key cannot reuse Range([range_key]) and # requires hash repartitioning. ########## @@ -1691,7 +1609,7 @@ ORDER BY non_range_key; ########## -# TEST 46: PartitionedTopK Reuses Range Subset Partitioning +# TEST 43: PartitionedTopK Reuses Range Subset Partitioning # With subset threshold met and preserve-file disabled, Range([range_key]) # satisfies partitioning by (range_key, non_range_key). ########## @@ -1732,7 +1650,7 @@ ORDER BY range_key, non_range_key; ########## -# TEST 47: Range Subset PartitionedTopK Rehashes Below Subset Threshold +# TEST 44: Range Subset PartitionedTopK Rehashes Below Subset Threshold # Range([range_key]) is only a subset of PARTITION BY (range_key, non_range_key), # so it should not satisfy the TopK partition key when subset satisfaction is # disabled. @@ -1770,7 +1688,7 @@ statement ok reset datafusion.optimizer.enable_window_topn; ########## -# TEST 48: Subset of Inputs Compatible Does Not Trigger InterleaveExec +# TEST 45: Subset of Inputs Compatible Does Not Trigger InterleaveExec # In a three-way union, two inputs share the same Range split points [10,20,30] # while the third has a partially-overlapping but different set [15,20,30]. # can_interleave requires ALL inputs to match, so UnionExec is kept. @@ -1826,7 +1744,7 @@ ORDER BY range_key, value; 35 350 ########## -# TEST 49: Incompatible Range Split Points Falls Back to UnionExec +# TEST 46: Incompatible Range Split Points Falls Back to UnionExec # Two range-partitioned inputs with different split points cannot be interleaved, # so the optimizer keeps UnionExec instead of converting to InterleaveExec. ########## @@ -1868,7 +1786,7 @@ ORDER BY range_key, value; 35 350 ########## -# TEST 50: InterleaveExec Propagates Range Partitioning to Aggregate +# TEST 47: InterleaveExec Propagates Range Partitioning to Aggregate # InterleaveExec outputs the same Range partitioning as its compatible inputs, # allowing a downstream aggregate on range_key to run SinglePartitioned without # a Hash repartition.