dwsmith1983 opened a new issue, #6711:
URL: https://github.com/apache/datafusion-comet/issues/6711
## Describe the bug
The JVM codegen dispatcher keeps one compiled kernel per task for each
distinct pair of serialized expression bytes and column specs
(`CometScalaUDFCodegen.kernelCache`). A kernel holds the deserialized
expression, and a Catalyst `Nondeterministic` node such as
`monotonically_increasing_id()` keeps its counter inside that expression. Each
dispatched subtree is bound on its own, so `f(x)` and `f(y)` both bind their
input to ordinal 0, and two occurrences of the same subtree in one projection
serialize to identical bytes. They then share one kernel and one counter, and
the second occurrence continues where the first stopped. Spark gives every
occurrence its own state.
## Steps to reproduce
Single-partition Parquet table from `spark.range(0, 64)`,
`spark.comet.batchSize=8`, a Java UDF registered with
`spark.udf.register("javaId", (UDF1[java.lang.Long, java.lang.Long]) identity,
LongType)` (a Java UDF has no encoders, so both calls below serialize
identically; a Scala UDF gets per-occurrence encoders and does not hit this):
```sql
SELECT id, javaId(monotonically_increasing_id()) AS a,
javaId(monotonically_increasing_id()) AS b FROM t
```
Spark returns `a = b = id`. Comet returns `a = 0..7, b = 8..15` in the first
batch, `a = 16..23, b = 24..31` in the second, and so on.
The same happens for any dispatched shape without encoders once two
identical copies appear in one projection, for example `arrays_zip(array(map(x,
monotonically_increasing_id())))` and the same over another column, once #5867
routes it through the dispatcher.
## Expected behavior
Each occurrence of a nondeterministic dispatched expression keeps its own
state, as in Spark.
## Additional context
Reproduced on `main` at 9dc8c3ca8 (Spark 4.1 profile). Found while reviewing
#5867
(https://github.com/apache/datafusion-comet/pull/5867#discussion_r4194624076);
#5526 carries a similar fix (`DispatchOccurrence`) inside a larger change.
Nondeterministic user functions (`ScalaUDF` with `asNondeterministic()`,
`Invoke`, `StaticInvoke`) keep their state in one object Spark shares between
calls; that is a separate question from the Catalyst `Nondeterministic` nodes
this issue is about.
--
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]
