feat: allow Range partitioning to satisfy KeyPartitioned requirement for Window/TopK

Author: stuhood

datafusion/physical-plan/src/sorts/partitioned_topk.rs

@@ -311,6 +311,7 @@ impl ExecutionPlan for PartitionedTopKExec {
         crate::InputDistributionRequirements::new(vec![Distribution::KeyPartitioned(
             partition_exprs,
         )])
+        .allow_range_satisfaction_for_key_partitioning()
     }
 
     fn maintains_input_order(&self) -> Vec<bool> {

datafusion/physical-plan/src/windows/bounded_window_agg_exec.rs

@@ -37,7 +37,8 @@ use crate::windows::{
 use crate::{
     ColumnStatistics, DisplayAs, DisplayFormatType, Distribution, ExecutionPlan,
     ExecutionPlanProperties, InputOrderMode, PlanProperties, RecordBatchStream,
-    SendableRecordBatchStream, Statistics, WindowExpr, check_if_same_properties,
+    SendableRecordBatchStream, Statistics, WindowExpr,
+    check_if_same_properties,
 };
 
 use arrow::compute::take_record_batch;
@@ -342,6 +343,7 @@ impl ExecutionPlan for BoundedWindowAggExec {
         } else {
             vec![Distribution::KeyPartitioned(self.partition_keys().clone())]
         })
+        .allow_range_satisfaction_for_key_partitioning()
     }
 
     fn maintains_input_order(&self) -> Vec<bool> {

datafusion/physical-plan/src/windows/window_agg_exec.rs

@@ -33,7 +33,8 @@ use crate::windows::{
 use crate::{
     ColumnStatistics, DisplayAs, DisplayFormatType, Distribution, ExecutionPlan,
     ExecutionPlanProperties, PhysicalExpr, PlanProperties, RecordBatchStream,
-    SendableRecordBatchStream, Statistics, WindowExpr, check_if_same_properties,
+    SendableRecordBatchStream, Statistics, WindowExpr,
+    check_if_same_properties,
 };
 
 use arrow::array::ArrayRef;
@@ -250,6 +251,7 @@ impl ExecutionPlan for WindowAggExec {
         } else {
             vec![Distribution::KeyPartitioned(self.partition_keys())]
         })
+        .allow_range_satisfaction_for_key_partitioning()
     }
 
     fn with_new_children(