rdtr opened a new issue, #13489:
URL: https://github.com/apache/gluten/issues/13489
### Backend
VL (Velox)
### Bug description
## Bug description
Gluten rewrites every `BloomFilterAggregate` and every
`BloomFilterMightContain` in a physical plan to `VeloxBloomFilterAggregate` /
`VeloxBloomFilterMightContain` (`BloomFilterMightContainJointRewriteRule`). The
Velox variants use Velox's serialized bloom filter format and hash function, so
the rewrite is only correct when the producer and the consumer of the bytes are
both rewritten. The rule cannot check that: under AQE a scalar subquery is
compiled into its own `AdaptiveSparkPlanExec`, so the physical rule sees the
`bloom_filter_agg` stage without its `might_contain`. The only exception today
is `DataFrame.stat.bloomFilter()`, which is detected by inspecting the driver
stack trace.
The expressions are reachable by users: `bloom_filter_agg` / `might_contain`
can be registered with `spark.sessionState.functionRegistry.registerFunction`
(that is what Spark's own `BloomFilterAggregateQuerySuite` does),
`bloom_filter_agg` is a registered internal function on Spark 4.0+
(`Column.internalFn("bloom_filter_agg", ...)`), and `df.stat.bloomFilter()` on
Spark Connect goes through that internal function on the server, where no stat
frame is on the stack.
Observed on Spark 4.0 with the functions registered as in Spark's
`BloomFilterAggregateQuerySuite`:
1. `bloom_filter_agg(stringCol)` paired with `might_contain`: the aggregate
fails validation (Velox only accepts bigint) and runs in the JVM as
`VeloxBloomFilterAggregate`, whose string updater throws
`UnsupportedOperationException: Not yet implemented` from
`VeloxBloomFilter.putBinary`. Vanilla Spark accepts byte/short/int/long/string.
2. `bloom_filter_agg(nullableLongCol)` with null rows: Velox throws `First
argument of bloom_filter_agg cannot be null`. Vanilla Spark ignores nulls.
3. `might_contain(<Spark-format bytes literal>, col)`: Velox throws
`Unsupported BloomFilter version: 0` because the first byte of Spark's format
is the big-endian version 1, while Velox's format starts with its version byte
1.
4. A build-only `bloom_filter_agg` whose bytes leave the query, e.g.
`Column.internalFn("bloom_filter_agg", col, lit(n), lit(bits))` followed by
`BloomFilter.readFrom`: the bytes are Velox-format, so `readFrom` fails with
`IOException: Unexpected Bloom filter version number (25165824)`. This is what
`df.stat.bloomFilter()` does on Spark Connect.
5. `bloom_filter_agg(intCol)`: correct, but the aggregate falls back to the
JVM because the Velox signature is bigint only. Vanilla Spark widens
int/short/byte to long.
Proposed fix: decide the rewrite on the logical plan, where the producer and
consumer are both visible, and leave no Spark `BloomFilterAggregate` /
`BloomFilterMightContain` behind that a later physical rule could mistake for a
runtime filter:
- A pre-CBO optimizer rule (runs before `InjectRuntimeFilter` and
`MergeScalarSubqueries`, and before AQE compiles subqueries) rewrites a
`might_contain` whose bloom filter is a scalar subquery computing
`bloom_filter_agg` over a bigint-compatible input to the Velox pair.
int/short/byte inputs are cast to bigint, nullable inputs get `FILTER (WHERE
input IS NOT NULL)` so Velox never sees a null. A Velox-format constant is also
accepted.
- Every other `bloom_filter_agg` / `might_contain` (build-only aggregates,
string inputs, Spark-format constants, subqueries that read a stored filter)
becomes a marker expression that behaves exactly like Spark's and is never
offloaded, so the bytes stay in Spark's format.
- The physical rule keeps rewriting the pairs that `InjectRuntimeFilter`
adds after the optimizer rule; those are the only Spark-class bloom filter
expressions left. The stack-trace detection of `stat.bloomFilter()` is no
longer needed.
### Gluten version
main branch
### Spark version
Spark-4.0.x
### Spark configurations
Defaults.
### System information
N/A
### Relevant logs
```bash
java.lang.UnsupportedOperationException: Not yet implemented
at
org.apache.gluten.utils.VeloxBloomFilter.putBinary(VeloxBloomFilter.java:...)
org.apache.gluten.exception.GlutenException: Exception: VeloxUserError
Error Code: INVALID_ARGUMENT
Reason: First argument of bloom_filter_agg cannot be null
org.apache.gluten.exception.GlutenException: Exception: VeloxUserError
Reason: Unsupported BloomFilter version: 0
java.io.IOException: Unexpected Bloom filter version number (25165824)
at org.apache.spark.util.sketch.BloomFilter.readFrom(...)
```
--
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]
