Optimize the join operators

[MrPowers]

DataFusion is underperforming the Polars streaming engine on some localhost join queries (1e8 rows of data on a Macbook M3 with 16GB of RAM):

Here are the join queries.

I am guessing the join operator can be optimized, similar to how the filtering and aggregation operations were optimized.

Here is an example of how the median function was made faster: #13550

See this epic for more info: #13548

[alamb]

FYI @jonathanc-n @UBarney @comphead

[jonathanc-n]

I'm having trouble runnigng flamegraphs on these h2o queries, not sure how to get these csv files to work with them. Anybody know how?

[comphead]

I'm having trouble runnigng flamegraphs on these h2o queries, not sure how to get these csv files to work with them. Anybody know how?

if you run them locally you can try Samply them like
https://github.com/apache/datafusion/blob/main/docs/source/library-user-guide/profiling.md#profiling-using-samply-cross-platform-profiler

[alamb]

I'm having trouble runnigng flamegraphs on these h2o queries, not sure how to get these csv files to work with them. Anybody know how?

I think you can generate the data using

./benchmarks/bench.sh data h2o_small_join

And also for medium and large

[jonathanc-n]

Ohh, I missed the part in the documentation which had the h2o_small_join able to downloaded, i thought the h2o_small had it all 😆

[Dandandan]

Besides profiling, I would like to suggest to research how the other engines are running the join and extract some high level learnings out of it:

• is it using a hash table or open addressing (df doesn't have the latter)
• are we using the correct/optimal join order?
• is the hash-repartition taking considerable time (RepartitionExec, which involves 2 copies of the input) - what are the other engines doing?

[alamb]

is it using a hash table or open addressing (df doesn't have the latter)

@XiangpengHao has mentioned several times that we think DuckDB uses radix trees (which work like hash tables but save the expensive hash).

He has an implementation here
https://github.com/XiangpengHao/congee

Maybe we could investigate using that structure rather than a hash table

is the hash-repartition taking considerable time (RepartitionExec, which involves 2 copies of the input) - what are the other engines doing?

Not only is it copying the values twice it hashes them twice which can be quite expensive

[Dandandan]

is it using a hash table or open addressing (df doesn't have the latter)

@XiangpengHao has mentioned several times that we think DuckDB uses radix trees (which work like hash tables but save the expensive hash).

He has an implementation here https://github.com/XiangpengHao/congee

Maybe we could investigate using that structure rather than a hash table

is the hash-repartition taking considerable time (RepartitionExec, which involves 2 copies of the input) - what are the other engines doing?

Not only is it copying the values twice it _hash_es them twice which can be quite expensive

Ah nice, wasn't aware. That seems a nice direction!

[jonathanc-n]

Another thing we can do is hash it once and use parts of the hash at a time during RepartitionExec and building the hashtable. This is made even better with if another repartition has to be done for spilling. If the optimizer repartitions inputs for a hash join we can refactor RepartitionExec to pass the hash values to the HashJoinExec. I'm know that Velox does something like this.

This would be a bit memory intensive as most of the time because it would need to compute a 64 bit hash (instead of possibly just using a u32 hash) up front but it gets to mask off bits to avoid re-hashing.

[Dandandan]

Also, referencing the direct indexing / perfect hash join here.
I think that should be relatively simple to implement.
duckdb/duckdb#1959
#816

[nuno-faria]

I have also been looking at join performance and I think the main limitation is the order, followed by the lack of join parameterization.
In TPC-H, 6 queries use a bad join order, although we can manually rewrite the SQL to force the correct one. Also, 3 TPC-H queries could benefit from join parameterization. As an extreme (and nonsensical) example, this query will take too long to complete with Datafusion due to the large joins, while only taking 200ms with DuckDB.

SELECT *
FROM lineitem, orders
WHERE l_orderkey = o_orderkey
    AND o_orderkey = 1
    AND l_quantity < (
        SELECT avg(l_quantity)
        FROM lineitem
        WHERE l_orderkey = o_orderkey
    );

If we emulate the parameterization here, Datafusion ends up more than 2x faster than DuckDB (altough manually parameterization is not feasible in most queries).

Here is the breakdown of the queries in TPC-H:

[Dandandan]

Thanks @nuno-faria that's a great insight (for TPC-H / very nested joins we probably should implement a smarter join order algorithm).

For h2o joins however, it seems it is only doing a single join (https://github.com/apache/datafusion/blob/main/benchmarks/queries/h2o/join.sql) so I would expect the order to be correct there?

[zhuqi-lucas]

DataFusion is underperforming the Polars streaming engine on some localhost join queries (1e8 rows of data on a Macbook M3 with 16GB of RAM):

Here are the [join queries](https://github.com/apache/datafusion/blob/main/benchmarks/queries/h2o/join.sql).

I am guessing the join operator can be optimized, similar to how the filtering and aggregation operations were optimized.

Here is an example of how the median function was made faster: #13550

See this epic for more info: #13548

Does this compare result based parquet or csv format? Our h2o benchmark tool currently is using csv format.

And i am curious why the result is for 4 sql result, but there are 5 sqls total? Is it from 1 to 4, or 2 to 5?

[mrpowers-wb]

@zhuqi-lucas - these benchmarks use Parquet files, see the querybench repo for the code: https://github.com/MrPowers/querybench. I think Parquet is a lot better for these benchmarks.

The data generation scripts are in falsa if you'd like to generate the files locally: https://github.com/mrpowers-io/falsa/ (thanks @SemyonSinchenko!)

DuckDB isn't included because it can't handle the joins on my machine with the 1e8 datasets. I guess it runs out of memory. It can handle the 1e7 datasets fine.

There are 5 h2o join queries and q5 is omitted (the join between two large tables) because no engine can handle joining the 1e8 table with another 1e8 table on my machine with 16GB of RAM.

[zhuqi-lucas]

@zhuqi-lucas - these benchmarks use Parquet files, see the querybench repo for the code: https://github.com/MrPowers/querybench. I think Parquet is a lot better for these benchmarks.

The data generation scripts are in falsa if you'd like to generate the files locally: https://github.com/mrpowers-io/falsa/ (thanks @SemyonSinchenko!)

DuckDB isn't included because it can't handle the joins on my machine with the 1e8 datasets. I guess it runs out of memory. It can handle the 1e7 datasets fine.

There are 5 h2o join queries and q5 is omitted (the join between two large tables) because no engine can handle joining the 1e8 table with another 1e8 table on my machine with 16GB of RAM.

Thank you @mrpowers-wb for good explain, i will submit a PR for datafusion h2o benchmark to support parquet format first, so we can optimize based the tool for this compare.