Preserve input's equivalence properties in UnnestExec

Author: vegarsti

datafusion/physical-expr/src/equivalence/ordering.rs

@@ -74,6 +74,11 @@ impl OrderingEquivalenceClass {
         result
     }
 
+    /// Returns the orderings in this ordering equivalence class.
+    pub fn orderings(&self) -> &[LexOrdering] {
+        &self.orderings
+    }
+
     /// Extend this ordering equivalence class with the given orderings.
     pub fn extend(&mut self, orderings: impl IntoIterator<Item = LexOrdering>) {
         self.orderings.extend(orderings);

datafusion/physical-plan/src/unnest.rs

@@ -51,8 +51,8 @@ use futures::{Stream, StreamExt};
 use log::trace;
 
 /// Unnest the given columns (either with type struct or list)
-/// For list unnesting, each rows is vertically transformed into multiple rows
-/// For struct unnesting, each columns is horizontally transformed into multiple columns,
+/// For list unnesting, each row is vertically transformed into multiple rows
+/// For struct unnesting, each column is horizontally transformed into multiple columns,
 /// Thus the original RecordBatch with dimension (n x m) may have new dimension (n' x m')
 ///
 /// See [`UnnestOptions`] for more details and an example.
@@ -101,8 +101,22 @@ impl UnnestExec {
         input: &Arc<dyn ExecutionPlan>,
         schema: SchemaRef,
     ) -> PlanProperties {
+        // Extract equivalence properties from input plan
+        let input_eq_properties = input.equivalence_properties();
+        let orderings = input_eq_properties.oeq_class().orderings().to_vec();
+        let eq_group = input_eq_properties.eq_group();
+        let constraints = input_eq_properties.constraints();
+
+        // Create new equivalence properties for the unnest plan based on the input plan
+        let mut eq_properties =
+            EquivalenceProperties::new_with_orderings(schema, orderings)
+                .with_constraints(constraints.to_owned());
+        eq_properties
+            .add_equivalence_group(eq_group.to_owned())
+            .unwrap(); // We can unwrap this because we know this is a valid equivalence group
+
         PlanProperties::new(
-            EquivalenceProperties::new(schema),
+            eq_properties,
             input.output_partitioning().to_owned(),
             input.pipeline_behavior(),
             input.boundedness(),

datafusion/sqllogictest/test_files/unnest.slt

@@ -941,3 +941,84 @@ where min_height * width1 = (
 )
 ----
 4 7 4 28
+
+## Unnest with ordering on unrelated column is preserved
+query TT
+EXPLAIN WITH unnested AS (SELECT
+    ROW_NUMBER() OVER () AS generated_id,
+    unnest(array[value]) as ar
+  FROM range(1,5)) SELECT array_agg(ar) FROM unnested group by generated_id;
+----
+logical_plan
+01)Projection: array_agg(unnested.ar)
+02)--Aggregate: groupBy=[[unnested.generated_id]], aggr=[[array_agg(unnested.ar)]]
+03)----SubqueryAlias: unnested
+04)------Projection: generated_id, __unnest_placeholder(make_array(range().value),depth=1) AS UNNEST(make_array(range().value)) AS ar
+05)--------Unnest: lists[__unnest_placeholder(make_array(range().value))|depth=1] structs[]
+06)----------Projection: row_number() ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING AS generated_id, make_array(range().value) AS __unnest_placeholder(make_array(range().value))
+07)------------WindowAggr: windowExpr=[[row_number() ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING]]
+08)--------------TableScan: range() projection=[value]
+physical_plan
+01)ProjectionExec: expr=[array_agg(unnested.ar)@1 as array_agg(unnested.ar)]
+02)--AggregateExec: mode=FinalPartitioned, gby=[generated_id@0 as generated_id], aggr=[array_agg(unnested.ar)], ordering_mode=Sorted
+03)----SortExec: expr=[generated_id@0 ASC NULLS LAST], preserve_partitioning=[true]
+04)------CoalesceBatchesExec: target_batch_size=8192
+05)--------RepartitionExec: partitioning=Hash([generated_id@0], 4), input_partitions=4
+06)----------AggregateExec: mode=Partial, gby=[generated_id@0 as generated_id], aggr=[array_agg(unnested.ar)], ordering_mode=Sorted
+07)------------ProjectionExec: expr=[generated_id@0 as generated_id, __unnest_placeholder(make_array(range().value),depth=1)@1 as ar]
+08)--------------UnnestExec
+09)----------------ProjectionExec: expr=[row_number() ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING@1 as generated_id, make_array(value@0) as __unnest_placeholder(make_array(range().value))]
+10)------------------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
+11)--------------------BoundedWindowAggExec: wdw=[row_number() ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING: Field { name: "row_number() ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING", data_type: UInt64, nullable: false, dict_id: 0, dict_is_ordered: false, metadata: {} }, frame: ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING], mode=[Sorted]
+12)----------------------LazyMemoryExec: partitions=1, batch_generators=[range: start=1, end=5, batch_size=8192]
+
+## Unnest without ordering
+query TT
+EXPLAIN WITH unnested AS (SELECT
+    random() AS generated_id,
+    unnest(array[value]) as ar
+  FROM range(1,5)) SELECT array_agg(ar) FROM unnested group by generated_id;
+----
+logical_plan
+01)Projection: array_agg(unnested.ar)
+02)--Aggregate: groupBy=[[unnested.generated_id]], aggr=[[array_agg(unnested.ar)]]
+03)----SubqueryAlias: unnested
+04)------Projection: generated_id, __unnest_placeholder(make_array(range().value),depth=1) AS UNNEST(make_array(range().value)) AS ar
+05)--------Unnest: lists[__unnest_placeholder(make_array(range().value))|depth=1] structs[]
+06)----------Projection: random() AS generated_id, make_array(range().value) AS __unnest_placeholder(make_array(range().value))
+07)------------TableScan: range() projection=[value]
+physical_plan
+01)ProjectionExec: expr=[array_agg(unnested.ar)@1 as array_agg(unnested.ar)]
+02)--AggregateExec: mode=FinalPartitioned, gby=[generated_id@0 as generated_id], aggr=[array_agg(unnested.ar)]
+03)----CoalesceBatchesExec: target_batch_size=8192
+04)------RepartitionExec: partitioning=Hash([generated_id@0], 4), input_partitions=4
+05)--------AggregateExec: mode=Partial, gby=[generated_id@0 as generated_id], aggr=[array_agg(unnested.ar)]
+06)----------ProjectionExec: expr=[generated_id@0 as generated_id, __unnest_placeholder(make_array(range().value),depth=1)@1 as ar]
+07)------------UnnestExec
+08)--------------ProjectionExec: expr=[random() as generated_id, make_array(value@0) as __unnest_placeholder(make_array(range().value))]
+09)----------------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
+10)------------------LazyMemoryExec: partitions=1, batch_generators=[range: start=1, end=5, batch_size=8192]
+
+## ???
+query TT
+explain with temp as (select random() as r, [value, value + 1] as ordered from range(1, 5)), temp2 as (select r, unnest(ordered) as x from temp) select r, sum(x) from temp2 group by r;
+----
+logical_plan
+01)Aggregate: groupBy=[[temp2.r]], aggr=[[sum(temp2.x)]]
+02)--SubqueryAlias: temp2
+03)----Projection: temp.r, __unnest_placeholder(temp.ordered,depth=1) AS UNNEST(temp.ordered) AS x
+04)------Unnest: lists[__unnest_placeholder(temp.ordered)|depth=1] structs[]
+05)--------Projection: temp.r, temp.ordered AS __unnest_placeholder(temp.ordered)
+06)----------SubqueryAlias: temp
+07)------------Projection: random() AS r, make_array(range().value, range().value + Int64(1)) AS ordered
+08)--------------TableScan: range() projection=[value]
+physical_plan
+01)AggregateExec: mode=FinalPartitioned, gby=[r@0 as r], aggr=[sum(temp2.x)]
+02)--CoalesceBatchesExec: target_batch_size=8192
+03)----RepartitionExec: partitioning=Hash([r@0], 4), input_partitions=4
+04)------AggregateExec: mode=Partial, gby=[r@0 as r], aggr=[sum(temp2.x)]
+05)--------ProjectionExec: expr=[r@0 as r, __unnest_placeholder(temp.ordered,depth=1)@1 as x]
+06)----------UnnestExec
+07)------------ProjectionExec: expr=[random() as r, make_array(value@0, value@0 + 1) as __unnest_placeholder(temp.ordered)]
+08)--------------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
+09)----------------LazyMemoryExec: partitions=1, batch_generators=[range: start=1, end=5, batch_size=8192]