High CPU during dynamic filter bound computation: min_batch/max_batch

[LiaCastaneda]

Describe the bug

Hello! We are very interested in using the Dynamic Filtering on hash joins optim feature, so I brought the dynamic filtering work DataDog#43 to test it on a production environment.

The API works as expected (we do receive dynamic filters). However, since bounds are computed for all queries using Hash Join, the change put all our resources at maximum CPU and profiling showed that almost 2/3 of the overall CPU usage was spent in min_batch and max_batch when comparing lists. Specifically the profiler showed the issue was in min_batch -> min_max_batch_generic -> ScalarValue::partial_cmp for lists (via partial_cmp_list).

It looks like comparing list values during bounds computation is expensive, is this expected?

To Reproduce

No response

Expected behavior

No response

Additional context

No response

[LiaCastaneda]

cc @adriangb -- sorry for the direct ping. Since you did most of the dynamic filtering work, do you know if this behavior is expected? 🙇‍♀️

[alamb]

It looks like comparing list values during bounds computation is expensive, is this expected?

I strongly suspect the comparison of lists and other structured scalars could be improved significantly.

[LiaCastaneda]

Disabling optimizer.enable_dynamic_filter_pushdown makes the issue go away, since the bounds are computed for every query containing hash join if the option is enabled. I wonder if it would make sense to compute dynamic filters only if/when there is a node that consumes/ asks for theDynamicFilterPhysicalExpr ? maybe while polling the right side?

[alamb]

I wonder if it would make sense to compute dynamic filters only if/when there is a node that consumes/ asks for the DynamicFilterPhysicalExpr ? maybe while polling the right side?

It certainly makes sense to me to avoid doing work for DynamicFilters that are never used

[adriangb]

I wonder if it would make sense to compute dynamic filters only if/when there is a node that consumes/ asks for the DynamicFilterPhysicalExpr ? maybe while polling the right side?

It certainly makes sense to me to avoid doing work for DynamicFilters that are never used

We could add some notion of "subscribers" to a dynamic filter?

[xudong963]

I'm worried about the scenario: if the physical optimizer doesn't make a reasonable join order, which means, the build side could be huge, the if we do computation for dynamic filter, things will be bad.

[alamb]

I think the key will be to minimize the overhead that computing the dynamic filters add to the join

I agree finding some way to avoid making a bad join order worse is important

[LiaCastaneda]

I agree finding some way to avoid making a bad join order worse is important

We don't pass the DataFusion logical optimizer to our logical plans, choosing the best join ordering is something we already account for when building the substrait plan, but I think it would be nice to have as an optim rule maybe

[adriangb]

@LiaCastaneda could you try this on main now? I wonder if #17444 made a difference. I'd be surprised if this is slow now: it's just tracking a min and max value for each partition (so doing a binary compare to another scalar for each row on the build side) vs. the already existing work of building a hash table.

[LiaCastaneda]

Will give it a try

[LiaCastaneda]

@LiaCastaneda could you try this on main now? I wonder if #17444 made a difference.

I can confirm it mitigates the CPU issue :)

[adriangb]

Amazing! Shall we close this then?

[LiaCastaneda]

yes, thanks for the fix! (even if it wasn’t meant to be one 😄 )