Push down InList or hash table references from HashJoinExec depending on the size of the build side #18393
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
github-actions
bot
added
core
github-actions
bot
added
the
documentation
adriangb
commented
Oct 31, 2025
There was a problem hiding this comment.
Leaving my review comments, will post benchmarks afterwards.
Although this PR is large I think there's a clear path to split it up into independent smaller PRs:
- Refactor
create_hashesto accept references (changes only tohash_utils.rs) - Refactor
InListExprto store arrays and support structs, re-usingcreate_hashes_from_arraysfrom (1) (changes only toin_list.rs). - Refactor the data structures used to track pushdown data in HashJoinExec (changes only to files in
datafusion/physical-plan/src/joins/hash_join/). - Introduce the
CASEstatement structure into the filter pushdown ofHashJoinExecand the repartition hash PhysicalExpr (changes only to files indatafusion/physical-plan/src/joins/hash_join/, addsHashExprindatafusion/physical-plan/src/joins/hash_join/partitioned_hash_eval.rs). - Add hash table pushdown (adds
HashTableLookupExprindatafusion/physical-plan/src/joins/hash_join/partitioned_hash_eval.rs), addsHashTabletoPushdownStrategy, addscreate_membership_predicate, etc.). - Add
InListExprpushdown (addsPushdownStrategy::InList, etc.)
There is also potential to remove the barrier expression by filtering out only from known partitions CASE ... ELSE true i.e. if we don't have information for all partitions only filter out data that we know won't match in our partition and then update the filter to ELSE false in the case where we have information from all partitions. This might be useful for the distributed case, not sure though.
I think we could somehow unify the hashing / inner data structures of the join hash table and the InList expression - they are very similar - to at least eliminate one round of hashing. I wonder if there's a version of an InList expression that avoids building a Vec<ScalarValue> altogether and instead just wraps an ArrayRef + metadata (data types etc) + an optional hash lookup. That would be quite versatile, we could essentially replace the join hash tables with that structure.
datafusion/common/src/hash_utils.rs
Outdated
Comment on lines
377
to
379
| #[cfg(not(feature = "force_hash_collisions"))] | ||
| pub fn create_hashes<'a>( | ||
| arrays: &[ArrayRef], | ||
| pub fn create_hashes_from_arrays<'a>( | ||
| arrays: &[&dyn Array], |
There was a problem hiding this comment.
Recommend this for its own PR.
I think this is a nice refactor for this function, however I decided not to deprecate / replace it to avoid churn. If this were it's own PR I think it would be worth it to just make a new version and deprecate the old one, replacing all references in DataFusion (which I imagine is most users; even if this is pub I think it's mostly pub to be used in other crates within this repo). I did not investigate if this can help avoid clones in any other call sites.
| - RepartitionExec: partitioning=Hash([a@0, b@1], 12), input_partitions=1 | ||
| - DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[a, b, e], file_type=test, pushdown_supported=true, predicate=DynamicFilter [ a@0 >= ab AND a@0 <= ab AND b@1 >= bb AND b@1 <= bb OR a@0 >= aa AND a@0 <= aa AND b@1 >= ba AND b@1 <= ba ] | ||
| - RepartitionExec: partitioning=Hash([a@0, b@1], 4), input_partitions=1 | ||
| - DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[a, b, e], file_type=test, pushdown_supported=true, predicate=DynamicFilter [ CASE hash_repartition % 4 WHEN 0 THEN a@0 >= aa AND a@0 <= aa AND b@1 >= ba AND b@1 <= ba AND struct(a@0, b@1) IN (SET) ([{c0:aa,c1:ba}]) WHEN 2 THEN a@0 >= ab AND a@0 <= ab AND b@1 >= bb AND b@1 <= bb AND struct(a@0, b@1) IN (SET) ([{c0:ab,c1:bb}]) ELSE false END ] |
There was a problem hiding this comment.
Note that this automatically excludes empty partitions and defaults to false, which works with inner joins. I'm not sure how we'd structure this for other join types.
| - RepartitionExec: partitioning=Hash([a@0, b@1], 12), input_partitions=1 | ||
| - DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[a, b, e], file_type=test, pushdown_supported=true, predicate=DynamicFilter [ a@0 >= aa AND a@0 <= ab AND b@1 >= ba AND b@1 <= bb ] | ||
| - RepartitionExec: partitioning=Hash([a@0, b@1], 4), input_partitions=1 | ||
| - DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[a, b, e], file_type=test, pushdown_supported=true, predicate=DynamicFilter [ a@0 >= aa AND a@0 <= ab AND b@1 >= ba AND b@1 <= bb AND struct(a@0, b@1) IN (SET) ([{c0:aa,c1:ba}, {c0:ab,c1:bb}]) ] |
There was a problem hiding this comment.
In this case there's only 1 partition with data (because of hash collisions) -> we optimize away the CASE expression. This is relevant because the same thing would happen with a point lookup primary key join.
Comment on lines
147
to
151
| /// Specialized Set implementation for StructArray | ||
| struct StructArraySet { | ||
| array: Arc<StructArray>, | ||
| hash_set: ArrayHashSet, | ||
| } |
There was a problem hiding this comment.
I think this is a nice improvement / feature that can easily be it's own PR.
| let has_nulls = self.array.null_count() != 0; | ||
|
|
||
| // Compute hashes for all rows in the input array | ||
| let mut input_hashes = vec![0u64; v.len()]; |
There was a problem hiding this comment.
Could we use a thread local to avoid repeated re-allocations here? I imagine that would work quite well.
Comment on lines
-110
to
-111
| /// Create a new `JoinLeftData` from its parts | ||
| pub(super) fn new( |
There was a problem hiding this comment.
Clippy was complaining about too many arguments, a constructor was not necessary anyway
Comment on lines
+1496
to
+1509
| let membership = if num_rows == 0 { | ||
| PushdownStrategy::Empty | ||
| } else { | ||
| // If the build side is small enough we can use IN list pushdown. | ||
| // If it's too big we fall back to pushing down a reference to the hash table. | ||
| // See `PushdownStrategy` for more details. | ||
| if let Some(in_list_values) = | ||
| build_struct_inlist_values(&left_values, max_inlist_size)? | ||
| { | ||
| PushdownStrategy::InList(in_list_values) | ||
| } else { | ||
| PushdownStrategy::HashTable(Arc::clone(&hash_map)) | ||
| } | ||
| }; |
There was a problem hiding this comment.
I did some remodeling of the data structures we use to track state. I think the new structure is much better - even if we didn't move forward with the rest of the PR.
Comment on lines
63
to
73
| // Size check using built-in method | ||
| // This is not 1:1 with the actual size of ScalarValues, but it is a good approximation | ||
| // and at this point is basically "free" to compute since we have the arrays already. | ||
| let estimated_size = join_key_arrays | ||
| .iter() | ||
| .map(|arr| arr.get_array_memory_size()) | ||
| .sum::<usize>(); | ||
|
|
||
| if estimated_size > max_size_bytes { | ||
| return Ok(None); | ||
| } |
There was a problem hiding this comment.
Note: this is where we check the size to set the size limit.
| @@ -0,0 +1,292 @@ | |||
| // Licensed to the Apache Software Foundation (ASF) under one | |||
There was a problem hiding this comment.
This is the part used to handle the cases where we push down the entire hash table.
Comment on lines
208
to
236
| /// Build-side data reported by a single partition | ||
| pub(crate) enum PartitionBuildData { | ||
| Partitioned { | ||
| partition_id: usize, | ||
| pushdown: PushdownStrategy, | ||
| bounds: PartitionBounds, | ||
| }, | ||
| CollectLeft { | ||
| pushdown: PushdownStrategy, | ||
| bounds: PartitionBounds, | ||
| }, | ||
| } | ||
|
|
||
| /// Per-partition accumulated data (Partitioned mode) | ||
| #[derive(Clone)] | ||
| struct PartitionData { | ||
| bounds: PartitionBounds, | ||
| pushdown: PushdownStrategy, | ||
| } | ||
|
|
||
| /// Build-side data organized by partition mode | ||
| enum AccumulatedBuildData { | ||
| Partitioned { | ||
| partitions: Vec<Option<PartitionData>>, | ||
| }, | ||
| CollectLeft { | ||
| data: Option<PartitionData>, | ||
| }, | ||
| } |
There was a problem hiding this comment.
This is the refactoring of the data structures we store to track state. It's now much cleaner, e.g. avoids comparing partitions by id.
|
Here's one interesting benchmark: COPY (SELECT uuid() as k, uuid() as v FROM generate_series(1, 5) t(i))
TO 'small_table_uuids.parquet'
OPTIONS (
'MAX_ROW_GROUP_SIZE' '50000',
'BLOOM_FILTER_ENABLED::k' 'true'
);
COPY (SELECT random()::text as v1, random()::text as v2, uuid() as k FROM generate_series(1, 100000000) t(i))
TO 'large_table_uuids.parquet'
OPTIONS (
'MAX_ROW_GROUP_SIZE' '50000',
'BLOOM_FILTER_ENABLED::k' 'true'
);
CREATE EXTERNAL TABLE small_table STORED AS PARQUET LOCATION 'small_table_uuids.parquet';
CREATE EXTERNAL TABLE large_table STORED AS PARQUET LOCATION 'large_table_uuids.parquet';
SET datafusion.execution.parquet.pushdown_filters = true;
SET datafusion.execution.parquet.reorder_filters = true;
SET datafusion.runtime.metadata_cache_limit = 0;
-- Join the two tables, with a filter on small_table
SELECT *
FROM small_table s JOIN large_table l ON s.k = l.k;This is similar to the query we benchmarked in our recent blog post but using UUIDs instead of Full benchmark script#!/usr/bin/env python3
"""
Benchmark script comparing DataFusion with/without inlist pushdown vs DuckDB.
Groups:
1. branch (no inlist): hash_join_inlist_pushdown_max_size = 0
2. branch (w/ inlist): hash_join_inlist_pushdown_max_size = default (999999)
3. main: using datafusion-cli-main
4. duckdb: using duckdb CLI
"""
import subprocess
import tempfile
import time
import os
import sys
from pathlib import Path
# Configuration
DATAFUSION_CLI = "./target/release/datafusion-cli"
DATAFUSION_CLI_MAIN = "./datafusion-cli-main"
DUCKDB_CLI = "duckdb"
NUM_RUNS = 5 # Number of times to run each benchmark
# Data generation settings
SMALL_TABLE_SIZE = 5
LARGE_TABLE_SIZE = 100_000_000
SMALL_TABLE_FILE = "small_table_uuids.parquet"
LARGE_TABLE_FILE = "large_table_uuids.parquet"
def run_command(cmd, input_sql=None, description=""):
"""Run a command and measure execution time."""
print(f" Running: {description}...", end=" ", flush=True)
start = time.time()
try:
if input_sql:
result = subprocess.run(
cmd,
input=input_sql,
capture_output=True,
text=True,
timeout=600 # 10 minute timeout
)
else:
result = subprocess.run(
cmd,
capture_output=True,
text=True,
timeout=600
)
elapsed = time.time() - start
if result.returncode != 0:
print(f"FAILED (exit code {result.returncode})")
print(f" stderr: {result.stderr}")
return None
print(f"{elapsed:.3f}s")
return elapsed
except subprocess.TimeoutExpired:
print("TIMEOUT")
return None
except Exception as e:
print(f"ERROR: {e}")
return None
def create_data():
"""Create test data files if they don't exist."""
if os.path.exists(SMALL_TABLE_FILE) and os.path.exists(LARGE_TABLE_FILE):
print(f"Data files already exist, skipping creation.")
return True
print(f"Creating test data...")
data_gen_sql = f"""
COPY (SELECT uuid() as k, uuid() as v FROM generate_series(1, {SMALL_TABLE_SIZE}) t(i))
TO '{SMALL_TABLE_FILE}'
OPTIONS (
'MAX_ROW_GROUP_SIZE' '50000',
'BLOOM_FILTER_ENABLED::k' 'true'
);
COPY (SELECT random()::text as v1, random()::text as v2, uuid() as k FROM generate_series(1, {LARGE_TABLE_SIZE}) t(i))
TO '{LARGE_TABLE_FILE}'
OPTIONS (
'MAX_ROW_GROUP_SIZE' '50000',
'BLOOM_FILTER_ENABLED::k' 'true'
);
"""
result = subprocess.run(
[DATAFUSION_CLI],
input=data_gen_sql,
capture_output=True,
text=True
)
if result.returncode != 0:
print(f"Failed to create data: {result.stderr}")
return False
print(f"Data created successfully.")
return True
def create_datafusion_sql(inlist_size):
"""Create SQL for DataFusion with specified inlist pushdown size."""
return f"""
CREATE EXTERNAL TABLE small_table STORED AS PARQUET LOCATION '{SMALL_TABLE_FILE}';
CREATE EXTERNAL TABLE large_table STORED AS PARQUET LOCATION '{LARGE_TABLE_FILE}';
SET datafusion.execution.parquet.pushdown_filters = true;
SET datafusion.execution.parquet.reorder_filters = true;
SET datafusion.optimizer.hash_join_inlist_pushdown_max_size = {inlist_size};
SET datafusion.runtime.metadata_cache_limit = '0M';
SELECT *
FROM small_table s JOIN large_table l ON s.k = l.k;
"""
def create_duckdb_sql():
"""Create SQL for DuckDB."""
return f"""
SELECT *
FROM '{SMALL_TABLE_FILE}' s JOIN '{LARGE_TABLE_FILE}' l ON s.k = l.k;
"""
def run_benchmark_group(name, cmd, sql_content, num_runs=NUM_RUNS):
"""Run a benchmark group multiple times and collect results."""
print(f"\n{name}:")
times = []
for i in range(num_runs):
elapsed = run_command(cmd, input_sql=sql_content, description=f"Run {i+1}/{num_runs}")
if elapsed is not None:
times.append(elapsed)
if times:
avg = sum(times) / len(times)
min_time = min(times)
max_time = max(times)
print(f" Results: min={min_time:.3f}s, avg={avg:.3f}s, max={max_time:.3f}s")
return times
else:
print(f" No successful runs")
return []
def main():
print("=" * 60)
print("DataFusion Inlist Pushdown Benchmark")
print("=" * 60)
# Verify executables exist
if not os.path.exists(DATAFUSION_CLI):
print(f"Error: {DATAFUSION_CLI} not found")
sys.exit(1)
if not os.path.exists(DATAFUSION_CLI_MAIN):
print(f"Error: {DATAFUSION_CLI_MAIN} not found")
sys.exit(1)
try:
subprocess.run([DUCKDB_CLI, "--version"], capture_output=True, check=True)
except (subprocess.CalledProcessError, FileNotFoundError):
print(f"Error: duckdb CLI not found or not working")
sys.exit(1)
# Create data
if not create_data():
sys.exit(1)
# Run benchmarks
results = {}
# 1. Branch without inlist pushdown
results["branch_no_inlist"] = run_benchmark_group(
"Branch (no inlist, size=0)",
[DATAFUSION_CLI],
create_datafusion_sql(0)
)
# 2. Branch with inlist pushdown
results["branch_with_inlist"] = run_benchmark_group(
"Branch (w/ inlist, size=999999)",
[DATAFUSION_CLI],
create_datafusion_sql(999999)
)
# 3. Main branch
results["main"] = run_benchmark_group(
"Main branch",
[DATAFUSION_CLI_MAIN],
create_datafusion_sql(999999)
)
# 4. DuckDB
results["duckdb"] = run_benchmark_group(
"DuckDB",
[DUCKDB_CLI],
create_duckdb_sql()
)
# Summary
print("\n" + "=" * 60)
print("SUMMARY")
print("=" * 60)
for name, times in results.items():
if times:
avg = sum(times) / len(times)
print(f"{name:25s}: {avg:.3f}s avg over {len(times)} runs")
else:
print(f"{name:25s}: No successful runs")
print("\nAll times (seconds):")
for name, times in results.items():
if times:
times_str = ", ".join(f"{t:.3f}" for t in times)
print(f" {name}: [{times_str}]")
if __name__ == "__main__":
main() |
|
|
Surprised that duckdb takes so long, I was thinking it also can push down a inlist for hash join |
As far as I know they only do min/max stats: https://duckdb.org/2024/09/09/announcing-duckdb-110#dynamic-filter-pushdown-from-joins |
|
Here are some new numbers after a couple more optimizations to InListExpr: I'd like to clarify why the InListExpr makes such a difference:
I think it's mainly that latter optimization that provides the win. And TPCH: |
- Fix error handling in shared_bounds.rs: properly propagate errors instead of silently swallowing with .ok() - Improve hash collision handling in InListExpr by sampling array values - Run cargo fmt to fix formatting issues Addresses CI feedback on PR apache#18438. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <[email protected]>
|
Marking as draft again. Looking at the changes while they're mostly in the right spirit there's a lot of details that need more manual care. I'll do that before marking as ready for review. |
|
I've done some cleanup, the first three PRs are ready for review:
That will be most of the changes, the only two follwup PRs will be to push down the hash table references and InListExpr |
This was referenced Nov 3, 2025
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
This PR is part of an EPIC to push down hash table references from HashJoinExec into scans. The EPIC is tracked in #17171.
A "target state" is tracked in #18393 (this PR).
There is a series of PRs to get us to this target state in smaller more reviewable changes that are still valuable on their own:
HashJoinExecand use CASE expressions for more precise filters #18451As those are merged I will rebase this PR to keep track of the "remaining work", and we can use this PR to explore big picture ideas or benchmarks of the final state.