Remove CoalescePartitions insertion from Joins by ctsk · Pull Request #15570 · apache/datafusion

ctsk · 2025-04-03T19:19:38Z

Rationale for this change + What changes are included in this PR?

The logic of HashJoin's execute implements differs from the other operators - The recursive call to the execute step of the build side are delayed until collect_left_join is poll'ed. This PR changes that to make it more alike to the standard implementations of execute.

Are these changes tested?

No changes in behaviour are expected.

Are there any user-facing changes?

No

Notable changes:

Fixed the bug that caused test failures (partition 0 of the left side was executed multiple times)
Applied the same changes the nested_loop_join and cross_join

This backs out commit 4346cb8.

berkaysynnada

The changes are LGTM, thank you @ctsk for tracking and finalizing this issue

datafusion/physical-plan/src/joins/cross_join.rs

alamb · 2025-04-06T13:17:59Z

I wonder if there is any way to write some tests for this (perhaps via EXPLAIN in .slt tests to demonstrate that the unecessary exec is removed)

alamb · 2025-04-06T13:20:19Z

🤖 ./gh_compare_branch.sh Benchmark Script Running
Linux aal-dev 6.8.0-1016-gcp #18-Ubuntu SMP Fri Oct 4 22:16:29 UTC 2024 x86_64 x86_64 x86_64 GNU/Linux
Comparing remove-hj-coalesce (995a95c) to 0b061be diff
Benchmarks: tpch_mem clickbench_partitioned clickbench_extended
Results will be posted here when complete

berkaysynnada · 2025-04-06T13:23:09Z

I wonder if there is any way to write some tests for this (perhaps via EXPLAIN in .slt tests to demonstrate that the unecessary exec is removed)

I don't think a redundant exec can be removed since if the input is 1 partition, it was already not introducing a new one, and what we see is input is always 1 (enforce distribution does its job 👏🏻), because we don't get any plan changes

alamb · 2025-04-06T13:40:37Z

I see -- this is code simplification 👍

alamb · 2025-04-06T14:03:06Z

🤖: Benchmark completed

Details

Comparing HEAD and remove-hj-coalesce
--------------------
Benchmark clickbench_extended.json
--------------------
┏━━━━━━━━━━━━━━┳━━━━━━━━━━━━┳━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━┓
┃ Query        ┃       HEAD ┃ remove-hj-coalesce ┃    Change ┃
┡━━━━━━━━━━━━━━╇━━━━━━━━━━━━╇━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━┩
│ QQuery 0     │  1983.64ms │          1974.40ms │ no change │
│ QQuery 1     │   699.38ms │           708.13ms │ no change │
│ QQuery 2     │  1421.28ms │          1456.67ms │ no change │
│ QQuery 3     │   708.08ms │           717.37ms │ no change │
│ QQuery 4     │  1510.24ms │          1496.34ms │ no change │
│ QQuery 5     │ 17349.43ms │         17050.26ms │ no change │
│ QQuery 6     │  6706.21ms │          6713.50ms │ no change │
└──────────────┴────────────┴────────────────────┴───────────┘
┏━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━┓
┃ Benchmark Summary                 ┃            ┃
┡━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━┩
│ Total Time (HEAD)                 │ 30378.27ms │
│ Total Time (remove-hj-coalesce)   │ 30116.68ms │
│ Average Time (HEAD)               │  4339.75ms │
│ Average Time (remove-hj-coalesce) │  4302.38ms │
│ Queries Faster                    │          0 │
│ Queries Slower                    │          0 │
│ Queries with No Change            │          7 │
└───────────────────────────────────┴────────────┘
--------------------
Benchmark clickbench_partitioned.json
--------------------
┏━━━━━━━━━━━━━━┳━━━━━━━━━━━━┳━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━┓
┃ Query        ┃       HEAD ┃ remove-hj-coalesce ┃        Change ┃
┡━━━━━━━━━━━━━━╇━━━━━━━━━━━━╇━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━┩
│ QQuery 0     │     1.99ms │             1.92ms │     no change │
│ QQuery 1     │    35.20ms │            34.74ms │     no change │
│ QQuery 2     │    91.74ms │            91.34ms │     no change │
│ QQuery 3     │   102.29ms │           101.18ms │     no change │
│ QQuery 4     │   744.08ms │           740.73ms │     no change │
│ QQuery 5     │   877.28ms │           865.60ms │     no change │
│ QQuery 6     │     1.91ms │             1.94ms │     no change │
│ QQuery 7     │    40.83ms │            39.88ms │     no change │
│ QQuery 8     │   936.92ms │           953.98ms │     no change │
│ QQuery 9     │  1256.02ms │          1281.07ms │     no change │
│ QQuery 10    │   275.47ms │           273.32ms │     no change │
│ QQuery 11    │   301.56ms │           305.70ms │     no change │
│ QQuery 12    │   937.22ms │           942.74ms │     no change │
│ QQuery 13    │  1273.43ms │          1275.80ms │     no change │
│ QQuery 14    │   881.06ms │           881.44ms │     no change │
│ QQuery 15    │  1070.13ms │          1068.40ms │     no change │
│ QQuery 16    │  1803.13ms │          1794.14ms │     no change │
│ QQuery 17    │  1638.06ms │          1651.44ms │     no change │
│ QQuery 18    │  3170.64ms │          3178.40ms │     no change │
│ QQuery 19    │    85.90ms │            87.13ms │     no change │
│ QQuery 20    │  1144.96ms │          1141.99ms │     no change │
│ QQuery 21    │  1380.02ms │          1353.25ms │     no change │
│ QQuery 22    │  2504.46ms │          2484.34ms │     no change │
│ QQuery 23    │  8618.09ms │          8678.11ms │     no change │
│ QQuery 24    │   483.76ms │           477.05ms │     no change │
│ QQuery 25    │   393.25ms │           397.38ms │     no change │
│ QQuery 26    │   550.26ms │           543.41ms │     no change │
│ QQuery 27    │  1738.81ms │          1705.99ms │     no change │
│ QQuery 28    │ 12715.67ms │         12666.45ms │     no change │
│ QQuery 29    │   530.60ms │           550.06ms │     no change │
│ QQuery 30    │   868.66ms │           848.77ms │     no change │
│ QQuery 31    │   913.96ms │           899.26ms │     no change │
│ QQuery 32    │  2762.98ms │          2813.21ms │     no change │
│ QQuery 33    │  3447.68ms │          3417.70ms │     no change │
│ QQuery 34    │  3430.75ms │          3448.18ms │     no change │
│ QQuery 35    │  1331.45ms │          1316.45ms │     no change │
│ QQuery 36    │   159.48ms │           154.87ms │     no change │
│ QQuery 37    │    88.14ms │            85.47ms │     no change │
│ QQuery 38    │   149.12ms │           153.57ms │     no change │
│ QQuery 39    │   227.91ms │           228.93ms │     no change │
│ QQuery 40    │    77.19ms │            79.93ms │     no change │
│ QQuery 41    │    76.35ms │            72.23ms │ +1.06x faster │
│ QQuery 42    │    65.96ms │            67.20ms │     no change │
└──────────────┴────────────┴────────────────────┴───────────────┘
┏━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━┓
┃ Benchmark Summary                 ┃            ┃
┡━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━┩
│ Total Time (HEAD)                 │ 59184.35ms │
│ Total Time (remove-hj-coalesce)   │ 59154.69ms │
│ Average Time (HEAD)               │  1376.38ms │
│ Average Time (remove-hj-coalesce) │  1375.69ms │
│ Queries Faster                    │          1 │
│ Queries Slower                    │          0 │
│ Queries with No Change            │         42 │
└───────────────────────────────────┴────────────┘
--------------------
Benchmark tpch_mem_sf1.json
--------------------
┏━━━━━━━━━━━━━━┳━━━━━━━━━━┳━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━┓
┃ Query        ┃     HEAD ┃ remove-hj-coalesce ┃        Change ┃
┡━━━━━━━━━━━━━━╇━━━━━━━━━━╇━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━┩
│ QQuery 1     │ 121.75ms │           123.41ms │     no change │
│ QQuery 2     │  24.15ms │            24.22ms │     no change │
│ QQuery 3     │  35.77ms │            36.30ms │     no change │
│ QQuery 4     │  21.25ms │            21.42ms │     no change │
│ QQuery 5     │  57.69ms │            58.52ms │     no change │
│ QQuery 6     │   8.21ms │             8.27ms │     no change │
│ QQuery 7     │ 110.68ms │           104.15ms │ +1.06x faster │
│ QQuery 8     │  26.96ms │            26.31ms │     no change │
│ QQuery 9     │  63.25ms │            63.22ms │     no change │
│ QQuery 10    │  61.03ms │            61.83ms │     no change │
│ QQuery 11    │  13.05ms │            13.21ms │     no change │
│ QQuery 12    │  37.51ms │            39.51ms │  1.05x slower │
│ QQuery 13    │  29.16ms │            29.59ms │     no change │
│ QQuery 14    │   9.79ms │            10.05ms │     no change │
│ QQuery 15    │  25.66ms │            25.54ms │     no change │
│ QQuery 16    │  24.00ms │            24.33ms │     no change │
│ QQuery 17    │  98.05ms │            96.97ms │     no change │
│ QQuery 18    │ 249.89ms │           249.44ms │     no change │
│ QQuery 19    │  28.19ms │            28.38ms │     no change │
│ QQuery 20    │  39.71ms │            40.64ms │     no change │
│ QQuery 21    │ 176.95ms │           169.94ms │     no change │
│ QQuery 22    │  17.44ms │            18.33ms │  1.05x slower │
└──────────────┴──────────┴────────────────────┴───────────────┘
┏━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━┓
┃ Benchmark Summary                 ┃           ┃
┡━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━┩
│ Total Time (HEAD)                 │ 1280.14ms │
│ Total Time (remove-hj-coalesce)   │ 1273.58ms │
│ Average Time (HEAD)               │   58.19ms │
│ Average Time (remove-hj-coalesce) │   57.89ms │
│ Queries Faster                    │         1 │
│ Queries Slower                    │         2 │
│ Queries with No Change            │        19 │
└───────────────────────────────────┴───────────┘

alamb · 2025-04-06T19:00:01Z

TLDR is benchmark results look good to me -- thanks @ctsk !

alamb

Thank you @ctsk and @berkaysynnada

alamb · 2025-04-07T13:09:25Z

datafusion/physical-plan/src/joins/cross_join.rs

        partition: usize,
        context: Arc<TaskContext>,
    ) -> Result<SendableRecordBatchStream> {
+        if self.left.output_partitioning().partition_count() != 1 {


* refactor: Catch PartitionMode:Auto in execute explicitly * refactor(hash_join): Move coalesce logic into function * refactor(hash_join): Move coalesce logic out of collect_left_input * refactor(hash_join): Execute build side earlier * chore(hash_join): Drop unnecessary clippy hint \o/ * Back out "refactor: Catch PartitionMode:Auto in execute explicitly" This backs out commit 4346cb8. * Remove CoalescePartitions from CollectLeft-HashJoins * Fix imports * Fix tests * Check CollectLeft-HJ distribution in execute * Fix repeated execute; propagate execute error * Remove CoalescePartitions from nested_loop_join * Execute left side earlier in nested_loop_join * Remove CoalescePartitions from cross_join * Execute left side earlier in cross_join * Remove unused import * Fix docs * Remove dead comment * Throw error on unexpected distribution

ctsk added 17 commits April 3, 2025 21:14

refactor: Catch PartitionMode:Auto in execute explicitly

60155c0

refactor(hash_join): Move coalesce logic into function

630301c

refactor(hash_join): Move coalesce logic out of collect_left_input

6a75748

refactor(hash_join): Execute build side earlier

0545793

chore(hash_join): Drop unnecessary clippy hint \o/

0922997

Back out "refactor: Catch PartitionMode:Auto in execute explicitly"

84d5f94

This backs out commit 4346cb8.

Remove CoalescePartitions from CollectLeft-HashJoins

197f8c9

Fix imports

c29700d

Fix tests

460196b

Check CollectLeft-HJ distribution in execute

d5c8314

Fix repeated execute; propagate execute error

2b26ed2

Remove CoalescePartitions from nested_loop_join

3b8e80c

Execute left side earlier in nested_loop_join

35eb148

Remove CoalescePartitions from cross_join

ac1dac2

Execute left side earlier in cross_join

b2bd6bf

Remove unused import

5717f44

Fix docs

995a95c

berkaysynnada approved these changes Apr 6, 2025

View reviewed changes

datafusion/physical-plan/src/joins/cross_join.rs Show resolved Hide resolved

ctsk added 2 commits April 7, 2025 12:46

Remove dead comment

9f21f69

Throw error on unexpected distribution

f661b04

alamb approved these changes Apr 7, 2025

View reviewed changes

alamb merged commit 3fe300c into apache:main Apr 7, 2025
27 checks passed

erratic-pattern mentioned this pull request Dec 26, 2025

Patched DF 51.0.0 (revision a) influxdata/arrow-datafusion#87

Open

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Remove CoalescePartitions insertion from Joins#15570

Remove CoalescePartitions insertion from Joins#15570
alamb merged 19 commits intoapache:mainfrom
ctsk:remove-hj-coalesce

ctsk commented Apr 3, 2025

Uh oh!

berkaysynnada left a comment

Uh oh!

Uh oh!

alamb commented Apr 6, 2025

Uh oh!

alamb commented Apr 6, 2025

Uh oh!

berkaysynnada commented Apr 6, 2025 •

edited

Loading

Uh oh!

alamb commented Apr 6, 2025

Uh oh!

alamb commented Apr 6, 2025

Uh oh!

alamb commented Apr 6, 2025

Uh oh!

alamb left a comment

Uh oh!

alamb Apr 7, 2025

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

Conversation

ctsk commented Apr 3, 2025

Rationale for this change + What changes are included in this PR?

Are these changes tested?

Are there any user-facing changes?

Uh oh!

berkaysynnada left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

alamb commented Apr 6, 2025

Uh oh!

alamb commented Apr 6, 2025

Uh oh!

berkaysynnada commented Apr 6, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

alamb commented Apr 6, 2025

Uh oh!

alamb commented Apr 6, 2025

Uh oh!

alamb commented Apr 6, 2025

Uh oh!

alamb left a comment

Choose a reason for hiding this comment

Uh oh!

alamb Apr 7, 2025

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

berkaysynnada commented Apr 6, 2025 •

edited

Loading