support dynamic filtering on partitioned data from file source

@adriangb

Is your feature request related to a problem or challenge?

When preserve_file_partitions is enabled, we currently disable dynamic filtering to avoid incorrect assumptions about hash partitioning. This avoids a bug but removes critical filtering. We want to retain dynamic filtering without breaking file‑partitioning guarantees.

Describe the solution you'd like

Enable dynamic filtering for file‑partitioned scans by pruning file groups based on partition values when the dynamic filter keys are a subset (which also implies exact match) of the partition columns.

This should allow dynamic filtering to work safely without requiring repartition or changing join planning.

Describe alternatives you've considered

Introduce a new partitioning implementation via a generic trait‑based scheme (hash/value/range/custom), where users (and datafusion) can implement any type of partitioning scheme they desire.

This would provide the interface needed to determine how partitioning schemes are compatible, what satisfies what, etc. The exact details are not fleshed out but this would be a powerful addition and clear up ambiguities in DataFusion's partitioning modes today.

Additional context

cc: @adriangb @NGA-TRAN @fmonjalet @gabotechs @LiaCastaneda

pruning file groups based on partition values when the dynamic filter keys are a subset (which also implies exact match) of the partition columns.

Sounds good.
This means when dynamic filter keys are superset of the partition key, we do not have enough information to do pruning

There is a lot of good discussion on #20901

@gene-bordegaray

@gene-bordegaray drew a really nice diagram and explained the problem they were having

It seems as if the core issue is that the underlying data is range partitioned, but since there is no way to represent range partitioning in DataFusion, @gene-bordegaray builds a plan that has output partitioning = Hash(join_key) even though the underlying data is not actually hash partitioned

However, there are some assumptions in the codebase (in this case dynamic filter creation) that rely on the data being actually hash partitioned

There is also some discussion here #19580 on not having the CASE statement inside the DynamicFilterPhysicalExpr but having it just a router in the HashJoinExec

@LiaCastaneda

As @LiaCastaneda says, one thing we talked about was this case

Apparently today, A plan that looks like

Join (A.key = B.key)
  BuildInput (A)
  ProbeInput (B)

It seems like the same dynamic filter is pushed today to all partitions of B looks like

CASE
  HASH(b.key)) % num_partitions
  WHEN 0 THEN <filter from A partition 0 hash table>
  WHEN 1 THEN <filter from A partition 1 hash table> 
...
END

This predicate can be pushed down to B regardless of how B is partitioned.

However if we know that both A and B are partitioned in the same way by the join key (in the sense that any tuple that could possibly match is in the same partition on both A and B) then a separate dynamic filter could be used for each partition

For example the filter for partition 0 of B, would be:

<filter from A partition 0 hash>

And the the filter for partition 1 of B, would be:

<filter from A partition 1 hash>

If DataFusion doesn't already make these per-partition optimized filters, I think it would be a good improvement for all plans (and it would effectively solve @gene-bordegaray 's problem too)

Disclaimer: this is based on what was said in the sync call I haven't looked at the code myself

Keep in mind there are two different partitioning: the hash table partitioning is explicitly different (different hash seed) from the repartition / stream level partitioning.

@gene-bordegaray

If DataFusion doesn't already make these per-partition optimized filters, I think it would be a good improvement for all plans (and it would effectively solve @gene-bordegaray 's problem too)

I imagine this would still imply having a way to route dynamic filters from partition 0 from the build to partition 0 from the probe right? even if it's not in the Dynamic Filter itself. I'm not sure if we have a way to know how to match these two (like having access to partition number) in this part of the code.

@gene-bordegaray

@gene-bordegaray asked me in the meeting why a CASE + hash was decided to route dynamic filters, @adriangb you might remember better than me, but I think it was to avoid keeping a hash table with partition info + the whole build side hash table (which could cause memory issues). There were some discussions here and here

Keep in mind there are two different partitioning: the hash table partitioning is explicitly different (different hash seed) from the repartition / stream level partitioning.

Yes and we highlighted this. The conclusion that I and think others came to is that a join sees that each side declares Hash partitioning on the same keys then DF should be able to just route the filter of partition x on the build side to partition x on the probe side.

If a user wants to declare hash partitioning on their data, it is their responsibility to ensure they uphold the promises they make to get correct results.

@gene-bordegaray asked me in the meeting why a CASE + hash was decided to route dynamic filters, @adriangb you might remember better than me, but I think it was to avoid keeping a hash table with partition info + the whole build side hash table (which could cause memory issues). There were some discussions here and here

I guess the question depends on what the alternative is. As far as I can remember / think of the two alternatives would be to try every partition's hash table (poor build side performance and questionable probe side performance) or to make a combined hash table with all values (poor memory use).

I imagine this would still imply having a way to route dynamic filters from partition 0 from the build to partition 0 from the probe right? even if it's not in the Dynamic Filter itself. I'm not sure if we have a way to know how to match these two (like having access to partition number)

What I was thinking about was that at planning time we could figure out if the partitions matched 1x1 and if so just create a specific dynamic filter for each partition

I guess the question depends on what the alternative is. As far as I can remember / think of the two alternatives would be to try every partition's hash table (poor build side performance and questionable probe side performance) or to make a combined hash table with all values (poor memory use).

I think in certain circumstances you only have to try a single partition's hash table

For example:

The join is an eqi-join (equality predicate)
The inputs have exactly the same hash (or range) partitioning on the columns used as join keys

In that case there is exactly one partition's hash table where matching tuples would be (if they are there at all)

Something else to consider:

For performance reasons it might be good to move away from modulo repartition in RepartitionExec using https://github.com/lemire/fastrange or alternatively change the implementation to https://docs.rs/strength_reduce/latest/strength_reduce/

See e.g.
#21830 (comment)
for good results on changing it to use fastrange

Navigation Menu

Repository navigation

support dynamic filtering on partitioned data from file source #20195

Is your feature request related to a problem or challenge?

Describe the solution you'd like

Describe alternatives you've considered

Additional context

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

support dynamic filtering on partitioned data from file source #20195

Description

Is your feature request related to a problem or challenge?

Describe the solution you'd like

Describe alternatives you've considered

Additional context

Activity

NGA-TRAN commented on Feb 6, 2026

alamb commented on Mar 31, 2026

alamb commented on Mar 31, 2026

LiaCastaneda commented on Mar 31, 2026

alamb commented on Mar 31, 2026

adriangb commented on Mar 31, 2026

LiaCastaneda commented on Mar 31, 2026

LiaCastaneda commented on Mar 31, 2026

gene-bordegaray commented on Mar 31, 2026

adriangb commented on Mar 31, 2026

alamb commented on Mar 31, 2026

alamb commented on Mar 31, 2026

Dandandan commented on Apr 24, 2026

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions