Repository navigation
High CPU during dynamic filter bound computation: min_batch/max_batch #17486
Description
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
Activity
LiaCastaneda commentedon Sep 9, 2025
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 commentedon Sep 9, 2025
I strongly suspect the comparison of lists and other structured scalars could be improved significantly.
LiaCastaneda commentedon Sep 9, 2025
Disabling
optimizer.enable_dynamic_filter_pushdownmakes 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 commentedon Sep 9, 2025
It certainly makes sense to me to avoid doing work for DynamicFilters that are never used
adriangb commentedon Sep 9, 2025
We could add some notion of "subscribers" to a dynamic filter?
xudong963 commentedon Sep 10, 2025
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 commentedon Sep 10, 2025
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 commentedon Sep 10, 2025
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 commentedon Sep 10, 2025
@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 commentedon Sep 10, 2025
Will give it a try
LiaCastaneda commentedon Sep 11, 2025
I can confirm it mitigates the CPU issue :)
adriangb commentedon Sep 11, 2025
Amazing! Shall we close this then?
LiaCastaneda commentedon Sep 11, 2025
yes, thanks for the fix! (even if it wasn’t meant to be one 😄 )
alamb commentedon Sep 11, 2025
I suspect this is because the old code always created a new ScalarValue for the min/max, even if the min/max didn't change. For ListArrays specifically this is quite expensive. The new code only makes a new Scalarvalue when the min/max changes
xudong963 commentedon Sep 11, 2025
Do you think we need to file an issue for this?
LiaCastaneda commentedon Sep 11, 2025
There are a couple of other things I noticed while using the API that I think are worth mentioning, but I can file them as separate issues. I’m also happy to take a stab at any of those if no one else is interested!
adriangb commentedon Sep 11, 2025
Please do! Happy to also do a quick call to discuss.