Repository navigation
Conversation
Plan `inline` through the native explode operator by flattening each unnested struct into one column per field, and plan `stack` as `inline` over an array holding one struct per output row. Flattening now folds a struct's null mask into its fields, so a null struct element yields a row of nulls as Spark's `inline` does. Closes apache#5125.
…atch - Flatten structs with `children_with_parent_nulls` from comet-common, which skips the full revalidation that rebuilding each field did. - Pick the generator in one match on `Inline`, `Stack` and `ExplodeBase` instead of mixing type matches with node-name checks. - Lift the fallback reasons of the array `stack` builds onto the generator, so the operator reports the failing argument's own reason. - Test a foldable `stack` row count and NULL structs built by `IF`.
A generator Comet has no native implementation of, such as a Hive UDTF, `json_tuple`, or the `ReplicateRows` that `EXCEPT ALL` plans, now runs on the JVM inside `CometExplodeExec` instead of falling back to Spark. `CometGeneratorUDF` is a `CometUDF` that returns, for each input row, the rows the generator emits as one `array<struct>` element, which the operator unnests and flattens as it does for `inline`. Each native plan gets its own generator instance, so state carries across batches. Once the input ends, `ExplodeExec` evaluates a new `terminate` expression and emits the rows of `Generator.terminate` with the last input row's other columns, as Spark's `GenerateExec` does. Gated by `spark.comet.exec.jvmGenerator.enabled`, on by default. Closes apache#7252.
andygrove
added
the
run-spark-4.1-tests
github-actions
Bot
added
enhancement
…s the generator's
rich7420
left a comment
rich7420
left a comment
There was a problem hiding this comment.
Three correctness issues inline.
| let input_schema = self.input.schema(); | ||
| let mut columns = match self.last_row.take() { | ||
| Some(row) => row.columns().to_vec(), | ||
| None => input_schema |
There was a problem hiding this comment.
Discard empty termination output before creating null placeholders; nonnullable carried columns currently make empty partitions fail.
| val outputRow = new GenericInternalRow(1) | ||
| def writeRows(rows: Any): Unit = { | ||
| val copies = new ArrayBuffer[Any] | ||
| iteratorOf(rows).foreach(row => copies += row.copy()) |
There was a problem hiding this comment.
Snapshot rows with UnsafeProjection; GenericInternalRow.copy() retains reused binary buffers, producing duplicate values instead of the original outputs.
| val copies = new ArrayBuffer[Any] | ||
| iteratorOf(rows).foreach(row => copies += row.copy()) | ||
| outputRow.update(0, new GenericArrayData(copies.toArray)) | ||
| writer.write(outputRow, 0) |
There was a problem hiding this comment.
Keep custom generators on Spark until this bridge preserves laziness; LIMIT 1 currently evaluates a throwing second row.
sunchao
left a comment
sunchao
left a comment
There was a problem hiding this comment.
Reviewed edb56380fb95f1c57ca45fffedd4186442b46772 across five independent scopes. The three existing correctness findings are still present on this head, so I am requesting changes without adding duplicate inline threads:
- Empty termination with nonnullable carried columns:
terminate_batchconstructs a null-filledRecordBatchbefore discovering that termination emitted no rows. - Reused mutable row buffers:
GenericInternalRow.copy()does not snapshot a reused binary buffer before the generator advances. - Generator laziness: draining every iterator before returning a batch can evaluate a throwing second row under
LIMIT 1, and can materialize unbounded output.
The existing suite/helper reuse and per-plan generator state are reasonable. Please add focused regression cases for those three boundaries with the fixes. Current-head PR checks are green and suite registration checks passed, but the added tests do not cover these cases. I inspected the current production and test paths; I did not rerun the full Spark/Comet suites.
sunchao
left a comment
sunchao
left a comment
There was a problem hiding this comment.
Re-reviewed edb56380fb95f1c57ca45fffedd4186442b46772 across five independent scopes. This head is unchanged, and the three existing correctness issues remain:
- Empty termination with nonnullable carried columns.
- Shallow copies of reused binary buffers.
- Eager generator evaluation under LIMIT.
The implementation and tests still need the fixes and focused regressions requested in those threads. No additional finding from this pass. Current-head checks have 35 successes and 18 skips; I rechecked the source and test coverage without rerunning the unchanged full suites. Keeping changes requested.
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
GenerateExecwhose 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:GenericUDTFs used inLATERAL VIEW;json_tuple;ReplicateRowsgenerator thatEXCEPT ALLandINTERSECT ALLplan;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(aCometUDF): for each input row, it returns the rows the generator emits as onearray<struct>element.CometExplodeExecthen unnests and flattens that the way feat: run inline, inline_outer, and stack generators natively #7251 does forinline. An input row that emits nothing gets an empty array, soouterkeeps it as one row of nulls, through the existing null handling.CometVectorrows, and writes the output with Comet'sArrowWriterinto a vector fromCometBatchKernelCodegen.allocateOutput, so the list and struct field names match what the native side declares.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 newterminateexpression on theExplodeproto is the same UDF called with a flag set.ExplodeExecevaluates 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'sGenerateExec, which pairs them with the last row as Hive does.outerdoes not apply to these rows. With no input rows, the other columns are null.CometGeneratorUDF.declineReasonandCometGeneratorUDF.calls. The generator falls back when:CometInvokeTargets);spark.comet.exec.jvmGenerator.enabled, on by default.CometScalaUDF.closureSerializeis pulled out for reuse, andArrowWriterandArrowFieldWriterare widened fromprivate[arrow]toprivate[comet].Python UDTFs (
@udtf) plan as their own exec nodes and are not covered.How are these changes tested?
ExplodeExectests:terminaterows carry the last input row across several batches;terminateadds noouterrow.CometGenerateExecSuite:GenericUDTF(registered without a Hive session) that keeps a running count and emits a total fromclose(). The suite checks it runs natively against Spark: with and withoutLATERAL VIEW OUTER, with no input rows, and as two identical UDTFs in one plan;close()row;json_tuple, andEXCEPT ALLthroughReplicateRows, both checked to run natively;CometGenerateExecSuite(59 passed),CometSetOpWithGroupBySuite, and theinline,stack,explodeandposexplodeSQL files.