Resolve errors introduced during rebase

apache · Dec 15, 2023 · b0f3633 · b0f3633
1 parent 8c433af
commit b0f3633
Show file tree

Hide file tree

Showing 4 changed files with 12 additions and 13 deletions.
diff --git a/datafusion/core/src/physical_optimizer/join_pipeline_selection.rs b/datafusion/core/src/physical_optimizer/join_pipeline_selection.rs
@@ -121,7 +121,7 @@ impl PlanState {
         self.plans[0]
             .children()
             .into_iter()
-            .map(|child| PlanState::new(child))
+            .map(PlanState::new)
             .collect()
     }
 }
@@ -1786,7 +1786,6 @@ mod order_preserving_join_swap_tests {
                 )
                 .unwrap()],
                 input.clone(),
-                input.schema(),
                 vec![],
                 crate::physical_plan::windows::PartitionSearchMode::Sorted,
             )
@@ -4060,7 +4059,7 @@ mod sql_fuzzy_tests {
         let config = SessionConfig::new()
             .with_target_partitions(1)
             .with_repartition_joins(false);
-        let ctx = SessionContext::with_config(config);
+        let ctx = SessionContext::new_with_config(config);
         let tables = ["orders", "lineitem", "customer", "nation", "region"];
         let can_be_infinite = [true, true, true, true, false];
         let ordered_columns = [

diff --git a/datafusion/core/tests/sql/joins.rs b/datafusion/core/tests/sql/joins.rs
@@ -127,11 +127,11 @@ async fn join_change_in_planner() -> Result<()> {
         [
             "SlidingHashJoinExec: join_type=Full, on=[(a2@1, a2@1)], filter=CAST(a1@0 AS Int64) > CAST(a1@1 AS Int64) + 3 AND CAST(a1@0 AS Int64) < CAST(a1@1 AS Int64) + 10",
             "  CoalesceBatchesExec: target_batch_size=8192",
-            "    SortPreservingRepartitionExec: partitioning=Hash([a2@1], 8), input_partitions=8",
+            "    SortPreservingRepartitionExec: partitioning=Hash([a2@1], 8), input_partitions=8, sort_exprs=a1@0 ASC NULLS LAST",
             "      RepartitionExec: partitioning=RoundRobinBatch(8), input_partitions=1",
             // "     CsvExec: file_groups={1 group: [[tempdir/left.csv]]}, projection=[a1, a2], has_header=false",
             "  CoalesceBatchesExec: target_batch_size=8192",
-            "    SortPreservingRepartitionExec: partitioning=Hash([a2@1], 8), input_partitions=8",
+            "    SortPreservingRepartitionExec: partitioning=Hash([a2@1], 8), input_partitions=8, sort_exprs=a1@0 ASC NULLS LAST",
             "      RepartitionExec: partitioning=RoundRobinBatch(8), input_partitions=1",
             // "     CsvExec: file_groups={1 group: [[tempdir/right.csv]]}, projection=[a1, a2], has_header=false"
         ]

diff --git a/datafusion/physical-plan/src/joins/prunability.rs b/datafusion/physical-plan/src/joins/prunability.rs
@@ -211,7 +211,7 @@ impl ExprPrunability {
         self.expr
             .children()
             .into_iter()
-            .map(|e| ExprPrunability::new(e))
+            .map(ExprPrunability::new)
             .collect()
     }
 

diff --git a/datafusion/sqllogictest/test_files/stream.slt b/datafusion/sqllogictest/test_files/stream.slt
@@ -295,20 +295,20 @@ ProjectionExec: expr=[amount_usd@0 as amount_usd]
 ----ProjectionExec: expr=[LAST_VALUE(e.rate) ORDER BY [e.sn2 ASC NULLS LAST]@1 as amount_usd, sn@0 as sn]
 ------AggregateExec: mode=FinalPartitioned, gby=[sn@0 as sn], aggr=[LAST_VALUE(e.rate)], ordering_mode=FullyOrdered
 --------CoalesceBatchesExec: target_batch_size=8192
-----------SortPreservingRepartitionExec: partitioning=Hash([sn@0], 4), input_partitions=4
+----------SortPreservingRepartitionExec: partitioning=Hash([sn@0], 4), input_partitions=4, sort_exprs=sn@0 ASC NULLS LAST
 ------------AggregateExec: mode=Partial, gby=[sn@0 as sn], aggr=[LAST_VALUE(e.rate)], ordering_mode=FullyOrdered
 --------------ProjectionExec: expr=[sn@1 as sn, sn2@4 as sn2, rate@6 as rate]
 ----------------ProjectionExec: expr=[ts@4 as ts, sn@5 as sn, currency@6 as currency, ts@0 as ts, sn2@1 as sn2, currency_from@2 as currency_from, rate@3 as rate]
 ------------------PartitionedHashJoinExec: join_type=Inner, on=[(currency_from@2, currency@2)], filter=ts@0 >= ts@1
 --------------------CoalesceBatchesExec: target_batch_size=8192
-----------------------SortPreservingRepartitionExec: partitioning=Hash([currency_from@2], 4), input_partitions=4
+----------------------SortPreservingRepartitionExec: partitioning=Hash([currency_from@2], 4), input_partitions=4, sort_exprs=sn2@1 ASC NULLS LAST
 ------------------------ProjectionExec: expr=[ts@0 as ts, sn2@1 as sn2, currency_from@2 as currency_from, rate@4 as rate]
 --------------------------CoalesceBatchesExec: target_batch_size=8192
 ----------------------------FilterExec: currency_to@3 = USD
 ------------------------------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
 --------------------------------CsvExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/exchange_rates.csv]]}, projection=[ts, sn2, currency_from, currency_to, rate], infinite_source=true, output_ordering=[sn2@1 ASC NULLS LAST], has_header=true
 --------------------CoalesceBatchesExec: target_batch_size=8192
-----------------------SortPreservingRepartitionExec: partitioning=Hash([currency@2], 4), input_partitions=4
+----------------------SortPreservingRepartitionExec: partitioning=Hash([currency@2], 4), input_partitions=4, sort_exprs=sn@1 ASC NULLS LAST
 ------------------------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
 --------------------------CsvExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/sales_global.csv]]}, projection=[ts, sn, currency], infinite_source=true, output_ordering=[sn@1 ASC NULLS LAST], has_header=true
 
@@ -388,20 +388,20 @@ ProjectionExec: expr=[amount_usd@0 as amount_usd]
 ----ProjectionExec: expr=[AVG(e.rate)@1 as amount_usd, sn@0 as sn]
 ------AggregateExec: mode=FinalPartitioned, gby=[sn@0 as sn], aggr=[AVG(e.rate)], ordering_mode=FullyOrdered
 --------CoalesceBatchesExec: target_batch_size=8192
-----------SortPreservingRepartitionExec: partitioning=Hash([sn@0], 4), input_partitions=4
+----------SortPreservingRepartitionExec: partitioning=Hash([sn@0], 4), input_partitions=4, sort_exprs=sn@0 ASC NULLS LAST
 ------------AggregateExec: mode=Partial, gby=[sn@0 as sn], aggr=[AVG(e.rate)], ordering_mode=FullyOrdered
 --------------ProjectionExec: expr=[sn@1 as sn, rate@5 as rate]
 ----------------ProjectionExec: expr=[ts@3 as ts, sn@4 as sn, currency@5 as currency, ts@0 as ts, currency_from@1 as currency_from, rate@2 as rate]
 ------------------SlidingHashJoinExec: join_type=Right, on=[(currency_from@1, currency@2)], filter=ts@0 >= ts@1 AND ts@1 >= ts@0 - 600000000000
 --------------------CoalesceBatchesExec: target_batch_size=8192
-----------------------SortPreservingRepartitionExec: partitioning=Hash([currency_from@1], 4), input_partitions=4
+----------------------SortPreservingRepartitionExec: partitioning=Hash([currency_from@1], 4), input_partitions=4, sort_exprs=ts@0 ASC NULLS LAST
 ------------------------ProjectionExec: expr=[ts@0 as ts, currency_from@1 as currency_from, rate@3 as rate]
 --------------------------CoalesceBatchesExec: target_batch_size=8192
 ----------------------------FilterExec: currency_to@2 = USD
 ------------------------------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
 --------------------------------CsvExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/exchange_rates.csv]]}, projection=[ts, currency_from, currency_to, rate], infinite_source=true, output_ordering=[ts@0 ASC NULLS LAST], has_header=true
 --------------------CoalesceBatchesExec: target_batch_size=8192
-----------------------SortPreservingRepartitionExec: partitioning=Hash([currency@2], 4), input_partitions=4
+----------------------SortPreservingRepartitionExec: partitioning=Hash([currency@2], 4), input_partitions=4, sort_exprs=ts@0 ASC NULLS LAST
 ------------------------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
 --------------------------CsvExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/sales_global.csv]]}, projection=[ts, sn, currency], infinite_source=true, output_ordering=[ts@0 ASC NULLS LAST], has_header=true
 
@@ -431,7 +431,7 @@ SortPreservingMergeExec: [ts@0 ASC NULLS LAST]
 --ProjectionExec: expr=[ts@0 as ts, sn@1 as sn, currency_from@2 as currency_from, currency_to@3 as currency_to, rate@4 as rate, AVG(e.rate) PARTITION BY [e.currency_from, e.currency_to] ORDER BY [e.ts ASC NULLS LAST] RANGE BETWEEN 10 MINUTE PRECEDING AND CURRENT ROW@5 as avg_rate]
 ----BoundedWindowAggExec: wdw=[AVG(e.rate) PARTITION BY [e.currency_from, e.currency_to] ORDER BY [e.ts ASC NULLS LAST] RANGE BETWEEN 10 MINUTE PRECEDING AND CURRENT ROW: Ok(Field { name: "AVG(e.rate) PARTITION BY [e.currency_from, e.currency_to] ORDER BY [e.ts ASC NULLS LAST] RANGE BETWEEN 10 MINUTE PRECEDING AND CURRENT ROW", data_type: Float64, nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }), frame: WindowFrame { units: Range, start_bound: Preceding(IntervalMonthDayNano("600000000000")), end_bound: CurrentRow }], mode=[Linear]
 ------CoalesceBatchesExec: target_batch_size=8192
---------SortPreservingRepartitionExec: partitioning=Hash([currency_from@2, currency_to@3], 4), input_partitions=4
+--------SortPreservingRepartitionExec: partitioning=Hash([currency_from@2, currency_to@3], 4), input_partitions=4, sort_exprs=ts@0 ASC NULLS LAST
 ----------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
 ------------CsvExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/exchange_rates.csv]]}, projection=[ts, sn, currency_from, currency_to, rate], infinite_source=true, output_ordering=[ts@0 ASC NULLS LAST], has_header=true