[FEAT] Outer joins for native executor #2860

colin-ho · 2024-09-19T19:03:57Z

Implement outer joins for Swordfish.

(Yes, this PR is a little big. But:

at least tests run in CI now, so you don't need to just take my word for it now.
A lot of the diff is because I moved left/right joins to be together with the outer join operator. Therefore the HashJoinProbe operator is now just InnerHashJoinProbeOperator)

Outer join probes (and left/right now) are implemented as a Streaming Sink.

During the execute phase of the streaming sink, probing is done concurrently via workers (this is the same implementation as all the other join types). The only difference is that during probing, workers will save the indices on the left side that have matches (using a mutable bitmap).
During the finalize phase, we merge together all the bitmaps across the concurrent workers (via a bitwise OR) to get a global view of all the indices that had matches. Then, we take all the indices that didn't get a match and return them (with nulls for the right side). This is the same logic we currently use for the python runner.
Why is left/right with outer joins now? In the future, we may want to choose the build side for left/right/outer joins based on cardinality. This means that we may need the used_indices bitmaps for left/right joins as well.

Note: I had to make Streaming Sink concurrency-aware to allow this. The changes in particular are:

Streaming Sinks can specify max concurrency, currently only LIMIT will have this set to 1.
execute accepts some mut state and finalize will consolidate all of the state, i.e. Vec<Box<dyn State>>.
In order to make sure that all the workers are done, they are spawned on a Worker Set, and return their state when done. This ensures that the finalize method doesn't get called before the workers are done with the executes.

codspeed-hq · 2024-09-19T19:19:11Z

CodSpeed Performance Report

Merging #2860 will not alter performance

_{Comparing colin/swordfish-outer-join (d4e8942) with main (4a8244b)}

Summary

✅ 17 untouched benchmarks

codecov · 2024-09-19T21:47:08Z

Codecov Report

Attention: Patch coverage is 98.68668% with 7 lines in your changes missing coverage. Please review.

Project coverage is 78.58%. Comparing base (4a8244b) to head (d4e8942).
Report is 1 commits behind head on main.

Files with missing lines	Patch %	Lines
...local-execution/src/sinks/outer_hash_join_probe.rs	98.52%	4 Missing ⚠️
...tion/src/intermediate_ops/inner_hash_join_probe.rs	97.16%	3 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #2860      +/-   ##
==========================================
+ Coverage   78.49%   78.58%   +0.09%     
==========================================
  Files         613      614       +1     
  Lines       72761    73064     +303     
==========================================
+ Hits        57111    57418     +307     
+ Misses      15650    15646       -4

Files with missing lines	Coverage Δ
.../src/intermediate_ops/anti_semi_hash_join_probe.rs	`96.61% <100.00%> (+0.24%)`	⬆️
src/daft-local-execution/src/pipeline.rs	`93.56% <100.00%> (+1.22%)`	⬆️
src/daft-local-execution/src/runtime_stats.rs	`55.67% <100.00%> (ø)`
.../daft-local-execution/src/sinks/hash_join_build.rs	`95.00% <100.00%> (-0.24%)`	⬇️
src/daft-local-execution/src/sinks/limit.rs	`100.00% <100.00%> (ø)`
...c/daft-local-execution/src/sinks/streaming_sink.rs	`80.15% <100.00%> (+14.27%)`	⬆️
src/daft-table/src/lib.rs	`90.05% <ø> (ø)`
src/daft-table/src/probeable/mod.rs	`100.00% <100.00%> (ø)`
...tion/src/intermediate_ops/inner_hash_join_probe.rs	`97.16% <97.16%> (ø)`
...local-execution/src/sinks/outer_hash_join_probe.rs	`98.52% <98.52%> (ø)`

... and 5 files with indirect coverage changes

src/arrow2/src/bitmap/mutable.rs

src/daft-local-execution/Cargo.toml

samster25 · 2024-09-25T20:25:20Z

src/daft-local-execution/src/intermediate_ops/inner_hash_join_probe.rs

+        input: &Arc<MicroPartition>,
+        state: &mut InnerHashJoinProbeState,
+    ) -> DaftResult<Arc<MicroPartition>> {
+        let (probe_table, tables) = state.get_probeable_and_table();


This may be worth making a struct type for

struct ProbeState { probe_table tables }

Makes sense! Implemented it

src/daft-local-execution/src/intermediate_ops/inner_hash_join_probe.rs

src/daft-local-execution/src/sinks/outer_hash_join_probe.rs

samster25 · 2024-09-25T20:56:50Z

src/daft-local-execution/src/sinks/outer_hash_join_probe.rs

+        let mut build_side_growable =
+            GrowableTable::new(&tables.iter().collect::<Vec<_>>(), true, 20)?;
+
+        for (table_idx, row_idx) in merged_bitmap.get_unused_indices() {


This can be much more performant using a BitmapIter

Daft/src/arrow2/src/bitmap/utils/iterator.rs

Line 8 in b1ea3b9

pub struct BitmapIter<'a> {

Which will compress the adjacent valid bits so we can reduce the calls to extend.

You can also convert the bitmap into a BooleanArray and use mask_filter https://github.com/Eventual-Inc/Daft/blob/b1ea3b9749e01512f48dfd45f9899a329fc9799f/src/daft-table/src/lib.rs#L321 instead of iterating

I thought the BitmapIter returns an iterator over the individual bits? I just checked and there is also SlicesIterator: https://github.com/Eventual-Inc/Daft/blob/b1ea3b9749e01512f48dfd45f9899a329fc9799f/src/arrow2/src/bitmap/utils/slice_iterator.rs which is a Iterator over a bitmap that returns slices of set regions, did you mean this one?

Whoops, yup thats the one

Actually, I prefer the BooleanArray as a mask_filter method more, it's a lot cleaner. Went with that in the latest commit.

src/daft-local-execution/src/sinks/outer_hash_join_probe.rs

samster25

Very clean now!

samster25 · 2024-10-21T23:08:15Z

src/daft-local-execution/src/intermediate_ops/inner_hash_join_probe.rs

+        }
+    }
+
+    fn get_probe_state(&self) -> Arc<ProbeState> {


prefer &Arc<ProbeState> since the caller may not need to clone.

samster25 · 2024-10-22T00:03:22Z

src/daft-local-execution/src/sinks/limit.rs

+        Self { remaining }
+    }
+
+    fn get_remaining(&self) -> usize {


You can also implement this as get_remaining_mut() -> &mut usize that will give you a mutable reference to remaining. That way you don't have to do it as two different steps.

initial implementation

c1c3182

github-actions bot added the enhancement New feature or request label Sep 19, 2024

Colin Ho added 4 commits September 19, 2024 13:26

clean up lil bit and fix test

3aa8aaf

get right side null schema in .new

86a5fa4

more cleanup

dcac8ba

reduce code

bb0a944

colin-ho marked this pull request as ready for review September 19, 2024 21:16

colin-ho requested a review from samster25 September 19, 2024 21:17

need to drop sender

c797da4

samster25 reviewed Sep 25, 2024

View reviewed changes

Colin Ho and others added 7 commits September 25, 2024 15:41

bitmap improvements + probe state

dbe7723

prefix sum

94245f8

Merge branch main into colin/swordfish-outer-join

9db56c3

cleanup

787f29f

fix deadlock;

a2d417a

check for empty

f93847a

use mask filter

d609336

colin-ho requested a review from samster25 September 26, 2024 17:36

colin-ho mentioned this pull request Sep 30, 2024

[FEAT] Enable concat for swordfish #2976

Merged

Merge branch main into colin/swordfish-outer-join

09c785c

samster25 approved these changes Oct 22, 2024

View reviewed changes

Colin Ho added 2 commits October 21, 2024 17:59

Merge branch main into colin/swordfish-outer-join

bc8e1d2

feedback

d4e8942

colin-ho enabled auto-merge (squash) October 22, 2024 01:20

colin-ho merged commit b9b2d72 into main Oct 22, 2024
40 checks passed

colin-ho deleted the colin/swordfish-outer-join branch October 22, 2024 01:21

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FEAT] Outer joins for native executor #2860

[FEAT] Outer joins for native executor #2860

colin-ho commented Sep 19, 2024 •

edited

Loading

codspeed-hq bot commented Sep 19, 2024 •

edited

Loading

codecov bot commented Sep 19, 2024 •

edited

Loading

samster25 Sep 25, 2024

colin-ho Sep 26, 2024

samster25 Sep 25, 2024

samster25 Sep 25, 2024

colin-ho Sep 26, 2024

samster25 Sep 26, 2024

colin-ho Sep 26, 2024

samster25 left a comment

samster25 Oct 21, 2024

samster25 Oct 22, 2024

[FEAT] Outer joins for native executor #2860

[FEAT] Outer joins for native executor #2860

Conversation

colin-ho commented Sep 19, 2024 • edited Loading

codspeed-hq bot commented Sep 19, 2024 • edited Loading

Merging #2860 will not alter performance

Summary

codecov bot commented Sep 19, 2024 • edited Loading

Codecov Report

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

samster25 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

colin-ho commented Sep 19, 2024 •

edited

Loading

codspeed-hq bot commented Sep 19, 2024 •

edited

Loading

codecov bot commented Sep 19, 2024 •

edited

Loading