Skip to content

Run user-defined generators (Hive UDTFs, custom Generators) in the Comet pipeline #7252

Open

Description

@andygrove

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:

  1. Native code sends an input batch. The JVM side evaluates the bound generator's interpreted eval once 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.
  2. The JVM returns the generated columns plus a parent-row index for each output row. Native code builds the output by running take on 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.
  3. 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 for Rand;
    • an end-of-input call so terminate() can emit its rows. GenerateExec pairs those rows with the carried-forward columns of the partition's last input row, matching Hive, so native code must keep that row.

HiveGenericUDTF is CodegenFallback and converts through its own ObjectInspectors inside eval, so it needs nothing Hive-specific beyond what eval already does.

Additional context

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Metadata

Metadata

Assignees

No one assigned

    Type

    No type

    Projects

    No projects

      Milestone

      No milestone

      Relationships

      None yet

      Development

      No branches or pull requests

      Issue actions