TPC-DS query 72 slow on modest (10) scale factors

[AdamGS]

When running TPC-DS q72, I've noticed that regardless of the underlying file format, latency increases dramatically even with relatively modest scale factors like 10. I've measured the query at around 2.4 seconds with SF=1, but over 60s when SF=10.

When running in my benchmarking setup, the plan is (as you can see - its extremely join heavy) here.

Profiling the query using samply (This is with branch-50 over parquet, SF=1):

By playing around with it, seems like most time is spent in the loop inside the chain_traverse macro. I've tried a few common performance techniques - making it an explicitly inlined generic function, changing how the indices and values memory is managed/written to, but nothing made a noticeable difference.

[adriangb]

Hmm if I’m understanding correctly most of the time is spent doing the join, in particular doing the hash map lookup?

[adriangb]

I wonder if this would be a good candidate query to test pushing down bloom filters

[adriangb]

Ref #17171 #16435

[xudong963]

I'm wondering if the projection pushdown works as expected here, #16799 (comment)

Have you compared the explain plan with the previous version?

[shehabgamin]

The changes in DataFusion 50 are unrelated to the slowness in TPC-DS query 72. If you change the order of join operations, the query will no longer be slow (see here).

From my understanding, a physical optimizer is needed for reordering joins in order to improve performance.

[xudong963]

From my understanding, a physical optimizer is needed for reordering joins in order to improve performance.

Yes, the current reorder algorithm is trivial, even without ndv.

I checked the part of the code, now it considers the ndv

[Dandandan]

chain_traverse traverses the chain for all hashes matching on the build side matching the hash on the probe side. If the build side is larger and every probe value matches many build side hashes, this algorithm is known to be not very (cache) efficient.

One thing to consider is implementing a different algorithm for the hash join.

[Dandandan]

But in this case, I agree the biggest improvement probably comes from a global join ordering algorithm rather than tweaks to the hash join algorithm.

[milenkovicm]

I had a look at this query yesterday (same 10 scale). Spark 4.0 takes 15 seconds, compared to datafusion which takes 3 mins 40 seconds.

Comparing two plans there are two notable differences, spark uses BroadcastHashJoin for all joins apart from most inner (first join it takes), for which it uses SortMergeJoin. If forced to use ShuffledHashJoin instead of SortMergeJoin time overall time increases from 15 seconds to like 1 min 30 seconds, which is still faster than DataFusion.

Lastly, if BroadcastHashJoin is disabled, performance will decrease further to around 2 minutes 55 seconds , which still faster than DataFusion.

DataFusion does repartitioning after each join, and I haven't really tried to force CollectLeft.

Overall join ordering does look same between spark and datafusion apart from selected physical joins.

Ballista is slightly faster than DataFusion with 3 minutes 20 seconds due to join re-ordering within stages, unfortunately CollectLeft is broken in ballista as @Dandandan pointed out in apache/datafusion-ballista#1055 and never PRed 😃

Hope it helps

[metegenez]

But in this case, I agree the biggest improvement probably comes from a global join ordering algorithm rather than tweaks to the hash join algorithm.

What we have should be sufficient if the statistical estimations are accurate (which is quite hard to do). The JoinSelection physical optimizer already performs join reordering, but I suspect the statistics aren’t precise enough to guide reordering effectively at higher join levels.

[Dandandan]

When running TPC-DS q72, I've noticed that regardless of the underlying file format, latency increases dramatically even with relatively modest scale factors like 10. I've measured the query at around 2.4 seconds with SF=1, but over 60s when SF=10.

When running in my benchmarking setup, the plan is (as you can see - its extremely join heavy) here.

Profiling the query using samply (This is with branch-50 over parquet, SF=1):

By playing around with it, seems like most time is spent in the loop inside the chain_traverse macro. I've tried a few common performance techniques - making it an explicitly inlined generic function, changing how the indices and values memory is managed/written to, but nothing made a noticeable difference.

One thing that seems off to me in this flamegraph is the amount spent in Vec::push. The capacity of the Vec is set to the rows limit, so push should be pretty inexpensive?
Or is the profile showing misleading info?

[AdamGS]

What I remember from my investigation at the time is that it seems like a lot of this is just a hot loop inside chain_traverse. Pushing into the vec is slower than just writing to it and setting the length once at the end without checking the capacity everytime.
I can also try and re-run the profile, I assume some of the code around it changed over the past few months so maybe things looks differently now.

[Dandandan]

I wonder what the plan looks like for query 72.
Maybe the join condition is relatively unselective, e.g. a lot of tuples match, which makes this loop very hot (i.e. we have to output a lot of rows).

[Dandandan]

Ah I see it mostly was reported here already lakehq/sail#759 (comment)

I guess that one of the right joins with a table scan on the build side is causing this slowness.
Still, we could optimize the join operator for cases when the build side is larger or contains lots of duplicates.