gene-bordegaray commented on code in PR #23623:
URL: https://github.com/apache/datafusion/pull/23623#discussion_r3597211379
##########
datafusion/physical-plan/src/union.rs:
##########
@@ -1288,6 +1295,165 @@ mod tests {
);
}
+ fn make_hash_exec(
+ schema: &SchemaRef,
+ hash_cols: Vec<&str>,
+ buckets: usize,
+ ) -> Result<Arc<dyn ExecutionPlan>> {
+ let exprs = hash_cols
+ .iter()
+ .map(|c| col(c, schema))
+ .collect::<Result<Vec<_>>>()?;
+ let base = Arc::new(TestMemoryExec::try_new(&[], Arc::clone(schema),
None)?);
+ Ok(Arc::new(RepartitionExec::try_new(
+ base,
+ Partitioning::Hash(exprs, buckets),
+ )?))
+ }
+
+ fn make_range_exec(
+ schema: &SchemaRef,
+ split_values: Vec<i32>,
+ sort_options: SortOptions,
+ ) -> Result<Arc<dyn ExecutionPlan>> {
+ let sort_expr =
+ PhysicalSortExpr::new(col(schema.field(0).name(), schema)?,
sort_options);
+ let ordering = LexOrdering::new(vec![sort_expr]).unwrap();
+ let split_points = split_values
+ .into_iter()
+ .map(|v| SplitPoint::new(vec![ScalarValue::Int32(Some(v))]))
+ .collect();
+ let base = Arc::new(TestMemoryExec::try_new(&[], Arc::clone(schema),
None)?);
+ Ok(Arc::new(RepartitionExec::try_new(
+ base,
+ Partitioning::Range(RangePartitioning::try_new(ordering,
split_points)?),
+ )?))
+ }
+
+ #[test]
+ fn test_can_interleave_matrix() -> Result<()> {
+ let name_column = "name";
+ let age_column = "age";
+ let schema = Arc::new(Schema::new(vec![
+ Field::new(name_column, DataType::Int32, true),
+ Field::new(age_column, DataType::Int32, true),
+ ]));
+
+ let ascending = SortOptions {
+ descending: false,
+ nulls_first: false,
+ };
+ let descending = SortOptions {
+ descending: true,
+ nulls_first: false,
+ };
+ let ascending_nulls_first = SortOptions {
+ descending: false,
+ nulls_first: true,
+ };
+
+ struct Case {
+ inputs: Vec<Arc<dyn ExecutionPlan>>,
+ expected: bool,
+ label: &'static str,
+ }
+
+ let cases = vec![
Review Comment:
I would say lets just keep:
- matchin hash multi col
- matching range
- subset hash
- range with diff sort options
- mixed hash / range
sorry I think this got a little too large 😅
##########
datafusion/sqllogictest/test_files/range_partitioning.slt:
##########
@@ -1203,3 +1203,140 @@ reset datafusion.explain.physical_plan_only;
statement ok
reset datafusion.optimizer.enable_window_topn;
+
+##########
+# TEST 34: 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.
+##########
+
+statement ok
+set datafusion.explain.physical_plan_only = true;
+
+query TT
+EXPLAIN SELECT range_key, value FROM range_partitioned
+UNION ALL
+SELECT range_key, value FROM range_partitioned
+UNION ALL
+SELECT range_key, value FROM range_partitioned_shifted;
+----
+physical_plan
+01)UnionExec
+02)--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
Review Comment:
🥳
##########
datafusion/physical-plan/src/union.rs:
##########
@@ -682,8 +685,9 @@ impl ExecutionPlan for InterleaveExec {
}
}
-/// If all the input partitions have the same Hash partition spec with the
first_input_partition
-/// The InterleaveExec is partition aware.
+/// Returns true if all inputs have the same [`Partitioning::Hash`] or
[`Partitioning::Range`]
+/// spec, making them safe to interleave. Two inputs are interleave-compatible
when partition
+/// `k` covers the identical key range or hash bucket across every input.
Review Comment:
can we add a note how we intentionally don't have to use
`InputDistributionRequirements::co_partitioned` because can_interleave checks
each child's compatibility sequentially
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
