Investigate use of `DynamicFilters` in ballista

[milenkovicm]

Is your feature request related to a problem or challenge? Please describe what you are trying to do.

There is great push in DataFusion community to support DynamicFilters, looks like there are some good query improvements when dynamic filters are used.

I'm not sure if they would improve ballista execution due to stage execution but at least we could investigate if they will yield any benefits for ballista

Describe the solution you'd like

It would be great if we could some effort to investigate if shuffle read would have any benefits of it.

• try to integrate dynamic filters with shuffle read to start with
• benchmark benefits (if any)
• provide road forward

Describe alternatives you've considered

Additional context

[milenkovicm]

@Huy1Ng would you be interested in having a look at this ?

[Huy1Ng]

might need this feature: apache/datafusion#18296

[Huy1Ng]

At the current stage, dynamic filter only work in a single task . Let' say you have cluster with 1 scheduler and 1 executor.
You make this query "select * from line_items sort by line_id asc". The schedule plans 2 task, and send to the sole executor. Each task will be able to leverage the new threshold it found in TopK to reduce number of rows scan upstream, but these tasks cannot share this information to each other.

My approach is to solve this local problem with by having tasks on a single executor share this dynamic filter with each others. Then we can proceed to look for ways to sync this information across the network boundary.

[milenkovicm]

I'm not sure we should share them between executors, it would get just too complex.

perhaps if we focus:

• could we use them per partition ?
• could we push them to shuffle reader to remove unneeded data from shuffle ?
• could we push them across to flight server to remove unneeded data before transport (in case of non block transport)

we can drop this task if it does not make any sense and focus on something else, please do not spend too much time its just investigation

[Dandandan]

One case where it works well is in broadcasted hash join (collect left mode), as it collects the build side to each executor, it can also push down the filter / hash table to the scan.

[milenkovicm]

thanks @Dandandan, help with the direction would be appreciated