Ensure dynamic filter expr is built before fetching probe batch in HashJoin

[rkrishn7]

Describe the bug

Presently, partition bounds are reported after the build side is collected during a hash join. However, there is no synchronization to wait until all bounds have been reported before we begin polling the right side.

This is so we can ensure that upstream APIs like FileOpener::open can immediately observe the updated dynamic filter as they may begin using it right away.

Without some synchronization, the behavior is racy and it's not guaranteed that that the dynamic filter is built prior to initiating the right side's execution plan.

To Reproduce

Reproducing is flaky given that the behavior is not determinate. I've been able to observe the placeholder dynamic filter expr (lit(true)) from time to time by adding a log to TestOpener::open in core::tests::physical_optimizer::filter_pushdown::utils.

Expected behavior

The updated dynamic filter should be immediately visible on the probe side.

Additional context

No response

[rkrishn7]

cc @adriangb

[adriangb]

Without some synchronization, the behavior is racy and it's not guaranteed that that the dynamic filter is built prior to initiating the right side's execution plan.

I think that's fine? For a big query it will be a non issue. It could make a difference in a small query. My worry is that if you get synchronization wrong or misunderstand something you can actually slow things down, end up with deadlocks, etc.

[rkrishn7]

Without some synchronization, the behavior is racy and it's not guaranteed that that the dynamic filter is built prior to initiating the right side's execution plan.

I think that's fine? For a big query it will be a non issue. It could make a difference in a small query. My worry is that if you get synchronization wrong or misunderstand something you can actually slow things down, end up with deadlocks, etc.

Hmm, I think we want the filter to be built prior to any scans being initiated on the right side. The underlying datasource implementation may try to make use of it immediately (e.g. pruning predicate for row group filtering). In my mind, we ideally don't the filter's presence to be a function of how long it takes to get the complete bounds information from all partitions on the left side of the join.

[adriangb]

I took a look at the PR it looks really nice. I think it's what we want. I just have to double check it with some more time. Nice work!