You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/07/14 18:42:03 UTC
(datafusion) 01/01: feat: allow Partitioning::Range to satisfy window Distribution::KeyPartitioned requirements (#23416)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-23416-d1e6d4576c4d279087610a91a840ca53a973c90c
in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit 4b68de61d37839ec40f12d3c1ad8995d5106bbf0
Author: Mithun Chicklore Yogendra <mi...@gmail.com>
AuthorDate: Wed Jul 15 00:11:38 2026 +0530
feat: allow Partitioning::Range to satisfy window Distribution::KeyPartitioned requirements (#23416)
## Which issue does this PR close?
- Closes #23289.
Rebased on main now that #23184 has merged. Overlaps with #23355, which
also opts the window execs into range satisfaction as part of its
PartitionedTopK work — happy to rebase whichever lands second.
## What changes are included in this PR?
In one commit:
* Opt `WindowAggExec` and `BoundedWindowAggExec` with partition keys
into range satisfaction via
`InputDistributionRequirements::allow_range_satisfaction_for_key_partitioning`,
mirroring `AggregateExec`. Compatible range-partitioned inputs then
satisfy the window key requirement without a hash repartition; subset
satisfaction and the hash fallback for incompatible keys come from the
existing satisfaction machinery. Windows without `PARTITION BY` keep
requiring a single partition.
## Are these changes tested?
Yes: new `slt` tests in `range_partitioning.slt` (exact and subset
reuse, rehash on incompatible keys, `subset_repartition_threshold` /
`preserve_file_partitions` / `target_partitions` behavior,
`WindowAggExec` via an unbounded frame, and no-`PARTITION BY`), plus
plan-shape tests in `enforce_distribution.rs` and acceptance/rejection
tests in `sanity_checker.rs`.
## Are there any user-facing changes?
No: only plan changes.
---
.../physical_optimizer/enforce_distribution.rs | 67 +++++-
.../tests/physical_optimizer/sanity_checker.rs | 77 ++++++-
.../core/tests/physical_optimizer/test_utils.rs | 18 +-
.../src/windows/bounded_window_agg_exec.rs | 18 +-
.../physical-plan/src/windows/window_agg_exec.rs | 18 +-
.../sqllogictest/test_files/range_partitioning.slt | 225 +++++++++++++++++++++
6 files changed, 403 insertions(+), 20 deletions(-)
diff --git a/datafusion/core/tests/physical_optimizer/enforce_distribution.rs b/datafusion/core/tests/physical_optimizer/enforce_distribution.rs
index e01311e25b..3292ada0a8 100644
--- a/datafusion/core/tests/physical_optimizer/enforce_distribution.rs
+++ b/datafusion/core/tests/physical_optimizer/enforce_distribution.rs
@@ -20,8 +20,8 @@ use std::ops::Deref;
use std::sync::Arc;
use crate::physical_optimizer::test_utils::{
- check_integrity, coalesce_partitions_exec, parquet_exec_with_sort,
- parquet_exec_with_stats, repartition_exec, schema, sort_exec,
+ 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,
};
@@ -870,6 +870,69 @@ fn range_inner_hash_join_rehashes_incompatible_range_partitioning() -> Result<()
Ok(())
}
+#[test]
+fn range_window_reuses_range_partitioning() -> Result<()> {
+ let input = parquet_exec_with_output_partitioning(range_partitioning(
+ "a",
+ [10, 20, 30],
+ SortOptions::default(),
+ )?);
+ let window = bounded_window_exec_with_can_repartition(
+ "a",
+ vec![],
+ &[col("a", &schema())?],
+ input,
+ true,
+ );
+
+ let plan = TestConfig::default()
+ .with_query_execution_partitions(4)
+ .to_plan(window, &DISTRIB_DISTRIB_SORT);
+
+ assert_plan!(
+ plan,
+ @r#"
+ BoundedWindowAggExec: wdw=[count: Field { "count": Int64 }, frame: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW], mode=[Sorted]
+ SortExec: expr=[a@0 ASC NULLS LAST], preserve_partitioning=[true]
+ DataSourceExec: file_groups={4 groups: [[p0], [p1], [p2], [p3]]}, projection=[a, b, c, d, e], output_partitioning=Range([a@0 ASC], [(10), (20), (30)], 4), file_type=parquet
+ "#
+ );
+
+ Ok(())
+}
+
+#[test]
+fn range_window_rehashes_incompatible_range_partitioning() -> Result<()> {
+ let input = parquet_exec_with_output_partitioning(range_partitioning(
+ "a",
+ [10, 20, 30],
+ SortOptions::default(),
+ )?);
+ let window = bounded_window_exec_with_can_repartition(
+ "b",
+ vec![],
+ &[col("b", &schema())?],
+ input,
+ true,
+ );
+
+ let plan = TestConfig::default()
+ .with_query_execution_partitions(4)
+ .to_plan(window, &DISTRIB_DISTRIB_SORT);
+
+ assert_plan!(
+ plan,
+ @r#"
+ BoundedWindowAggExec: wdw=[count: Field { "count": Int64 }, frame: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW], mode=[Sorted]
+ SortExec: expr=[b@1 ASC NULLS LAST], preserve_partitioning=[true]
+ RepartitionExec: partitioning=Hash([b@1], 4), input_partitions=4
+ DataSourceExec: file_groups={4 groups: [[p0], [p1], [p2], [p3]]}, projection=[a, b, c, d, e], output_partitioning=Range([a@0 ASC], [(10), (20), (30)], 4), file_type=parquet
+ "#
+ );
+
+ Ok(())
+}
+
#[test]
fn multi_hash_joins() -> Result<()> {
let left = parquet_exec();
diff --git a/datafusion/core/tests/physical_optimizer/sanity_checker.rs b/datafusion/core/tests/physical_optimizer/sanity_checker.rs
index e759156282..e5718f5b3d 100644
--- a/datafusion/core/tests/physical_optimizer/sanity_checker.rs
+++ b/datafusion/core/tests/physical_optimizer/sanity_checker.rs
@@ -19,9 +19,10 @@ use insta::assert_snapshot;
use std::sync::Arc;
use crate::physical_optimizer::test_utils::{
- bounded_window_exec, global_limit_exec, hash_join_exec, local_limit_exec,
- memory_exec, projection_exec, repartition_exec, sort_exec, sort_expr,
- sort_expr_options, sort_merge_join_exec, sort_preserving_merge_exec, union_exec,
+ bounded_window_exec, bounded_window_exec_with_can_repartition, global_limit_exec,
+ hash_join_exec, local_limit_exec, memory_exec, projection_exec, repartition_exec,
+ sort_exec, sort_exec_with_preserve_partitioning, sort_expr, sort_expr_options,
+ sort_merge_join_exec, sort_preserving_merge_exec, union_exec,
};
use arrow::compute::SortOptions;
@@ -501,6 +502,76 @@ async fn test_bounded_window_agg_no_sort_requirement() -> Result<()> {
Ok(())
}
+#[tokio::test]
+/// Tests that a window over a compatible range-partitioned input satisfies
+/// the window's key distribution requirement without a hash repartition.
+async fn test_bounded_window_agg_range_partitioning() -> Result<()> {
+ let schema = create_test_schema2();
+ let source = range_partitioned_exec(&schema, "a", [10, 20, 30])?;
+ let ordering: LexOrdering = [sort_expr_options(
+ "a",
+ &schema,
+ SortOptions {
+ descending: false,
+ nulls_first: false,
+ },
+ )]
+ .into();
+ let partition_by = vec![col("a", &schema)?];
+ let sort = sort_exec_with_preserve_partitioning(ordering, source);
+ let bw =
+ bounded_window_exec_with_can_repartition("a", vec![], &partition_by, sort, true);
+ let plan_str = displayable(bw.as_ref()).indent(true).to_string();
+ let actual = plan_str.trim();
+ assert_snapshot!(
+ actual,
+ @r#"
+ BoundedWindowAggExec: wdw=[count: Field { "count": Int64 }, frame: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW], mode=[Sorted]
+ SortExec: expr=[a@0 ASC NULLS LAST], preserve_partitioning=[true]
+ RepartitionExec: partitioning=Range([a@0 ASC], [(10), (20), (30)], 4), input_partitions=1
+ DataSourceExec: partitions=1, partition_sizes=[0]
+ "#
+ );
+ assert_sanity_check(&bw, true);
+ Ok(())
+}
+
+#[tokio::test]
+/// Tests that a window over an incompatible range-partitioned input fails
+/// the window's key distribution requirement.
+async fn test_bounded_window_agg_incompatible_range_partitioning() -> Result<()> {
+ let schema = create_test_schema2();
+ let source = range_partitioned_exec(&schema, "a", [10, 20, 30])?;
+ let ordering: LexOrdering = [sort_expr_options(
+ "b",
+ &schema,
+ SortOptions {
+ descending: false,
+ nulls_first: false,
+ },
+ )]
+ .into();
+ let partition_by = vec![col("b", &schema)?];
+ let sort = sort_exec_with_preserve_partitioning(ordering, source);
+ let bw =
+ bounded_window_exec_with_can_repartition("b", vec![], &partition_by, sort, true);
+ let plan_str = displayable(bw.as_ref()).indent(true).to_string();
+ let actual = plan_str.trim();
+ assert_snapshot!(
+ actual,
+ @r#"
+ BoundedWindowAggExec: wdw=[count: Field { "count": Int64 }, frame: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW], mode=[Sorted]
+ SortExec: expr=[b@1 ASC NULLS LAST], preserve_partitioning=[true]
+ RepartitionExec: partitioning=Range([a@0 ASC], [(10), (20), (30)], 4), input_partitions=1
+ DataSourceExec: partitions=1, partition_sizes=[0]
+ "#
+ );
+ // Range([a]) does not colocate `b` values, so the window's key
+ // distribution requirement is not satisfied.
+ assert_sanity_check(&bw, false);
+ Ok(())
+}
+
#[tokio::test]
/// A valid when a single partition requirement
/// is satisfied.
diff --git a/datafusion/core/tests/physical_optimizer/test_utils.rs b/datafusion/core/tests/physical_optimizer/test_utils.rs
index 915bd9a05f..74230b24e2 100644
--- a/datafusion/core/tests/physical_optimizer/test_utils.rs
+++ b/datafusion/core/tests/physical_optimizer/test_utils.rs
@@ -274,6 +274,22 @@ pub fn bounded_window_exec_with_partition(
sort_exprs: impl IntoIterator<Item = PhysicalSortExpr>,
partition_by: &[Arc<dyn PhysicalExpr>],
input: Arc<dyn ExecutionPlan>,
+) -> Arc<dyn ExecutionPlan> {
+ bounded_window_exec_with_can_repartition(
+ col_name,
+ sort_exprs,
+ partition_by,
+ input,
+ false,
+ )
+}
+
+pub fn bounded_window_exec_with_can_repartition(
+ col_name: &str,
+ sort_exprs: impl IntoIterator<Item = PhysicalSortExpr>,
+ partition_by: &[Arc<dyn PhysicalExpr>],
+ input: Arc<dyn ExecutionPlan>,
+ can_repartition: bool,
) -> Arc<dyn ExecutionPlan> {
let sort_exprs = sort_exprs.into_iter().collect::<Vec<_>>();
let schema = input.schema();
@@ -296,7 +312,7 @@ pub fn bounded_window_exec_with_partition(
vec![window_expr],
Arc::clone(&input),
InputOrderMode::Sorted,
- false,
+ can_repartition,
)
.unwrap(),
)
diff --git a/datafusion/physical-plan/src/windows/bounded_window_agg_exec.rs b/datafusion/physical-plan/src/windows/bounded_window_agg_exec.rs
index bb475edbfb..97cafd24c0 100644
--- a/datafusion/physical-plan/src/windows/bounded_window_agg_exec.rs
+++ b/datafusion/physical-plan/src/windows/bounded_window_agg_exec.rs
@@ -36,8 +36,9 @@ use crate::windows::{
};
use crate::{
ColumnStatistics, DisplayAs, DisplayFormatType, Distribution, ExecutionPlan,
- ExecutionPlanProperties, InputOrderMode, PlanProperties, RecordBatchStream,
- SendableRecordBatchStream, Statistics, WindowExpr, check_if_same_properties,
+ ExecutionPlanProperties, InputDistributionRequirements, InputOrderMode,
+ PlanProperties, RecordBatchStream, SendableRecordBatchStream, Statistics, WindowExpr,
+ check_if_same_properties,
};
use arrow::compute::take_record_batch;
@@ -324,13 +325,16 @@ impl ExecutionPlan for BoundedWindowAggExec {
self.input_distribution_requirements().into_per_child()
}
- fn input_distribution_requirements(&self) -> crate::InputDistributionRequirements {
- crate::InputDistributionRequirements::new(if self.partition_keys().is_empty() {
+ fn input_distribution_requirements(&self) -> InputDistributionRequirements {
+ if self.partition_keys().is_empty() {
debug!("No partition defined for BoundedWindowAggExec!!!");
- vec![Distribution::SinglePartition]
+ InputDistributionRequirements::new(vec![Distribution::SinglePartition])
} else {
- vec![Distribution::KeyPartitioned(self.partition_keys().clone())]
- })
+ InputDistributionRequirements::new(vec![Distribution::KeyPartitioned(
+ self.partition_keys(),
+ )])
+ .allow_range_satisfaction_for_key_partitioning()
+ }
}
fn maintains_input_order(&self) -> Vec<bool> {
diff --git a/datafusion/physical-plan/src/windows/window_agg_exec.rs b/datafusion/physical-plan/src/windows/window_agg_exec.rs
index bae7cfcfd8..4e8dbc06f0 100644
--- a/datafusion/physical-plan/src/windows/window_agg_exec.rs
+++ b/datafusion/physical-plan/src/windows/window_agg_exec.rs
@@ -32,8 +32,9 @@ use crate::windows::{
};
use crate::{
ColumnStatistics, DisplayAs, DisplayFormatType, Distribution, ExecutionPlan,
- ExecutionPlanProperties, PhysicalExpr, PlanProperties, RecordBatchStream,
- SendableRecordBatchStream, Statistics, WindowExpr, check_if_same_properties,
+ ExecutionPlanProperties, InputDistributionRequirements, PhysicalExpr, PlanProperties,
+ RecordBatchStream, SendableRecordBatchStream, Statistics, WindowExpr,
+ check_if_same_properties,
};
use arrow::array::ArrayRef;
@@ -233,12 +234,15 @@ impl ExecutionPlan for WindowAggExec {
self.input_distribution_requirements().into_per_child()
}
- fn input_distribution_requirements(&self) -> crate::InputDistributionRequirements {
- crate::InputDistributionRequirements::new(if self.partition_keys().is_empty() {
- vec![Distribution::SinglePartition]
+ fn input_distribution_requirements(&self) -> InputDistributionRequirements {
+ if self.partition_keys().is_empty() {
+ InputDistributionRequirements::new(vec![Distribution::SinglePartition])
} else {
- vec![Distribution::KeyPartitioned(self.partition_keys())]
- })
+ InputDistributionRequirements::new(vec![Distribution::KeyPartitioned(
+ self.partition_keys(),
+ )])
+ .allow_range_satisfaction_for_key_partitioning()
+ }
}
fn with_new_children(
diff --git a/datafusion/sqllogictest/test_files/range_partitioning.slt b/datafusion/sqllogictest/test_files/range_partitioning.slt
index 5d004ca7fd..2cf3b87e3c 100644
--- a/datafusion/sqllogictest/test_files/range_partitioning.slt
+++ b/datafusion/sqllogictest/test_files/range_partitioning.slt
@@ -682,5 +682,230 @@ ORDER BY range_key, value;
35 350
35 350
+statement ok
+set datafusion.execution.target_partitions = 4;
+
+statement ok
+set datafusion.optimizer.subset_repartition_threshold = 4;
+
+statement ok
+set datafusion.optimizer.preserve_file_partitions = 0;
+
+
+##########
+# TEST 19: Window on Range Partition Column
+# Range([range_key]) colocates equal range_key values, so
+# PARTITION BY range_key is satisfied without a hash repartition.
+##########
+
+query TT
+EXPLAIN SELECT range_key, SUM(value) OVER (PARTITION BY range_key ORDER BY value) FROM range_partitioned;
+----
+physical_plan
+01)ProjectionExec: expr=[range_key@0 as range_key, sum(range_partitioned.value) PARTITION BY [range_partitioned.range_key] ORDER BY [range_partitioned.value ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW@2 as sum(range_partitioned.value) PARTITION BY [range_partitioned.range_key] ORDER BY [range_partitioned.value ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW]
+02)--BoundedWindowAggExec: wdw=[sum(range_partitioned.value) PARTITION BY [range_partitioned.range_key] ORDER BY [range_partitioned.value ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW: Field { "sum(range_partitioned.value) PARTITION BY [range_partitioned.range_key] ORDER BY [range_partitioned.value ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW": nullable Int64 }, frame: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW], mode=[Sorted]
+03)----SortExec: expr=[range_key@0 ASC NULLS LAST, value@1 ASC NULLS LAST], preserve_partitioning=[true]
+04)------DataSourceExec: file_groups=<slt:ignore>, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false
+
+query II
+SELECT range_key, SUM(value) OVER (PARTITION BY range_key ORDER BY value) FROM range_partitioned ORDER BY range_key;
+----
+1 10
+5 50
+10 100
+15 150
+20 200
+25 250
+30 300
+35 350
+
+
+##########
+# TEST 20: 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.
+##########
+
+query TT
+EXPLAIN SELECT range_key, SUM(value) OVER (PARTITION BY range_key ORDER BY value ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING) FROM range_partitioned;
+----
+physical_plan
+01)ProjectionExec: expr=[range_key@0 as range_key, sum(range_partitioned.value) PARTITION BY [range_partitioned.range_key] ORDER BY [range_partitioned.value ASC NULLS LAST] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING@2 as sum(range_partitioned.value) PARTITION BY [range_partitioned.range_key] ORDER BY [range_partitioned.value ASC NULLS LAST] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING]
+02)--WindowAggExec: wdw=[sum(range_partitioned.value) PARTITION BY [range_partitioned.range_key] ORDER BY [range_partitioned.value ASC NULLS LAST] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING: Ok(Field { name: "sum(range_partitioned.value) PARTITION BY [range_partitioned.range_key] ORDER BY [range_partitioned.value ASC NULLS LAST] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING", data_type: Int64, nullable: true }), frame: WindowFrame { units: Rows, start_bound: Prec [...]
+03)----SortExec: expr=[range_key@0 ASC NULLS LAST, value@1 ASC NULLS LAST], preserve_partitioning=[true]
+04)------DataSourceExec: file_groups=<slt:ignore>, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false
+
+query II
+SELECT range_key, SUM(value) OVER (PARTITION BY range_key ORDER BY value ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING) FROM range_partitioned ORDER BY range_key;
+----
+1 10
+5 50
+10 100
+15 150
+20 200
+25 250
+30 300
+35 350
+
+
+##########
+# TEST 21: 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.
+##########
+
+query TT
+EXPLAIN SELECT non_range_key, SUM(value) OVER (PARTITION BY non_range_key ORDER BY value) FROM range_partitioned;
+----
+physical_plan
+01)ProjectionExec: expr=[non_range_key@0 as non_range_key, sum(range_partitioned.value) PARTITION BY [range_partitioned.non_range_key] ORDER BY [range_partitioned.value ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW@2 as sum(range_partitioned.value) PARTITION BY [range_partitioned.non_range_key] ORDER BY [range_partitioned.value ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW]
+02)--BoundedWindowAggExec: wdw=[sum(range_partitioned.value) PARTITION BY [range_partitioned.non_range_key] ORDER BY [range_partitioned.value ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW: Field { "sum(range_partitioned.value) PARTITION BY [range_partitioned.non_range_key] ORDER BY [range_partitioned.value ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW": nullable Int64 }, frame: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW], mode=[Sorted]
+03)----SortExec: expr=[non_range_key@0 ASC NULLS LAST, value@1 ASC NULLS LAST], preserve_partitioning=[true]
+04)------RepartitionExec: partitioning=Hash([non_range_key@0], 4), input_partitions=4
+05)--------DataSourceExec: file_groups=<slt:ignore>, projection=[non_range_key, value], output_partitioning=UnknownPartitioning(4), file_type=csv, has_header=false
+
+query III
+SELECT non_range_key, value, SUM(value) OVER (PARTITION BY non_range_key ORDER BY value) FROM range_partitioned ORDER BY non_range_key, value;
+----
+1 10 10
+1 100 110
+1 200 310
+1 300 610
+2 50 50
+2 150 200
+2 250 450
+2 350 800
+
+
+##########
+# TEST 22: 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.
+##########
+
+query TT
+EXPLAIN SELECT non_range_key, SUM(value) OVER (PARTITION BY non_range_key ORDER BY value ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING) FROM range_partitioned;
+----
+physical_plan
+01)ProjectionExec: expr=[non_range_key@0 as non_range_key, sum(range_partitioned.value) PARTITION BY [range_partitioned.non_range_key] ORDER BY [range_partitioned.value ASC NULLS LAST] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING@2 as sum(range_partitioned.value) PARTITION BY [range_partitioned.non_range_key] ORDER BY [range_partitioned.value ASC NULLS LAST] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING]
+02)--WindowAggExec: wdw=[sum(range_partitioned.value) PARTITION BY [range_partitioned.non_range_key] ORDER BY [range_partitioned.value ASC NULLS LAST] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING: Ok(Field { name: "sum(range_partitioned.value) PARTITION BY [range_partitioned.non_range_key] ORDER BY [range_partitioned.value ASC NULLS LAST] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING", data_type: Int64, nullable: true }), frame: WindowFrame { units: Rows, start_bou [...]
+03)----SortExec: expr=[non_range_key@0 ASC NULLS LAST, value@1 ASC NULLS LAST], preserve_partitioning=[true]
+04)------RepartitionExec: partitioning=Hash([non_range_key@0], 4), input_partitions=4
+05)--------DataSourceExec: file_groups=<slt:ignore>, projection=[non_range_key, value], output_partitioning=UnknownPartitioning(4), file_type=csv, has_header=false
+
+query III
+SELECT non_range_key, value, SUM(value) OVER (PARTITION BY non_range_key ORDER BY value ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING) FROM range_partitioned ORDER BY non_range_key, value;
+----
+1 10 610
+1 100 610
+1 200 610
+1 300 610
+2 50 800
+2 150 800
+2 250 800
+2 350 800
+
+
+##########
+# TEST 23: 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.
+##########
+
+statement ok
+set datafusion.optimizer.subset_repartition_threshold = 4;
+
+query TT
+EXPLAIN SELECT range_key, SUM(value) OVER (PARTITION BY range_key, non_range_key ORDER BY value) FROM range_partitioned;
+----
+physical_plan
+01)ProjectionExec: expr=[range_key@0 as range_key, sum(range_partitioned.value) PARTITION BY [range_partitioned.range_key, range_partitioned.non_range_key] ORDER BY [range_partitioned.value ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW@3 as sum(range_partitioned.value) PARTITION BY [range_partitioned.range_key, range_partitioned.non_range_key] ORDER BY [range_partitioned.value ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW]
+02)--BoundedWindowAggExec: wdw=[sum(range_partitioned.value) PARTITION BY [range_partitioned.range_key, range_partitioned.non_range_key] ORDER BY [range_partitioned.value ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW: Field { "sum(range_partitioned.value) PARTITION BY [range_partitioned.range_key, range_partitioned.non_range_key] ORDER BY [range_partitioned.value ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW": nullable Int64 }, frame: RANGE BETWEEN [...]
+03)----SortExec: expr=[range_key@0 ASC NULLS LAST, non_range_key@1 ASC NULLS LAST, value@2 ASC NULLS LAST], preserve_partitioning=[true]
+04)------DataSourceExec: file_groups=<slt:ignore>, projection=[range_key, non_range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false
+
+query II
+SELECT range_key, SUM(value) OVER (PARTITION BY range_key, non_range_key ORDER BY value) FROM range_partitioned ORDER BY range_key;
+----
+1 10
+5 50
+10 100
+15 150
+20 200
+25 250
+30 300
+35 350
+
+
+##########
+# TEST 24: 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.
+##########
+
+statement ok
+set datafusion.optimizer.subset_repartition_threshold = 5;
+
+query TT
+EXPLAIN SELECT range_key, SUM(value) OVER (PARTITION BY range_key, non_range_key ORDER BY value) FROM range_partitioned;
+----
+physical_plan
+01)ProjectionExec: expr=[range_key@0 as range_key, sum(range_partitioned.value) PARTITION BY [range_partitioned.range_key, range_partitioned.non_range_key] ORDER BY [range_partitioned.value ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW@3 as sum(range_partitioned.value) PARTITION BY [range_partitioned.range_key, range_partitioned.non_range_key] ORDER BY [range_partitioned.value ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW]
+02)--BoundedWindowAggExec: wdw=[sum(range_partitioned.value) PARTITION BY [range_partitioned.range_key, range_partitioned.non_range_key] ORDER BY [range_partitioned.value ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW: Field { "sum(range_partitioned.value) PARTITION BY [range_partitioned.range_key, range_partitioned.non_range_key] ORDER BY [range_partitioned.value ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW": nullable Int64 }, frame: RANGE BETWEEN [...]
+03)----SortExec: expr=[range_key@0 ASC NULLS LAST, non_range_key@1 ASC NULLS LAST, value@2 ASC NULLS LAST], preserve_partitioning=[true]
+04)------RepartitionExec: partitioning=Hash([range_key@0, non_range_key@1], 4), input_partitions=4
+05)--------DataSourceExec: file_groups=<slt:ignore>, projection=[range_key, non_range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false
+
+query II
+SELECT range_key, SUM(value) OVER (PARTITION BY range_key, non_range_key ORDER BY value) FROM range_partitioned ORDER BY range_key;
+----
+1 10
+5 50
+10 100
+15 150
+20 200
+25 250
+30 300
+35 350
+
+statement ok
+reset datafusion.optimizer.subset_repartition_threshold;
+
+statement ok
+reset datafusion.optimizer.preserve_file_partitions;
+
+
+##########
+# TEST 25: Window Without Partition Keys Uses a Single Partition
+# A window with no PARTITION BY requires a single partition; range
+# partitioning is not applicable.
+##########
+
+query TT
+EXPLAIN SELECT range_key, SUM(value) OVER (ORDER BY value) FROM range_partitioned;
+----
+physical_plan
+01)ProjectionExec: expr=[range_key@0 as range_key, sum(range_partitioned.value) ORDER BY [range_partitioned.value ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW@2 as sum(range_partitioned.value) ORDER BY [range_partitioned.value ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW]
+02)--BoundedWindowAggExec: wdw=[sum(range_partitioned.value) ORDER BY [range_partitioned.value ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW: Field { "sum(range_partitioned.value) ORDER BY [range_partitioned.value ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW": nullable Int64 }, frame: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW], mode=[Sorted]
+03)----SortPreservingMergeExec: [value@1 ASC NULLS LAST]
+04)------SortExec: expr=[value@1 ASC NULLS LAST], preserve_partitioning=[true]
+05)--------DataSourceExec: file_groups=<slt:ignore>, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false
+
+query II
+SELECT range_key, SUM(value) OVER (ORDER BY value) FROM range_partitioned ORDER BY range_key;
+----
+1 10
+5 60
+10 160
+15 310
+20 510
+25 760
+30 1060
+35 1410
+
statement ok
reset datafusion.explain.physical_plan_only;
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org