Improvement: Distribute Join Across Stages and Tasks

[NGA-TRAN]

@gabotechs made great improvements in PR #145, and our query plan is much better now. Looking at Q3’s plan, there’s still room to optimize.

Currently, both joins run in the same stage and task because we retain the build-side hash in memory. To scale further, we should distribute each join into its own stage and across multiple tasks. This requires shuffling both sides of the join on their join key—unless they’re already partitioned appropriately, in which case we can skip repartitioning but still keep them distributed. See issue #119 for a related example.

[jonathanc-n]

Currently it is not being distributed due to it being CollectLeft hash join execution (CoalescePartitions -> Hash Join).

This could probably use a NetworkBroadcastExec node (new type of network boundary). Insert a RoundRobin partition for the larger side, and then send copies of the small table to each 'large table partition' (broadcast join).

We would need to identify when the large side is already partitioned (in this case, we would just send copies of the small table to the workers which the 'large table partition's already reside in to avoid the network cost; no boundary + can use a different node? ex. BroadcastExec) vs. large table is not partitioned thus we use the NetworkBroadcastExec node idea.

I do think that a CollectLeft hash join shouldn't just default to a broadcast join. A different size threshold should be put in place for when to use broadcast joins. Here is the two cases:

1. Single node datafusion gives a CollectLeft hash join but the smaller table is above a size threshold we have for broadcast joins. in this case we remove the CoalescePartitionsExec that was for the CollectLeft execution and replace it with a repartitioning on the join key for both tables. Proceed with a regular partitioned shuffle hash join.
2. If the smaller table is under the size threshold for broadcast joins we can just do a broadcast join.

[NGA-TRAN]

Thanks @jonathanc-n for providing ideas.

Currently, we do not have cost model and join order enumeration yet, so we do not want to make anything complicated. I would avoid RoundRobin partitioning and Broadcast at this time.

When someone works on this, I suggest we only do 2 options:

1. If the 2 tables are small/medium size, we will do in one stage as we do currently
2. If 2 tables are large, we will partition them on join key.

The large is relative here and just a configuration number

When we work on apache/datafusion#17718, we will consider more options.
There are a lot of papers out there but this Vertica Query Optimizer has many things for us to consider. The plan for this is to build a framework with non-opinionated join enumeration, join ranking and cost model to provide flexibility to users (such as Datadog, InlfuxDB, ...) to choose their own way. We will talk more about these in that ticket

[jonathanc-n]

If both tables are large I believe hash join would automatically insert a repartitionexec and it would turn into a partitioned join. I will try to validate this if I have the time.