Implement scalable distributed joins about datafusion HOT 3 CLOSED

Here is some additional information. When I run TPC-H query 5 in the benchmarks, against DataFusion, I see that the physical plan used partitioned joins.

For example, I see that both inputs to the join are partitioned on the join keys, and the join mode is Partitioned.

HashJoinExec: mode=Partitioned, join_type=Inner, on=[("c_custkey", "o_custkey")]
  RepartitionExec: partitioning=Hash([Column { name: "c_custkey" }], 24)
    ParquetExec: batch_size=8192, limit=None, partitions=[...]
  RepartitionExec: partitioning=Hash([Column { name: "o_custkey" }], 24)
    FilterExec: o_orderdate >= CAST(1994-01-01 AS Date32) AND o_orderdate < CAST(1995-01-01 AS Date32)
      ParquetExec: batch_size=8192, limit=None, partitions=[...]

This means that the join can run in parallel because the inputs are partitioned. So partition 1 of the join reads partition 1 of the left and right inputs, and so on.

When I run the same query against Ballista, I see.

HashJoinExec: mode=CollectLeft, join_type=Inner, on=[("c_custkey", "o_custkey")]
  ParquetExec: batch_size=8192, limit=None, partitions=[...]
  FilterExec: o_orderdate >= CAST(1994-01-01 AS Date32) AND o_orderdate < CAST(1995-01-01 AS Date32)
    ParquetExec: batch_size=8192, limit=None, partitions=[

Here, we see join mode CollectLeft, which means that each partition being executed will go and fetch the entire left-side of the join into memory. This is very inefficient both in terms of memory and compute and potentially gets exponentially slower the more partitions we have.

What we need to do is apply the same "partitioned hash join" pattern to Ballista.

from datafusion.

I created a Google doc to discuss the design, and planned work, in more detail.

https://docs.google.com/document/d/1yUnGWsHKYOAxWijDJisEFYU4dIym_GSRSMpwfWjVZq8/edit?usp=sharing

from datafusion.

I'd love to work on this if someone can provide further reading material and/or the area in the code

from datafusion.