Push down entire hash table from HashJoinExec into scans

In #16445 we discussed two approaches:

Push down min/max of the hash table. This is what DuckDB does and what we ended up implementing in Add dynamic filter (bounds) pushdown to HashJoinExec #16445
It's also possible for us to push down the entire hash table to prune rows during scans. This is what this issue tracks.

Some questions to answer:

Should we always push down both the hash table and bounds?
Should we push down the entire hash table as is or should we build bloom filters, an IN LIST expression or other?

@Dandandan

cc @Dandandan

Hello! This is quite interesting.

Should we always push down both the hash table and bounds?

Should we push down the entire hash table as is or should we build bloom filters, an IN LIST expression or other?

My initial thought is that the number of distinct values (NDV) on the build side could serve as a useful signal for which filter to push down. For example:

If the join column has high cardinality, the values are likely fairly random, in which case pushing down an IN LIST may be more effective. I think existing pruning machinery already handles IN LIST quite well — e.g. enabling Parquet row group pruning via Bloom filters.
On the other hand, if the column shows low variance, then min/max statistics may provide better pruning.

Curious to hear your thoughts on this approach!

iirc, df stats doesn't include NDV now

@xudong963

@xudong963 I was thinking we can simply derive the proportion of unique values on the left side, given that we collect the left side in memory.

If the left side has a high number of distinct values for a given column, it may be more likely that the right side follows this as well (UUIDs, for example). In which case pushing down the literal values would likely make pruning more effective.

But let me know if I'm missing something!

@adriangb

I've been playing around with pushing down literal guarantees from the values on the left hand side. Some initial thoughts/observations:

As @adriangb pointed out the question is really when to push down the left column values, since there are likely diminishing returns to doing so. For example, we probably wouldn't want to push down a list of 1000 values, since:

The effectiveness of pruning groups of data decreases as the number of distinct values increases, I imagine.
There is likely a point where the cost of evaluating the bloom filter outweighs the cost of unnecessary data scanning.

I think the above two are directly correlated. That is, they both are influenced by the size of the literal guarantees we build.

However, with point lookup queries, for example, the effectiveness over min/max stats pruning can clearly be observed. Take the following query for example:

SELECT *
FROM customer
JOIN orders on c_custkey = o_custkey
WHERE c_name = 'Customer#000077000';

I observed around a 3x speedup on TPCH SF20 (down from ~3s to ~1s) when building the dynamic filter as an IN LIST vs. min/max bounds. Note that this required me to build the TPCH data with bloom filters included.

Thank you for investigating! I really wonder what performance would be like for some of these pushing down a bounded size bloom filter. I think other systems do this (citation needed). But it does cost at build time, etc.

Citation: https://15721.courses.cs.cmu.edu/spring2024/notes/09-hashjoins.pdf

@mbutrovich

Thank you for investigating! I really wonder what performance would be like for some of these pushing down a bounded size bloom filter. I think other systems do this (citation needed). But it does cost at build time, etc.

I think this is what spark does.

https://issues.apache.org/jira/browse/SPARK-32268 has a bit of background

I remember @mbutrovich was also working on this at some point

@Dandandan

I'm keen to try bloom filters (hence #16435). I don't have numbers to back it up (it's not implemented) but:

Like Andrew said several other systems use bloom filters for precisely this purpose.
From the CMU lecture above: "Create a Bloom Filter during the build phase when the key is likely to not exist in the hash
table [4]. Threads check the filter before probing the hash table. This will be faster since the filter will fit in
CPU caches."; this is not a benchmark on our system but the logic is sound to me, bloom filters are very efficient data structures.
Bloom filters can be serialized across the wire and can be represented in a platform agnostic way so they are compatible with distributed users of DataFusion.

I think the main argument against bloom filters and in favor of pushing down a reference to the hash table itself is that building a bloom filter can be expensive (I remember @Dandandan making this point). My counter argument to this would be that this is only a problem if the size of your build side ≈ the size of your probe side, but if that's the case you already probably have a suboptimal or slow query (and maybe a hash join wasn't even the right choice). If your build side is small then the overhead of building a bloom filter should be small relative to the speedup you get if lookups are even 15% faster than a hash table.

My counter argument to this would be that this is only a problem if the size of your build side ≈ the size of your probe side, but if that's the case you already probably have a suboptimal or slow query (and maybe a hash join wasn't even the right choice). If your build side is small then the overhead of building a bloom filter should be small relative to the speedup you get if lookups are even 15% faster than a hash table.

Maybe we can dynamically choose here too -- like if the bloom filter size isn't big enough for the number of distinct values then stop building it or something 🤔

This is now complete!

This is now complete!

Nice! I added it to the list of things to mention in the 52 release blog

Refactor create_hashes to accept array references (apache#18448)

Refactor InListExpr to support structs by re-using existing hashing i…

Refactor state management in `HashJoinExec` and use CASE expressions …

Push down InList or hash table references from HashJoinExec depending…

Refactor create_hashes to accept array references (apache#18448)

Refactor InListExpr to support structs by re-using existing hashing i…

Refactor state management in `HashJoinExec` and use CASE expressions …

Push down InList or hash table references from HashJoinExec depending…

Refactor InListExpr to support structs by re-using existing hashing i…

Refactor state management in `HashJoinExec` and use CASE expressions …

Push down InList or hash table references from HashJoinExec depending…

Bring IN LIST Dynamic Filtering work (#63)

Navigation Menu

Repository navigation

Push down entire hash table from HashJoinExec into scans #17171

30 remaining items

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Push down entire hash table from HashJoinExec into scans #17171

Description

Activity

adriangb commented on Aug 13, 2025

rkrishn7 commented on Sep 5, 2025

xudong963 commented on Sep 5, 2025

rkrishn7 commented on Sep 6, 2025

rkrishn7 commented on Sep 8, 2025

adriangb commented on Sep 8, 2025

adriangb commented on Sep 8, 2025

alamb commented on Sep 8, 2025

adriangb commented on Sep 8, 2025

alamb commented on Sep 8, 2025

30 remaining items

adriangb commented on Dec 13, 2025

alamb commented on Dec 16, 2025

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions