andygrove opened a new pull request, #7253:
URL: https://github.com/apache/datafusion-comet/pull/7253
## Which issue does this PR close?
Closes #7252.
Stacked on #7251 (`inline` / `stack`), whose commits appear here until it
merges. Only the last commit is new.
## Rationale for this change
A `GenerateExec` whose generator Comet has no native implementation of falls
back to Spark. That costs a columnar-to-row transition before it and a
row-to-columnar transition after it, and leaves the rest of the stage in Spark.
This affects:
- Hive `GenericUDTF`s used in `LATERAL VIEW`;
- `json_tuple`;
- the `ReplicateRows` generator that `EXCEPT ALL` and `INTERSECT ALL` plan;
- custom Catalyst generators.
The generator itself still runs on the JVM one row at a time, so the gain is
keeping the stage native around it.
## What changes are included in this PR?
The issue proposed returning generated columns plus a parent-row index. This
uses a simpler shape that fits the existing UDF bridge, which requires one
output value per input row:
- **`CometGeneratorUDF`** (a `CometUDF`): for each input row, it returns the
rows the generator emits as one `array<struct>` element. `CometExplodeExec`
then unnests and flattens that the way #7251 does for `inline`. An input row
that emits nothing gets an empty array, so `outer` keeps it as one row of
nulls, through the existing null handling.
- **Inputs and outputs:** it reads the bound generator's columns through
`CometVector` rows, and writes the output with Comet's `ArrowWriter` into a
vector from `CometBatchKernelCodegen.allocateOutput`, so the list and struct
field names match what the native side declares.
- **State:** each native plan gets its own generator instance, keyed by
digest and plan, and dropped in `releasePlan`, so the state a Hive UDTF keeps
carries across batches. The serde folds a per-occurrence number into the
digest, so two identical generators in one plan don't share an instance.
- **`terminate()`:** a new `terminate` expression on the `Explode` proto is
the same UDF called with a flag set. `ExplodeExec` evaluates it once the input
is exhausted and emits its rows with the last input row's other columns, which
it keeps as a one-row copy. That matches Spark's `GenerateExec`, which pairs
them with the last row as Hive does. `outer` does not apply to these rows. With
no input rows, the other columns are null.
- **Serde:** `CometGeneratorUDF.declineReason` and
`CometGeneratorUDF.calls`. The generator falls back when:
- it is nondeterministic (the existing gate);
- its tree holds a subquery, an unevaluable node or a native UDF call;
- it calls code outside Spark (`CometInvokeTargets`);
- its output type cannot be serialized.
- **Config:** `spark.comet.exec.jvmGenerator.enabled`, on by default.
- **Small refactors:** `CometScalaUDF.closureSerialize` is pulled out for
reuse, and `ArrowWriter` and `ArrowFieldWriter` are widened from
`private[arrow]` to `private[comet]`.
- **Docs:** a Generators section in the Scala/Java UDF page, plus the
operator, expression and plan pages.
Python UDTFs (`@udtf`) plan as their own exec nodes and are not covered.
## How are these changes tested?
- **Rust `ExplodeExec` tests:**
- `terminate` rows carry the last input row across several batches;
- with no input rows the other columns are null;
- an empty `terminate` adds no `outer` row.
- **`CometGenerateExecSuite`:**
- a test Hive `GenericUDTF` (registered without a Hive session) that keeps
a running count and emits a total from `close()`. The suite checks it runs
natively against Spark: with and without `LATERAL VIEW OUTER`, with no input
rows, and as two identical UDTFs in one plan;
- a three-partition run with AQE off, checking that each partition ends
with its own `close()` row;
- `json_tuple`, and `EXCEPT ALL` through `ReplicateRows`, both checked to
run natively;
- the fallback reason when the config is off.
- **Local runs:** on the default Spark 4.1 profile, `CometGenerateExecSuite`
(59 passed), `CometSetOpWithGroupBySuite`, and the `inline`, `stack`, `explode`
and `posexplode` SQL files.
--
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]
