peter-toth opened a new pull request, #59866:
URL: https://github.com/apache/spark/pull/59866
### What changes were proposed in this pull request?
A runtime bloom filter holds the join keys of the whole creation side. When
the join key of each side comes from a V2 scan, and the two scans report a
keyed partitioning by the same transform of that key, a partition of the
application side only meets the keys of one partition of the creation side.
This PR adds one bloom filter per partition key for that case, behind the new
`spark.sql.optimizer.runtime.bloomFilter.perPartition.enabled` (off by default).
`InjectRuntimeFilter` adds a left semi join of the application side with an
aggregate that builds one bloom filter per partition key of the creation side:
```
Project [fact columns]
+- Join LeftSemi, (fk = pk) AND might_contain(bloom, xxhash64(fact.k))
:- Project [fact columns, bucket(4, fact.k) AS fk]
: +- <application side>
KeyedPartitioning(bucket(4, k))
+- Aggregate [bucket(4, dim.k)], [bucket(4, dim.k) AS pk,
bloom_filter_agg(xxhash64(dim.k)) AS bloom]
+- <creation side>
KeyedPartitioning(bucket(4, k))
```
- The equality on the partition keys gives each application row the bloom of
its partition, and lets `EnsureRequirements` pair the two sides as a
storage-partitioned join does, without a shuffle.
- The new physical rule `ChoosePerPartitionBloomFilters` runs before
`EnsureRequirements`. It keeps the semi join only when the semi join is free. A
free semi join adds no shuffle, and the stage that reads the application side
runs as many tasks as without it. That holds when the join above is a
storage-partitioned join that groups the application side's splits the way the
semi join does.
- When the semi join is free, the global bloom filter of the same key
goes, from the `Filter` and from the scan's runtime filters. It stays when a V2
scan also applies it as a storage filter.
- Otherwise the semi join goes, and the global bloom filter, if any, stays.
- Each bloom filter is sized for the creation side's rows spread over its
partitions. The creation side's size threshold applies to one partition the
same way.
It applies when:
- the conf and `spark.sql.sources.v2.bucketing.enabled` are on;
- the join qualifies for a runtime bloom filter, with the size threshold per
partition as above;
- each side's join key comes from a column of a V2 scan below it, and the
scan reports a keyed partitioning by a single transform (not `identity`) of
that column;
- the two transforms are the same function, with the same bucket count;
- Spark can evaluate the transform as is, i.e. it is a scalar function that
asks for exactly the types of its arguments.
Three commits are prerequisites:
1. A partition transform generates the code of its function call. So a
projection or a grouping by it stays in whole-stage codegen when its function
has an `invoke` method.
2. `might_contain` reads its bloom filter from a column, or from any other
expression without subqueries, and reuses the bloom it deserialized while the
same array comes back.
3. A hash semi join condition reads a binary build column again only when
the build row changes, rather than for every streamed row. This one also helps
such a join that a query writes.
### Why are the changes needed?
A storage-partitioned join does not shuffle the application side. So a bloom
filter applied above the scan saves only the work between the scan and the
join. It pays off when the scan applies it, skipping the rows it rules out
while it reads. That is a follow-up, with the storage-filter API of #59715. It
needs a bloom filter per partition, and this PR builds it.
Compared with the global bloom filter that `InjectRuntimeFilter` adds to
such a join today, the per-partition one:
- is sized for one partition's keys, which also gets it a lower false
positive rate;
- is built in the task of its partition, with no separate job, no single
task merging all partial filters and no filter shipped to every task;
- also exists for a creation side too big for one bloom filter of all its
keys.
### Does this PR introduce _any_ user-facing change?
Yes, with `spark.sql.optimizer.runtime.bloomFilter.perPartition.enabled` on,
which is off by default. A join that qualifies gets a left semi
`ShuffledHashJoin` with a per-partition bloom filter aggregate, in place of the
global bloom filter. That happens when the semi join adds no shuffle and does
not cut the tasks that read the application side. `EXPLAIN` shows it.
### How was this patch tested?
- `TransformExpressionSuite`: the generated code computes what `eval` does
for each of the three calls a function can resolve to, a transform whose keys a
join reduced generates no code, and a transform whose call falls back keeps its
arguments out of subexpression elimination and of a `With`.
- `WholeStageCodegenSuite`: a transform over a function with an `invoke`
method stays in whole-stage codegen, and one with only `produceResult` runs
outside it.
- `BloomFilterAggregateQuerySuite`: `might_contain` reads a bloom filter
from a column and from an inlined aggregate, and is null for a null value, with
and without whole-stage codegen and with no codegen at all. A hash semi join
checks each row against the bloom of its build row, in the same three modes.
`might_contain` reads the bloom again only for another array, and a semi join
of the same shape that a query wrote is left alone.
- `JoinSuite`: a hash semi join condition that reads a binary build column,
and gets one array per build row, for a shuffled hash join and a broadcast hash
join, in the same three modes.
- New `PerPartitionRuntimeFilterSuite`:
- `InjectRuntimeFilter` adds the semi join next to the global bloom filter;
- with AQE on and off, the semi join stays and the global bloom filter
goes when the partitions pair, and the other way around when they would not;
- with AQE on and off, the semi join goes when the join above shuffles the
application side anyway;
- with partially clustered distribution, the semi join keeps the
application side's splits when the aggregate is the smaller side, and goes when
it would group the splits the join keeps, also under a limit, a top-N and a
write;
- when `MergeSubplans` merged the global bloom filters of two join keys
into one subquery, the semi join drops only its own key's;
- without a row count, each bloom filter expects one bucket's share of the
default number of items, or one partition's when the scan reports fewer
partitions than buckets;
- the creation side's size threshold applies to one partition;
- a later join on the same key still gets its global bloom filter when the
semi join has none, and no second semi join;
- the semi join follows the join key through a storage-partitioned join
below it, with AQE on and off;
- the semi join replaces the global bloom filter of a renamed join key,
with AQE on and off;
- with AQE on and off, a scan that offers its bucket column for runtime
filtering, as Iceberg's does, also loses the global bloom filter's copy among
its runtime filters;
- no semi join with the conf off, for an identity partitioning, for
different transforms, or for a transform whose input types differ from the
column's.
Known limits:
- a partitioning by a single transform, the same on both sides;
- the semi join is a shuffled hash join only, since its build side holds one
row per partition key;
- the semi join stays only when it is free, so a join above that shuffles
the application side gets the global bloom filter;
- each bloom filter is sized for an even share of the creation side's rows,
counted over the partitions the scan reports, or over the buckets when there
are fewer buckets. So a bloom filter can be too small, with more false
positives:
- for a skewed partition key;
- for every key, when the scan reports several partitions per key, as a
transform other than `bucket` with many files per value can;
- when `MergeSubplans` merged the global bloom filter with others into one
subquery, that subquery still builds it after the semi join replaced its check;
- with dynamic partition pruning on, a scan that offers the key column for
runtime filtering, as Iceberg's does for a `bucket` source column, gets no
runtime bloom filter on that key. `InjectRuntimeFilter` skips a key that has a
dynamic pruning filter, and a storage-partitioned join then drops that filter,
since it has no broadcast to reuse. #58635 would lift this.
Possible follow-ups:
- Different transforms on the two sides, such as `bucket(4)` against
`bucket(8)`, or an unpartitioned creation side. The aggregate can group the
creation side by the application side's transform over the creation key, and
the physical check becomes "no new shuffle on the application side".
- Applying the per-partition bloom filter inside the scan, with the
storage-filter API of #59715.
- Sizing the bloom filters from the real number of keys, and a creation-side
size threshold of their own.
- Pruning a replaced bloom filter from a merged subquery.
- Generated code for `ApplyFunctionExpression`, so a transform over a
function with only `produceResult` stays in whole-stage codegen too.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code with Claude Opus 5.5
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
