sunchao opened a new pull request, #5533:
URL: https://github.com/apache/datafusion-comet/pull/5533
## Which issue does this PR close?
Closes #5532.
## Rationale for this change
A query can succeed in Spark even when some input rows contain malformed
Base64: an operator may finish without asking the decoder to evaluate those
rows. Comet needs to preserve that behavior. Otherwise, enabling Comet can turn
a successful query into an error.
For example, suppose a scan produces `YWJj` followed by malformed `A` in one
partition:
```sql
SELECT hex(unbase64(bad))
FROM t
LIMIT 1;
```
With that physical input order, Spark returns `616263` (the bytes for `abc`)
and never decodes the second value. Comet can instead decode the whole batch
before the limit takes effect, failing with `Last unit does not have enough
valid bits` on a row Spark skips. The example relies on that physical input
order; SQL without `ORDER BY` does not guarantee a result order.
A semi join can skip work for the same reason:
```sql
SELECT /*+ BROADCAST(r) */ l.*
FROM left_rows l
LEFT SEMI JOIN candidates r
ON l.k = r.k AND unbase64(r.bad) > l.expected;
```
If `l.expected` is `X'616262'` (`abb`), a candidate containing `YWJj`
(`abc`) satisfies the condition. When Spark encounters that candidate before a
malformed candidate with the same key, it can return the left row without
evaluating the remaining candidates. An anti join can likewise stop after a
match. Evaluating all candidates in a batch can still reach the malformed value
and throw.
Both the native decoder and Comet's JVM codegen dispatcher evaluate batches,
so routing the expression through the dispatcher alone does not restore Spark's
behavior. Malformed terminal Base64 units can throw with ANSI mode either
enabled or disabled.
## What changes are included in this PR?
The planner now recognizes the surrounding operators that let Spark skip
decoding and keeps the affected path in Spark's row pipeline. This covers
limits, semi/anti join conditions, and early termination in ordered top-K and
`WindowGroupLimit` plans when the input already satisfies the required
ordering. It also covers compound inputs and `to_binary(..., 'base64')`, which
use Spark's `UnBase64` expression through JVM dispatch.
Keeping only the decoder in Spark is insufficient if an intermediate
conversion back to batches evaluates more rows before the consumer needs them.
The fallback therefore protects the intervening row pipeline and remains
effective when Adaptive Query Execution reuses an already converted native
subtree. Where a final aggregation must fall back, its partial aggregation also
stays in Spark if the two engines cannot share the intermediate buffer format.
The fallback is conservative within these affected paths, including when
their data happens to be valid. It stops propagating a limit's protection
through operators that must consume their input first. For example, an outer
limit cannot skip malformed values below a blocking sort, so decoding there
remains eligible for native execution. Ordinary projections, inner joins, and
semi/anti joins without decoder conditions also remain eligible.
The native decoder and Base64 validation rules are unchanged. The protection
works with JVM dispatch enabled or disabled and does not introduce a general
fallback for other throwing expressions.
## How are these changes tested?
Validated `4e2511a7` with fresh full reactor compilation, Spotless, and
Scalastyle, followed by the targeted ScalaTest suites below. The native library
was reused after verifying that the PR changes none of its source or build
inputs.
| Targeted checks | Spark 4.0.4 | Spark 4.1.3 |
| --- | --- | --- |
| Complete `CometExecRuleSuite` | 37 passed | 37 passed |
| Executed `unbase64` SQL fixture instances | 13 passed | 13 passed |
| Complete `CometScalarFunctionSuite` | 14 passed | 14 passed |
All **128 executed test cases passed**, with zero failures. These counts
exclude one existing Spark 3.5-only SQL fixture skipped by each Spark 4.x run.
A separate Spark 4.1.3 JNI probe matched Spark in all **88 comparisons**
across JVM dispatch, ANSI mode, and AQE on/off. This includes all 28 mismatches
reproduced on unchanged upstream, plus checks that safe projections and inner
joins still execute natively.
The SQL regressions compare Spark and Comet on skipped malformed values,
consumed malformed values, and valid/null inputs. They vary ANSI mode and AQE,
and the general operator fixture also varies JVM dispatch. Native-execution
controls check that safe projections and joins remain native.
The planner tests exercise all physical limit variants, ordered top-K,
`WindowGroupLimit` ranks and modes, each semi/anti join strategy, intermediate
batch conversions, repeated rule application, reused native subtrees, and
keeping partial and final aggregates together when their buffer formats are
incompatible. They also verify that materialization barriers retain native
decoding.
The full repository test suite was not run locally.
--
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]
