[Epic] Remove Sort Merge Join Experimental status

[comphead]

Is your feature request related to a problem or challenge?

Hi all

I was going through SMJ implementation and suddenly stepped on the comments

// Sort-Merge join support currently is experimental

https://github.com/apache/arrow-datafusion/blob/81c96fc3db0ea35638278f32df066be63b745a51/datafusion/core/src/physical_planner.rs#L1141

I think it would be nice to revisit it and understand if Sort Merge Join Exec is still experimental.
And if so is there any strategies to make it stable, or to run benchmarks to prove the join is stable?

• Adding TPCH benchmarks for Sort Merge Join #10092
• Trigger the Sort Merge Join benchmark using the GitHub action #10109
• fix Sort Merge Join to pass TPCH tests #10100
• Memory account not adding up in SortExec #10073
• Document Sort Merge Join algorithm #10357
• Sort Merge Join. LeftSemi issues #10379
• Sort Merge Join. LeftAnti issues #10380
• Add spilling in SortMergeJoin #9359
• SortMergeJoin: Add RightSemi join support #13471
• SortMergeJoin: Add RightAnti join support #13472
• SortMergeJoin: Add fuzz tests for SMJ when the JoinFilter is set #10659
• SortMergeJoin: The query stuck when join filter is set and more matched rows than batch size #10491
• SMJ LeftAnti filtered join fuzz test cannot pass #10872
• SMJ: incorrect result for filtered right outer join #10882
• SMJ producing different results than HashJoin when doing a semi join #10886
• Add fuzz tests for SortMergeJoin spilling  #11541
• Intermittent failures in fuzz_cases::join_fuzz::test_anti_join_1k_filtered #11555
• Flaky fuzz tests for filtered outer SortMergeJoin #12359
• Test TPCH with sort merge join #13573

Describe the solution you'd like

No response

Describe alternatives you've considered

No response

Additional context

No response

[comphead]

@alamb @ozankabak @viirya @mustafasrepo @berkaysynnada @metesynnada appreciate your inputs.

[alamb]

From my experience, I have never seen SortMergeJoin used in any plan I looked at in DataFusion, so therefore I think it is still "experimental" or at least "not used by datafusion by default" (which maybe is the same thing)

It looks like there was some past interest in SortMergeJoin -- https://github.com/apache/arrow-datafusion/issues?q=is%3Aissue+is%3Aopen+sortmergejoin

Also the people interested in that operator seem to be the people focused on Spark

[comphead]

it is used if next conditions met https://github.com/apache/arrow-datafusion/blob/81c96fc3db0ea35638278f32df066be63b745a51/datafusion/core/src/physical_planner.rs#L1136

There is also a small set of tests introduced in sort_merge_join.slt. And the plans there shows SMJ

To enforce SMJ its needed to set

set datafusion.optimizer.prefer_hash_join = false;

Probably we can revisit tests and run some benchmarks with SMJ enforced to make a decision?

[metesynnada]

I believe we can add fuzz tests for SMJ to ensure it is robust.

[comphead]

I'm thinking if its enough to add fuzz tests, prob we also need to run benchmarks on top of SMJ? Afaik now benchmarks are on top of the HJ?

[metesynnada]

Is there a rule of thumb for choosing SMJ over HJ?

[Dandandan]

Is there a rule of thumb for choosing SMJ over HJ?

I wonder how SMJ in DataFusion compares against HJ at the moment.

Some ideas for when SMJ could be chosen over HJ:

• When input data is already sorted on relevant keys, it is likely faster/requires less memory to plan a SMJ than HJ.
• HJ might require more memory than SMJ, so whenever e.g. data skew is expected one might choose sort merge over hash join.

[alamb]

Is there a rule of thumb for choosing SMJ over HJ?

I believe current state of the art in query processing is

1. If the data is already sorted by join keys, use MergeJoin (as @Dandandan says)
2. If the data is not already sorted on join key, use HashJoin
3. If HashJoin runs out of memory building the hash table, spill the table to disk (possibly switching to merge join internally)

The only benefit SMJ has over HJ at the moment in Datafusion is that we could plausibly join relations that are larger than memory using SMJ (using the fact that we can spill the inputs) -- this may be what @Dandandan is saying in #9846 (comment)

I think it is close to impossible to make SMJ beat HJ for raw performance when the relations fit in memory

[comphead]

we shouldn't be comparing HJ vs SMJ 1:1, but the performance has to be quite close? What I'm trying to solve is to find a strategy to remove the experimental flag from SMJ and prove it is stable.

btw I found the fuzz tests are in place https://github.com/apache/arrow-datafusion/blob/daf182dc789230dbd9cf21ca2e975789213a5365/datafusion/core/tests/fuzz_cases/join_fuzz.rs#L128

[comphead]

I ran TPCH benchmarks for SMJ and got

thread 'tokio-runtime-worker' panicked at datafusion/physical-plan/src/joins/sort_merge_join.rs:1357:22:
index out of bounds: the len is 0 but the index is 1
note: run with `RUST_BACKTRACE=1` environment variable to display a backtrace
thread 'tokio-runtime-worker' panicked at datafusion/physical-plan/src/joins/sort_merge_join.rs:1357:22:
index out of bounds: the len is 0 but the index is 1
thread 'tokio-runtime-worker' panicked at datafusion/physical-plan/src/joins/sort_merge_join.rs:1357:22:
index out of bounds: the len is 0 but the index is 1
thread 'tokio-runtime-worker' panicked at datafusion/physical-plan/src/joins/sort_merge_join.rs:1357:22:
index out of bounds: the len is 0 but the index is 1
Error: Context("Join Error", External(JoinError::Panic(Id(88693), ...)))

[alamb]

Seems like a good reason to keep it marked as experimental

[comphead]

Seems like a good reason to keep it marked as experimental

I'll create a separate issue on it.
Once TPCH passed we can get back on SMJ status