Should Ballista use sort-merge join rather than hash join by default?

[andygrove]

Is your feature request related to a problem or challenge? Please describe what you are trying to do.
DataFusion's hash join implementation does not have spilling yet, so can easily OOM. This is even more likely in Ballista because each executor is processing multiple tasks in parallel.

For example, I cannot get the integration tests to pass (./dev/integration-test.sh) due to OOM, unless I disable hash joins.

Describe the solution you'd like
Personally, I think it would be better for Ballista to default to sort-merge join, and let the user override that if they want hash join. This would be the opposite of DataFusion's config, which enables hash join by default.

Describe alternatives you've considered
It would be nice if DataFusion had a spilling hash join implementation. I am working on one in Comet in apache/datafusion-comet#4184

Additional context

[andygrove]

@milenkovicm WDYT?

[milenkovicm]

I agree, it's sensible default until we get AQE to make that decision

[milenkovicm]

Do you know which rules spark follows to switch between sort and hash join? It we can capture it we can add it as a rule in AQE?

[sandugood]

In Spark's AQE implementation there is a DynamicJoinSelection, that is defined in: here

It is all about operating on the LogicalPlan and creating hints.

The logic (forgive my Scala skills, I don't use it quite often) is that:

• apply is being called and it checks if a strategy was pre-defined by user (i.e broadcast() used on rhs of a join)
• then there are two bools, deciding whether to demote BroadcastHashJoin or not: manyEmptyInPlan, manyEmptyInOther and canBroadcastPlan
• then it is all about branching and creating hints:

if (demoteBroadcastHash && preferShuffleHash) {
  Some(SHUFFLE_HASH)
} else if (demoteBroadcastHash) {
  Some(NO_BROADCAST_HASH)
} else if (preferShuffleHash) {
  Some(PREFER_SHUFFLE_HASH)
} else {
  None
}

After hints are injected, it's on JoinSelection to select the proper physical plan: here (there is a good object's docstring and apply() method docstring explaining the Spark's strategy)

Also, Spark has config parameters, that serve:

1. as a flag to transform join to broadcast one, if one size is smaller than the threshold : autoBroadcastJoinThreshold
2. preferSortMergeJoin - self-explanatory

[milenkovicm]

A prototype is worth thousands of design documents 😀 if you want to experiment please do @sandugood

I think it will simplify things, but we would need custom query planner