From 47353981c045415ba28a382976340c894d389bd8 Mon Sep 17 00:00:00 2001 From: Stu Hood Date: Mon, 14 Sep 2026 16:11:26 -0700 Subject: [PATCH] Do not broadcast when co-partitioned. --- .../physical_optimizer/join_selection.rs | 260 ++++++++++++++++-- .../physical-optimizer/src/join_selection.rs | 17 +- .../physical-plan/src/joins/hash_join/exec.rs | 145 +++++++++- .../repartition_subset_satisfaction.slt | 18 +- 4 files changed, 391 insertions(+), 49 deletions(-) diff --git a/datafusion/core/tests/physical_optimizer/join_selection.rs b/datafusion/core/tests/physical_optimizer/join_selection.rs index d4561decc633..2d5234800f85 100644 --- a/datafusion/core/tests/physical_optimizer/join_selection.rs +++ b/datafusion/core/tests/physical_optimizer/join_selection.rs @@ -35,7 +35,9 @@ use datafusion_physical_expr::PhysicalExprRef; use datafusion_physical_expr::expressions::col; use datafusion_physical_expr::expressions::{BinaryExpr, Column, NegativeExpr}; use datafusion_physical_expr::intervals::utils::check_support; -use datafusion_physical_expr::{EquivalenceProperties, Partitioning, PhysicalExpr}; +use datafusion_physical_expr::{ + EquivalenceProperties, Partitioning, PhysicalExpr, RangePartitioning, SplitPoint, +}; use datafusion_physical_expr_common::sort_expr::PhysicalSortExpr; use datafusion_physical_optimizer::PhysicalOptimizerContext; use datafusion_physical_optimizer::PhysicalOptimizerRule; @@ -1220,6 +1222,230 @@ async fn test_join_selection_partitioned() { check_join_partition_mode(big, empty, join_on, false, PartitionMode::Partitioned); } +#[derive(Clone, Copy, Debug)] +enum PartitionCase { + CoPartitionedHash, + CoPartitionedRange, + MismatchedRange, + NonJoinKeyHash, + SinglePartition, +} + +struct PartitionedTestInputs { + left: Arc, + right: Arc, + on: Vec<(PhysicalExprRef, PhysicalExprRef)>, +} + +fn create_partitioned_test_inputs( + case: PartitionCase, + left_stats: Statistics, + right_stats: Statistics, +) -> Result { + let schema_left = Schema::new(vec![ + Field::new("k1", DataType::Int32, false), + Field::new("other1", DataType::Int32, false), + ]); + let schema_right = Schema::new(vec![ + Field::new("k2", DataType::Int32, false), + Field::new("other2", DataType::Int32, false), + ]); + + let key_left = col("k1", &schema_left)?; + let key_right = col("k2", &schema_right)?; + let other_left = col("other1", &schema_left)?; + let other_right = col("other2", &schema_right)?; + + let (left_part, right_part) = match case { + PartitionCase::CoPartitionedHash => ( + Partitioning::Hash(vec![Arc::clone(&key_left)], 2), + Partitioning::Hash(vec![Arc::clone(&key_right)], 2), + ), + PartitionCase::CoPartitionedRange => { + let split_points = vec![SplitPoint::new(vec![ScalarValue::Int32(Some(100))])]; + ( + Partitioning::Range(RangePartitioning::try_new( + [PhysicalSortExpr::new_default(Arc::clone(&key_left))].into(), + split_points.clone(), + )?), + Partitioning::Range(RangePartitioning::try_new( + [PhysicalSortExpr::new_default(Arc::clone(&key_right))].into(), + split_points, + )?), + ) + } + PartitionCase::MismatchedRange => ( + Partitioning::Range(RangePartitioning::try_new( + [PhysicalSortExpr::new_default(Arc::clone(&key_left))].into(), + vec![SplitPoint::new(vec![ScalarValue::Int32(Some(100))])], + )?), + Partitioning::Range(RangePartitioning::try_new( + [PhysicalSortExpr::new_default(Arc::clone(&key_right))].into(), + vec![SplitPoint::new(vec![ScalarValue::Int32(Some(200))])], + )?), + ), + PartitionCase::NonJoinKeyHash => ( + Partitioning::Hash(vec![other_left], 2), + Partitioning::Hash(vec![other_right], 2), + ), + PartitionCase::SinglePartition => ( + Partitioning::Hash(vec![Arc::clone(&key_left)], 1), + Partitioning::Hash(vec![Arc::clone(&key_right)], 1), + ), + }; + + let make_stats = |base: Statistics| Statistics { + num_rows: base.num_rows, + total_byte_size: base.total_byte_size, + column_statistics: vec![ + ColumnStatistics::new_unknown(), + ColumnStatistics::new_unknown(), + ], + }; + + let left = Arc::new( + StatisticsExec::new(make_stats(left_stats), schema_left) + .with_partitioning(left_part), + ); + let right = Arc::new( + StatisticsExec::new(make_stats(right_stats), schema_right) + .with_partitioning(right_part), + ); + let on = vec![(key_left, key_right)]; + + Ok(PartitionedTestInputs { left, right, on }) +} + +#[rstest( + case, + expected_mode, + case::co_partitioned_hash( + PartitionCase::CoPartitionedHash, + PartitionMode::Partitioned + ), + case::co_partitioned_range( + PartitionCase::CoPartitionedRange, + PartitionMode::Partitioned + ), + case::mismatched_range(PartitionCase::MismatchedRange, PartitionMode::CollectLeft), + case::non_join_key_hash(PartitionCase::NonJoinKeyHash, PartitionMode::CollectLeft), + case::single_partition(PartitionCase::SinglePartition, PartitionMode::CollectLeft) +)] +#[tokio::test] +async fn test_join_selection_co_partitioned_scenarios( + case: PartitionCase, + expected_mode: PartitionMode, +) -> Result<()> { + let inputs = + create_partitioned_test_inputs(case, small_statistics(), small_statistics())?; + check_join_partition_mode(inputs.left, inputs.right, inputs.on, false, expected_mode); + Ok(()) +} + +#[tokio::test] +async fn test_join_selection_co_partitioned_initial_collect_left_switches_to_partitioned() +-> Result<()> { + let inputs = create_partitioned_test_inputs( + PartitionCase::CoPartitionedHash, + small_statistics(), + small_statistics(), + )?; + let join = Arc::new(HashJoinExec::try_new( + inputs.left, + inputs.right, + inputs.on, + None, + &JoinType::Inner, + None, + PartitionMode::CollectLeft, + NullEquality::NullEqualsNothing, + false, + )?); + check_hash_join_mode(join, false, PartitionMode::Partitioned); + Ok(()) +} + +#[tokio::test] +async fn test_join_selection_co_partitioned_swaps_smaller_side_to_build() -> Result<()> { + let inputs = create_partitioned_test_inputs( + PartitionCase::CoPartitionedHash, + big_statistics(), + small_statistics(), + )?; + let join = Arc::new(HashJoinExec::try_new( + inputs.left, + inputs.right, + inputs.on, + None, + &JoinType::Inner, + None, + PartitionMode::Auto, + NullEquality::NullEqualsNothing, + false, + )?); + let optimized = check_hash_join_mode(join, true, PartitionMode::Partitioned); + let swapped_join = optimized + .downcast_ref::() + .unwrap() + .input() + .downcast_ref::() + .unwrap(); + // Right (smaller) became the left (build) child + assert_eq!(swapped_join.left().schema().field(0).name(), "k2"); + assert_eq!(swapped_join.right().schema().field(0).name(), "k1"); + Ok(()) +} + +#[tokio::test] +async fn test_join_selection_co_partitioned_null_aware_remains_collect_left() -> Result<()> +{ + let inputs = create_partitioned_test_inputs( + PartitionCase::CoPartitionedHash, + small_statistics(), + small_statistics(), + )?; + let join = Arc::new(HashJoinExec::try_new( + inputs.left, + inputs.right, + inputs.on, + None, + &JoinType::LeftAnti, + None, + PartitionMode::CollectLeft, + NullEquality::NullEqualsNothing, + true, + )?); + check_hash_join_mode(join, false, PartitionMode::CollectLeft); + Ok(()) +} + +fn check_hash_join_mode( + join: Arc, + is_swapped: bool, + expected_mode: PartitionMode, +) -> Arc { + let optimized_join = JoinSelection::new() + .optimize(join, &ConfigOptions::new()) + .unwrap(); + + let hash_join = if !is_swapped { + optimized_join + .downcast_ref::() + .expect("The type of the plan should not be changed") + } else { + let swapping_projection = optimized_join + .downcast_ref::() + .expect("A proj is required to swap columns back to their original order"); + swapping_projection + .input() + .downcast_ref::() + .expect("The type of the plan should not be changed") + }; + + assert_eq!(*hash_join.partition_mode(), expected_mode); + optimized_join +} + fn check_join_partition_mode( left: Arc, right: Arc, @@ -1241,27 +1467,7 @@ fn check_join_partition_mode( ) .unwrap(), ); - - let optimized_join = JoinSelection::new() - .optimize(join, &ConfigOptions::new()) - .unwrap(); - - if !is_swapped { - let swapped_join = optimized_join - .downcast_ref::() - .expect("The type of the plan should not be changed"); - assert_eq!(*swapped_join.partition_mode(), expected_mode); - } else { - let swapping_projection = optimized_join - .downcast_ref::() - .expect("A proj is required to swap columns back to their original order"); - let swapped_join = swapping_projection - .input() - .downcast_ref::() - .expect("The type of the plan should not be changed"); - - assert_eq!(*swapped_join.partition_mode(), expected_mode); - } + check_hash_join_mode(join, is_swapped, expected_mode); } #[derive(Debug)] @@ -1443,6 +1649,16 @@ impl StatisticsExec { } } + pub fn with_partitioning(mut self, partitioning: Partitioning) -> Self { + self.cache = Arc::new(PlanProperties::new( + EquivalenceProperties::new(Arc::clone(&self.schema)), + partitioning, + EmissionType::Incremental, + Boundedness::Bounded, + )); + self + } + /// This function creates the cache object that stores the plan properties such as schema, equivalence properties, ordering, partitioning, etc. fn compute_properties(schema: SchemaRef) -> PlanProperties { PlanProperties::new( diff --git a/datafusion/physical-optimizer/src/join_selection.rs b/datafusion/physical-optimizer/src/join_selection.rs index f11e8612de37..45b8a467c295 100644 --- a/datafusion/physical-optimizer/src/join_selection.rs +++ b/datafusion/physical-optimizer/src/join_selection.rs @@ -182,10 +182,15 @@ fn can_swap_hash_join(hash_join: &HashJoinExec) -> bool { /// Tries to create a [`HashJoinExec`] in [`PartitionMode::CollectLeft`] when possible. /// -/// This function will first consider the given join type and check whether the -/// `CollectLeft` mode is applicable. Otherwise, it will try to swap the join sides. -/// When the `ignore_threshold` is false, this function will also check left -/// and right sizes in bytes or rows. +/// This function will first check whether the join inputs already satisfy the +/// distribution requirements for [`PartitionMode::Partitioned`]. If so, it returns +/// `None` to allow the caller to preserve or select [`PartitionMode::Partitioned`] +/// (task-local join with 0 shuffle) rather than degrading to broadcast (`CollectLeft`). +/// +/// Otherwise, it will consider the given join type and check whether the +/// `CollectLeft` mode is applicable. It will also try to swap the join sides if beneficial. +/// When the `ignore_threshold` is false, this function will check left +/// and right sizes in bytes or rows against the configured thresholds. /// /// Used configurations /// - `optimizer.hash_join_single_partition_threshold`: byte threshold for `CollectLeft` @@ -196,6 +201,10 @@ pub(crate) fn try_collect_left( ignore_threshold: bool, context: &dyn PhysicalOptimizerContext, ) -> Result>> { + if hash_join.inputs_satisfy_partitioned_requirements()? { + return Ok(None); + } + let left = hash_join.left(); let right = hash_join.right(); let optimizer_config = &context.config_options().optimizer; diff --git a/datafusion/physical-plan/src/joins/hash_join/exec.rs b/datafusion/physical-plan/src/joins/hash_join/exec.rs index b72e180543f9..66d24199ecdb 100644 --- a/datafusion/physical-plan/src/joins/hash_join/exec.rs +++ b/datafusion/physical-plan/src/joins/hash_join/exec.rs @@ -1098,6 +1098,45 @@ impl HashJoinExec { self.null_equality } + /// Return the [`InputDistributionRequirements`] that this hash join would + /// require if executed in [`PartitionMode::Partitioned`]. + pub fn partitioned_input_distribution_requirements( + &self, + ) -> InputDistributionRequirements { + let (left_expr, right_expr) = self + .on + .iter() + .map(|(l, r)| (Arc::clone(l), Arc::clone(r))) + .unzip(); + InputDistributionRequirements::co_partitioned(vec![ + Distribution::KeyPartitioned(left_expr), + Distribution::KeyPartitioned(right_expr), + ]) + } + + /// Returns `true` if both inputs have more than 1 partition and already + /// satisfy the distribution and co-partitioning requirements for + /// [`PartitionMode::Partitioned`]. + /// + /// Null-aware joins cannot use [`PartitionMode::Partitioned`] and return `false`. + pub fn inputs_satisfy_partitioned_requirements(&self) -> Result { + if self.null_aware { + return Ok(false); + } + + if self.left.output_partitioning().partition_count() <= 1 + || self.right.output_partitioning().partition_count() <= 1 + { + return Ok(false); + } + + let requirements = self.partitioned_input_distribution_requirements(); + let children = [self.left.as_ref(), self.right.as_ref()]; + let unsatisfied = + requirements.unsatisfied_co_partitioned_children(self.name(), &children)?; + Ok(unsatisfied.is_empty()) + } + /// Returns the dynamic filter expression produced by this hash join, if set. #[deprecated( since = "55.0.0", @@ -1441,15 +1480,7 @@ impl ExecutionPlan for HashJoinExec { fn input_distribution_requirements(&self) -> InputDistributionRequirements { match self.mode { PartitionMode::Partitioned => { - let (left_expr, right_expr) = self - .on - .iter() - .map(|(l, r)| (Arc::clone(l), Arc::clone(r))) - .unzip(); - InputDistributionRequirements::co_partitioned(vec![ - Distribution::KeyPartitioned(left_expr), - Distribution::KeyPartitioned(right_expr), - ]) + self.partitioned_input_distribution_requirements() } PartitionMode::CollectLeft => InputDistributionRequirements::new(vec![ Distribution::SinglePartition, @@ -9114,24 +9145,36 @@ mod tests { Ok((join, on)) } - fn with_hash_partitioned_children( + fn with_partitioned_children( join: &HashJoinExec, - on: &JoinOn, + left_partitioning: Partitioning, + right_partitioning: Partitioning, ) -> Result { join.builder() .with_new_children(vec![ Arc::new(PartitionedTestExec::try_new( join.left().schema(), - Partitioning::Hash(vec![Arc::clone(&on[0].0)], 2), + left_partitioning, )?), Arc::new(PartitionedTestExec::try_new( join.right().schema(), - Partitioning::Hash(vec![Arc::clone(&on[0].1)], 2), + right_partitioning, )?), ])? .build() } + fn with_hash_partitioned_children( + join: &HashJoinExec, + on: &JoinOn, + ) -> Result { + with_partitioned_children( + join, + Partitioning::Hash(vec![Arc::clone(&on[0].0)], 2), + Partitioning::Hash(vec![Arc::clone(&on[0].1)], 2), + ) + } + #[test] fn test_partitioned_dynamic_filter_pushdown_allows_supported_partitioning() -> Result<()> { @@ -9208,4 +9251,80 @@ mod tests { assert!(join.set_dynamic_filter(df).is_err()); Ok(()) } + + #[derive(Clone, Copy, Debug)] + enum PartitionRequirementCase { + MatchingRange, + MatchingHash, + MismatchedRange, + NonJoinKeyHash, + UnknownPartitioning, + SinglePartition, + NullAware, + } + + fn build_partition_case_join( + scenario: PartitionRequirementCase, + ) -> Result { + let (range_join, on) = range_partitioned_dynamic_filter_test_join(10, 10)?; + match scenario { + PartitionRequirementCase::MatchingRange => Ok(range_join), + PartitionRequirementCase::MatchingHash => { + with_hash_partitioned_children(&range_join, &on) + } + PartitionRequirementCase::MismatchedRange => { + let (join, _) = range_partitioned_dynamic_filter_test_join(10, 11)?; + Ok(join) + } + PartitionRequirementCase::NonJoinKeyHash => { + let non_join_key_left = + Arc::new(Column::new_with_schema("a1", &range_join.left().schema())?) + as _; + let non_join_key_right = Arc::new(Column::new_with_schema( + "a2", + &range_join.right().schema(), + )?) as _; + with_partitioned_children( + &range_join, + Partitioning::Hash(vec![non_join_key_left], 2), + Partitioning::Hash(vec![non_join_key_right], 2), + ) + } + PartitionRequirementCase::UnknownPartitioning => with_partitioned_children( + &range_join, + Partitioning::UnknownPartitioning(2), + Partitioning::UnknownPartitioning(2), + ), + PartitionRequirementCase::SinglePartition => with_partitioned_children( + &range_join, + Partitioning::Hash(vec![Arc::clone(&on[0].0)], 1), + Partitioning::Hash(vec![Arc::clone(&on[0].1)], 1), + ), + PartitionRequirementCase::NullAware => { + let hash_join = with_hash_partitioned_children(&range_join, &on)?; + hash_join + .builder() + .with_type(JoinType::LeftAnti) + .with_null_aware(true) + .build() + } + } + } + + #[rstest] + #[case::matching_range(PartitionRequirementCase::MatchingRange, true)] + #[case::matching_hash(PartitionRequirementCase::MatchingHash, true)] + #[case::mismatched_range(PartitionRequirementCase::MismatchedRange, false)] + #[case::non_join_key_hash(PartitionRequirementCase::NonJoinKeyHash, false)] + #[case::unknown_partitioning(PartitionRequirementCase::UnknownPartitioning, false)] + #[case::single_partition(PartitionRequirementCase::SinglePartition, false)] + #[case::null_aware(PartitionRequirementCase::NullAware, false)] + fn test_inputs_satisfy_partitioned_requirements( + #[case] scenario: PartitionRequirementCase, + #[case] expected: bool, + ) -> Result<()> { + let join = build_partition_case_join(scenario)?; + assert_eq!(join.inputs_satisfy_partitioned_requirements()?, expected); + Ok(()) + } } diff --git a/datafusion/sqllogictest/test_files/repartition_subset_satisfaction.slt b/datafusion/sqllogictest/test_files/repartition_subset_satisfaction.slt index 5371ca59beea..0394a51ab321 100644 --- a/datafusion/sqllogictest/test_files/repartition_subset_satisfaction.slt +++ b/datafusion/sqllogictest/test_files/repartition_subset_satisfaction.slt @@ -376,11 +376,10 @@ physical_plan 08)--------------AggregateExec: mode=FinalPartitioned, gby=[f_dkey@0 as f_dkey, date_bin(IntervalMonthDayNano("IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 30000000000 }"),j.timestamp)@1 as date_bin(IntervalMonthDayNano("IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 30000000000 }"),j.timestamp), env@2 as env], aggr=[max(j.value)], ordering_mode=PartiallySorted([0, 1]) 09)----------------RepartitionExec: partitioning=Hash([f_dkey@0, date_bin(IntervalMonthDayNano("IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 30000000000 }"),j.timestamp)@1, env@2], 3), input_partitions=3, preserve_order=true, sort_exprs=f_dkey@0 ASC NULLS LAST, date_bin(IntervalMonthDayNano("IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 30000000000 }"),j.timestamp)@1 ASC NULLS LAST 10)------------------AggregateExec: mode=Partial, gby=[f_dkey@0 as f_dkey, date_bin(IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 30000000000 }, timestamp@2) as date_bin(IntervalMonthDayNano("IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 30000000000 }"),j.timestamp), env@1 as env], aggr=[max(j.value)], ordering_mode=PartiallySorted([0, 1]) -11)--------------------HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(d_dkey@1, f_dkey@2)], projection=[f_dkey@4, env@0, timestamp@2, value@3] -12)----------------------CoalescePartitionsExec -13)------------------------FilterExec: service@1 = log, projection=[env@0, d_dkey@2] -14)--------------------------DataSourceExec: file_groups={3 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/dimension/d_dkey=A/data.parquet, WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/dimension/d_dkey=D/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/dimension/d_dkey=B/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/dimension/d_dkey=C/data.parquet]]}, projection=[env, service, d_dkey], output_partitioning=Hash([d_dkey@2], 3), file_type=parquet, predicate=service@1 = log, pruning_predicate=service_null_count@2 != row_count@3 AND service_min@0 <= log AND log <= service_max@1, required_guarantees=[service in (log)] -15)----------------------DataSourceExec: file_groups={3 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/fact/f_dkey=A/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/fact/f_dkey=B/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/fact/f_dkey=C/data.parquet]]}, projection=[timestamp, value, f_dkey], output_ordering=[f_dkey@2 ASC NULLS LAST, timestamp@0 ASC NULLS LAST], output_partitioning=Hash([f_dkey@2], 3), file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible +11)--------------------HashJoinExec: mode=Partitioned, join_type=Inner, on=[(d_dkey@1, f_dkey@2)], projection=[f_dkey@4, env@0, timestamp@2, value@3] +12)----------------------FilterExec: service@1 = log, projection=[env@0, d_dkey@2] +13)------------------------DataSourceExec: file_groups={3 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/dimension/d_dkey=A/data.parquet, WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/dimension/d_dkey=D/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/dimension/d_dkey=B/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/dimension/d_dkey=C/data.parquet]]}, projection=[env, service, d_dkey], output_partitioning=Hash([d_dkey@2], 3), file_type=parquet, predicate=service@1 = log, pruning_predicate=service_null_count@2 != row_count@3 AND service_min@0 <= log AND log <= service_max@1, required_guarantees=[service in (log)] +14)----------------------DataSourceExec: file_groups={3 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/fact/f_dkey=A/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/fact/f_dkey=B/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/fact/f_dkey=C/data.parquet]]}, projection=[timestamp, value, f_dkey], output_ordering=[f_dkey@2 ASC NULLS LAST, timestamp@0 ASC NULLS LAST], output_partitioning=Hash([f_dkey@2], 3), file_type=parquet # Verify results without subset satisfaction query TPR rowsort @@ -471,11 +470,10 @@ physical_plan 06)----------AggregateExec: mode=Partial, gby=[env@1 as env, time_bin@0 as time_bin], aggr=[avg(a.max_bin_value)] 07)------------ProjectionExec: expr=[date_bin(IntervalMonthDayNano("IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 30000000000 }"),j.timestamp)@1 as time_bin, env@2 as env, max(j.value)@3 as max_bin_value] 08)--------------AggregateExec: mode=SinglePartitioned, gby=[f_dkey@0 as f_dkey, date_bin(IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 30000000000 }, timestamp@2) as date_bin(IntervalMonthDayNano("IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 30000000000 }"),j.timestamp), env@1 as env], aggr=[max(j.value)], ordering_mode=PartiallySorted([0, 1]) -09)----------------HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(d_dkey@1, f_dkey@2)], projection=[f_dkey@4, env@0, timestamp@2, value@3] -10)------------------CoalescePartitionsExec -11)--------------------FilterExec: service@1 = log, projection=[env@0, d_dkey@2] -12)----------------------DataSourceExec: file_groups={3 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/dimension/d_dkey=A/data.parquet, WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/dimension/d_dkey=D/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/dimension/d_dkey=B/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/dimension/d_dkey=C/data.parquet]]}, projection=[env, service, d_dkey], output_partitioning=Hash([d_dkey@2], 3), file_type=parquet, predicate=service@1 = log, pruning_predicate=service_null_count@2 != row_count@3 AND service_min@0 <= log AND log <= service_max@1, required_guarantees=[service in (log)] -13)------------------DataSourceExec: file_groups={3 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/fact/f_dkey=A/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/fact/f_dkey=B/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/fact/f_dkey=C/data.parquet]]}, projection=[timestamp, value, f_dkey], output_ordering=[f_dkey@2 ASC NULLS LAST, timestamp@0 ASC NULLS LAST], output_partitioning=Hash([f_dkey@2], 3), file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible +09)----------------HashJoinExec: mode=Partitioned, join_type=Inner, on=[(d_dkey@1, f_dkey@2)], projection=[f_dkey@4, env@0, timestamp@2, value@3] +10)------------------FilterExec: service@1 = log, projection=[env@0, d_dkey@2] +11)--------------------DataSourceExec: file_groups={3 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/dimension/d_dkey=A/data.parquet, WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/dimension/d_dkey=D/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/dimension/d_dkey=B/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/dimension/d_dkey=C/data.parquet]]}, projection=[env, service, d_dkey], output_partitioning=Hash([d_dkey@2], 3), file_type=parquet, predicate=service@1 = log, pruning_predicate=service_null_count@2 != row_count@3 AND service_min@0 <= log AND log <= service_max@1, required_guarantees=[service in (log)] +12)------------------DataSourceExec: file_groups={3 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/fact/f_dkey=A/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/fact/f_dkey=B/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/fact/f_dkey=C/data.parquet]]}, projection=[timestamp, value, f_dkey], output_ordering=[f_dkey@2 ASC NULLS LAST, timestamp@0 ASC NULLS LAST], output_partitioning=Hash([f_dkey@2], 3), file_type=parquet # Verify results match with subset satisfaction query TPR rowsort