refactor: modified `JoinHashMap` build order for `HashJoinStream` #8658

korowa · 2023-12-26T16:37:09Z

Which issue does this PR close?

Closes #8130.

Rationale for this change

The least harmful way of allowing HashJoinStream to process probe-side batch records in original order -- is to adjust build-side input to produce FIFO hash map as a result of update_hash, using current data structure without any modifications, and this can be achieved by reverse iteration over build-side input.

What changes are included in this PR?

collect_left_input iterates over collected batches in reverse order
update_hash and build_equal_condition_join_indices now accept fifo_hashmap argument, which determines the iteration order for input batches (SymmetricsHashJoin execution has not changed)
default implemenations for update_from_iter & get_matched_indices in JoinHashMapType trait -- to unload join helper functions from HashMap-specific code

Are these changes tested?

assert_batches_sorted_eq replaced with assert_batches_eq for Inner / RightSemi / RightAnti non-partitioned join test cases to validate that join output preserves input(s) order.

Are there any user-facing changes?

No.

metesynnada · 2023-12-28T06:54:33Z

Hi @korowa, this is a great extension. I suggest adding explanations in the code base, including docstrings for the mechanism and updating the docstring of the hash join. You can use https://textik.com, a great tool for drawing diagrams in docstring.

I have another question regarding keeping the "head" inside the JoinHashMap next array. I've noticed that with each new row addition, the previous row indices need to be scanned, making the addition complexity linear instead of constant. Have you run performance benchmarks on low cardinality key columns to see how this affects performance?

korowa · 2023-12-28T19:23:16Z

@metesynnada thank you!

Regarding explanations -- I've added extended explanation of how build phase works in HashJoinExec docstring -- hope with this description of "the whole picture", the other modified / added comments will make more sense.

Regarding performance -- benchmark results are

--------------------
Benchmark tpch_mem.json
--------------------
┏━━━━━━━━━━━━━━┳━━━━━━━━━━┳━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━┓
┃ Query        ┃   master ┃ hash_join_build_order ┃        Change ┃
┡━━━━━━━━━━━━━━╇━━━━━━━━━━╇━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━┩
│ QQuery 1     │ 156.09ms │              155.86ms │     no change │
│ QQuery 2     │  36.49ms │               36.30ms │     no change │
│ QQuery 3     │  70.99ms │               71.59ms │     no change │
│ QQuery 4     │  61.01ms │               58.94ms │     no change │
│ QQuery 5     │ 104.64ms │              101.76ms │     no change │
│ QQuery 6     │  16.71ms │               16.67ms │     no change │
│ QQuery 7     │ 274.75ms │              283.66ms │     no change │
│ QQuery 8     │  65.54ms │               65.67ms │     no change │
│ QQuery 9     │ 103.82ms │              104.10ms │     no change │
│ QQuery 10    │ 123.52ms │              123.01ms │     no change │
│ QQuery 11    │  28.87ms │               27.55ms │     no change │
│ QQuery 12    │  50.39ms │               52.08ms │     no change │
│ QQuery 13    │  59.11ms │               58.02ms │     no change │
│ QQuery 14    │  21.04ms │               18.94ms │ +1.11x faster │
│ QQuery 15    │  47.09ms │               46.57ms │     no change │
│ QQuery 16    │  32.32ms │               33.87ms │     no change │
│ QQuery 17    │ 150.53ms │              140.37ms │ +1.07x faster │
│ QQuery 18    │ 324.83ms │              322.20ms │     no change │
│ QQuery 19    │  46.00ms │               45.74ms │     no change │
│ QQuery 20    │  83.35ms │               83.45ms │     no change │
│ QQuery 21    │ 261.97ms │              265.00ms │     no change │
│ QQuery 22    │  28.01ms │               25.16ms │ +1.11x faster │
└──────────────┴──────────┴───────────────────────┴───────────────┘
┏━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━┓
┃ Benchmark Summary                    ┃           ┃
┡━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━┩
│ Total Time (master)                  │ 2147.08ms │
│ Total Time (hash_join_build_order)   │ 2136.50ms │
│ Average Time (master)                │   97.59ms │
│ Average Time (hash_join_build_order) │   97.11ms │
│ Queries Faster                       │         3 │
│ Queries Slower                       │         0 │
│ Queries with No Change               │        19 │
└──────────────────────────────────────┴───────────┘

(faster queries probably are no more than outliers)

I've noticed that with each new row addition, the previous row indices need to be scanned, making the addition complexity linear instead of constant.

Could you, please, recheck or point the LOC where scanning takes place? If I got it right -- the problem should be in JoinHashMap::update_from_iter, but the worst case for insertion there seems to be

obtain currently stored index index in hashmap for hash value
place it at specific position in next vector
update hashmap with a new index

which should not produce linear complexity 🤔 (at least it definitely shouldn't perform worse than before)

alamb

Thank you @korowa -- this is a really nice refactor. I read it carefully and it looks very nice to me.

cc @Dandandan and @liukun4515 for your interest

alamb · 2023-12-28T20:18:15Z

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

@@ -156,8 +155,48 @@ impl JoinLeftData {
 ///
 /// Execution proceeds in 2 stages:
 ///
-/// 1. the **build phase** where a hash table is created from the tuples of the
-/// build side.
+/// 1. the **build phase** creates a hash table from the tuples of the build side,


thank you for this

alamb · 2023-12-28T20:20:33Z

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

@@ -1558,7 +1573,9 @@ mod tests {
            "| 3  | 5  | 9  | 20 | 5  | 80 |",
            "+----+----+----+----+----+----+",
        ];
-        assert_batches_sorted_eq!(expected, &batches);
+
+        // Inner join output is expected to preserve both inputs order


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

alamb · 2023-12-28T20:21:20Z

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

            "| 3  | 5  | 9  | 20 | 5  | 80 |",
+            "| 2  | 5  | 8  | 20 | 5  | 80 |",


this also is neat to see it keep the same order as the input

alamb · 2023-12-28T20:23:20Z

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

@@ -151,6 +151,82 @@ pub trait JoinHashMapType {
    fn get_map(&self) -> &RawTable<(u64, u64)>;
    /// Returns a reference to the next.
    fn get_list(&self) -> &Self::NextType;
+


JoinHashMap and this trait are getting big enough maybe to warrant moving to its own module (in a separate PR, perhaps)

Maybe partial join output will add some more logic to JoinHashMap and make separate module reasonable (or even required) at that stage

metesynnada

I guess I misunderstood an explanation in the related issue. This shouldn't degrade the performance, nicely done. However, are the JoinHashMap struct comments relevant after this change?

korowa · 2023-12-29T10:31:42Z

However, are the JoinHashMap struct comments relevant after this change?

@metesynnada, yes, the PR is only related to the JoinHashMap input -- all its internals remains as they were before so all docs and comments for both JoinHashMap and moved pieces of code are still relevant. The only comment modification is related to missing offset argument in update_from_iter as its caller responsibility to provide correct row indices in iterator.

korowa · 2023-12-29T12:02:31Z

Failed tests look like related to chrono-tz 0.8.5 + eggert/tz@1d0ae80 🤔

korowa · 2023-12-29T12:22:45Z

Pushed separate PR cause new chrono-tz release will affect test run on any branch.

alamb · 2023-12-30T14:08:57Z

Thank you @korowa -- both for this PR and for debugging #8677

maintaining fifo hashmap in hash join

32c62dd

korowa changed the title ~~feat: modified JoinHashMap build order for HashJoinStream~~ refactor: modified JoinHashMap build order for HashJoinStream Dec 26, 2023

ozankabak requested a review from metesynnada December 26, 2023 17:04

alamb mentioned this pull request Dec 27, 2023

DataFusion weekly project plan (Andrew Lamb) - Dec 25, 2023 #8655

Closed

7 tasks

extended HashJoinExec docstring on build phase

c424eb3

alamb approved these changes Dec 28, 2023

View reviewed changes

metesynnada approved these changes Dec 29, 2023

View reviewed changes

korowa added 2 commits December 29, 2023 12:08

testcases for randomly ordered build side input

7311bd9

Merge remote-tracking branch 'upstream/main' into hash_join_build_order

95fa3b5

trigger ci

00b8997

Merge remote-tracking branch 'upstream/main' into hash_join_build_order

fd82b35

alamb merged commit 00a679a into apache:main Dec 30, 2023
22 checks passed

alamb mentioned this pull request Dec 30, 2023

join statement causes panic #7931

Closed

matthewgapp mentioned this pull request Jan 11, 2024

matt/feat/recursive ctes/config flag matthewgapp/arrow-datafusion#3

Closed

This was referenced Mar 10, 2024

Further refine the Top K sort operator #9417

Open

Generate GroupByHash output in multiple RecordBatches rather than one large one #9562

Open

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

refactor: modified `JoinHashMap` build order for `HashJoinStream` #8658

refactor: modified `JoinHashMap` build order for `HashJoinStream` #8658

korowa commented Dec 26, 2023

metesynnada commented Dec 28, 2023 •

edited

Loading

korowa commented Dec 28, 2023 •

edited

Loading

alamb left a comment •

edited

Loading

alamb Dec 28, 2023

alamb Dec 28, 2023

alamb Dec 28, 2023

alamb Dec 28, 2023

korowa Dec 28, 2023 •

edited

Loading

metesynnada left a comment

korowa commented Dec 29, 2023

korowa commented Dec 29, 2023

korowa commented Dec 29, 2023

alamb commented Dec 30, 2023

		"\| 3 \| 5 \| 9 \| 20 \| 5 \| 80 \|",
		"\| 2 \| 5 \| 8 \| 20 \| 5 \| 80 \|",

refactor: modified JoinHashMap build order for HashJoinStream #8658

refactor: modified JoinHashMap build order for HashJoinStream #8658

Conversation

korowa commented Dec 26, 2023

Which issue does this PR close?

Rationale for this change

What changes are included in this PR?

Are these changes tested?

Are there any user-facing changes?

metesynnada commented Dec 28, 2023 • edited Loading

korowa commented Dec 28, 2023 • edited Loading

alamb left a comment • edited Loading

Choose a reason for hiding this comment

alamb Dec 28, 2023

Choose a reason for hiding this comment

alamb Dec 28, 2023

Choose a reason for hiding this comment

alamb Dec 28, 2023

Choose a reason for hiding this comment

alamb Dec 28, 2023

Choose a reason for hiding this comment

korowa Dec 28, 2023 • edited Loading

Choose a reason for hiding this comment

metesynnada left a comment

Choose a reason for hiding this comment

korowa commented Dec 29, 2023

korowa commented Dec 29, 2023

korowa commented Dec 29, 2023

alamb commented Dec 30, 2023

refactor: modified `JoinHashMap` build order for `HashJoinStream` #8658

refactor: modified `JoinHashMap` build order for `HashJoinStream` #8658

metesynnada commented Dec 28, 2023 •

edited

Loading

korowa commented Dec 28, 2023 •

edited

Loading

alamb left a comment •

edited

Loading

korowa Dec 28, 2023 •

edited

Loading