Repository navigation
Run user-defined generators (Hive UDTFs, custom Generators) in the Comet pipeline #7252
Description
What is the problem the feature request solves?
A GenerateExec whose generator is user-defined falls back to Spark. This covers Hive GenericUDTFs (CREATE FUNCTION ... AS 'com.example.MyUDTF' used in LATERAL VIEW, planned as HiveGenericUDTF) and custom Catalyst Generator expressions injected through session extensions. The fallback costs a columnar-to-row transition before the generator and a row-to-columnar transition after it, and the rest of the stage, often an aggregate over the generated rows, runs in Spark.
The generator would still run on the JVM one row at a time, so the generator itself would not get faster. The gain is keeping the stage native around it.
Comet already runs scalar ScalaUDFs in the native pipeline through codegen dispatch, but that path rejects any Generator (CometBatchKernelCodegen.canHandle), because its kernel contract is one output value per input row.
Describe the potential solution
A native generate operator that calls back into the JVM through the existing UDF bridge:
- Native code sends an input batch. The JVM side evaluates the bound generator's interpreted
evalonce per row, reading through the Arrow-backed row getters that codegen dispatch already uses, and writes the generated rows into Arrow vectors with the existing output writers. - The JVM returns the generated columns plus a parent-row index for each output row. Native code builds the output by running
takeon the carried-forward columns with those indices and appending the generated columns.outer(one row of NULLs when the generator yields nothing) is an index with a null generated row. - The bridge gains two lifecycle pieces it lacks today:
- per-partition generator state that lives across batches, with
Nondeterministic.initialize(partitionIndex), as the per-plan kernel cache already does forRand; - an end-of-input call so
terminate()can emit its rows.GenerateExecpairs those rows with the carried-forward columns of the partition's last input row, matching Hive, so native code must keep that row.
- per-partition generator state that lives across batches, with
HiveGenericUDTF is CodegenFallback and converts through its own ObjectInspectors inside eval, so it needs nothing Hive-specific beyond what eval already does.
Additional context
- Python UDTFs (
@udtf) plan as their own Python UDTF exec nodes, notGenerateExec, and belong with the Arrow Python UDF work (Extend native Arrow UDF path to scalar Python UDFs (ArrowEvalPythonExec) #5386, Extend native Arrow UDF path to grouped-aggregate, window, and applyInArrow Python operators #5123). - feat: run inline, inline_outer, and stack generators natively #7251 (for Support inline and stack generators in GenerateExec #5125) runs
inlineandstacknatively. The generated-columns-plus-parent-index shape here could also replace the list unnesting those use, thoughExplodeExecalready handles them well.
Activity