feat: run user-defined generators inside the native generate operator

docs/source/contributor-guide/expression-audits/generator_funcs.md

@@ -37,4 +37,16 @@
 
 - Same `CometExplodeExec` path as `posexplode`. Compatible for array inputs; empty and NULL arrays both emit one row with null `pos` and null `value` per Spark's `outer` semantics, which the planner requests as DataFusion's `NullHandling::PreserveAndExpandEmpty`.
 
+## inline
+
+- Same `CometExplodeExec` path as `explode`, with the unnested struct flattened into one column per field. Spark 3.4.3, 3.5.8, 4.0.1, and 4.1.1 share one implementation: a NULL struct element yields a row of NULLs, which the native flattening reproduces by folding the struct's validity into each field. Every generated column is reported nullable.
+
+## inline_outer
+
+- Same path as `inline`. Empty and NULL arrays both emit one row of NULLs, which the planner requests as DataFusion's `NullHandling::PreserveAndExpandEmpty`.
+
+## stack
+
+- Planned as `inline` over an array holding one struct per output row, so `stack(n, v1, ..., vk)` evaluates each argument once per input row, as Spark does, and pads the last row with NULLs. Spark 3.4.3, 3.5.8, 4.0.1, and 4.1.1 share one implementation. An argument that cannot run natively falls back with that argument's own reason.
+
 [Spark Expression Support]: ../../user-guide/latest/expressions.md

docs/source/user-guide/latest/expressions.md

@@ -336,20 +336,22 @@ The type-name conversion functions (`bigint`, `binary`, `boolean`, `date`, `deci
 
 ## generator_funcs
 
-`explode`, `explode_outer`, `posexplode`, and `posexplode_outer` are supported via
-`CometExplodeExec` (operator-level, not expression-level) for array input; map input falls back
-to Spark ([#2837](https://github.com/apache/datafusion-comet/issues/2837)). Enabled by default via
-`spark.comet.exec.explode.enabled`.
+`explode`, `explode_outer`, `posexplode`, `posexplode_outer`, `inline`, `inline_outer`, and
+`stack` are supported via `CometExplodeExec` (operator-level, not expression-level). `explode` and
+`posexplode` support array input; map input falls back to Spark
+([#2837](https://github.com/apache/datafusion-comet/issues/2837)). Enabled by default via
+`spark.comet.exec.explode.enabled`. Other deterministic generators run on the JVM inside
+`CometExplodeExec`; see [Generators](scala_java_udfs.md#generators).
 
 | Function | Status | Implementation | Notes |
 | --- | --- | --- | --- |
 | `explode` | ✅ | — | via `CometExplodeExec` |
 | `explode_outer` | ✅ | — | via `CometExplodeExec` ([audit](../../contributor-guide/expression-audits/generator_funcs.md#explode_outer)) |
-| `inline` | 🔜 | — | Operator-level generator (like `explode`) |
-| `inline_outer` | 🔜 | — | Operator-level generator (like `explode`) |
+| `inline` | ✅ | — | via `CometExplodeExec` ([audit](../../contributor-guide/expression-audits/generator_funcs.md#inline)) |
+| `inline_outer` | ✅ | — | via `CometExplodeExec` ([audit](../../contributor-guide/expression-audits/generator_funcs.md#inline_outer)) |
 | `posexplode` | ✅ | — | via `CometExplodeExec` |
 | `posexplode_outer` | ✅ | — | via `CometExplodeExec` ([audit](../../contributor-guide/expression-audits/generator_funcs.md#posexplode_outer)) |
-| `stack` | 🔜 | — | Operator-level generator |
+| `stack` | ✅ | — | via `CometExplodeExec`, planned as `inline` ([audit](../../contributor-guide/expression-audits/generator_funcs.md#stack)) |
 
 ---
 
@@ -375,7 +377,7 @@ to Spark ([#2837](https://github.com/apache/datafusion-comet/issues/2837)). Enab
 | `get_json_object` | ✅ | Hybrid | Routes through the JVM codegen dispatcher by default; the native path is opt-in via allowIncompatible ([audit](../../contributor-guide/expression-audits/json_funcs.md#get_json_object)) |
 | `json_array_length` | ✅ | Hybrid | Routes through the JVM codegen dispatcher by default; the native path, which differs for single-quoted JSON, unescaped control characters, and trailing content, is opt-in via allowIncompatible ([audit](../../contributor-guide/expression-audits/json_funcs.md#json_array_length)) |
 | `json_object_keys` | ✅ | Codegen dispatch |  |
-| `json_tuple` | 🔜 | — | [#3160](https://github.com/apache/datafusion-comet/issues/3160) |
+| `json_tuple` | ✅ | — | Runs on the JVM inside `CometExplodeExec` ([Generators](scala_java_udfs.md#generators)); [#3160](https://github.com/apache/datafusion-comet/issues/3160) tracks a native implementation |
 | `schema_of_json` | ✅ | Codegen dispatch |  |
 | `to_json` | ✅ | Hybrid | Routes through the JVM codegen dispatcher by default, including options and map/array inputs; the native path is opt-in via allowIncompatible ([audit](../../contributor-guide/expression-audits/json_funcs.md#to_json)) |
 

docs/source/user-guide/latest/operators.md

@@ -124,12 +124,12 @@ combined with `OR`, which Spark plans as an existence `BroadcastNestedLoopJoinEx
 
 ## Generators and set operations
 
-| Operator       | Status | Notes                                                                                                            |
-| -------------- | ------ | ---------------------------------------------------------------------------------------------------------------- |
-| `GenerateExec` | ✅     | Supports `explode`, `explode_outer`, `posexplode`, `posexplode_outer` over arrays. `inline` / `stack` fall back. |
-| `ExpandExec`   | ✅     |                                                                                                                  |
-| `UnionExec`    | ✅     |                                                                                                                  |
-| `CoalesceExec` | ✅     |                                                                                                                  |
+| Operator       | Status | Notes                                                                                                                                                                                                                                                                |
+| -------------- | ------ | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
+| `GenerateExec` | ✅     | Supports `explode`, `explode_outer`, `posexplode`, `posexplode_outer` over arrays, `inline`, `inline_outer`, and `stack`. Other deterministic generators, including Hive UDTFs, run on the JVM inside the operator; see [Generators](scala_java_udfs.md#generators). |
+| `ExpandExec`   | ✅     |                                                                                                                                                                                                                                                                      |
+| `UnionExec`    | ✅     |                                                                                                                                                                                                                                                                      |
+| `CoalesceExec` | ✅     |                                                                                                                                                                                                                                                                      |
 
 ## Writes
 

docs/source/user-guide/latest/scala_java_udfs.md

@@ -21,7 +21,7 @@
 
 Comet executes Spark's Scala and Java [scalar user-defined functions (UDFs)](https://spark.apache.org/docs/latest/sql-ref-functions-udf-scalar.html) within the Comet pipeline. The presence of a UDF does not force the enclosing operator out of the Comet pipeline; surrounding Rust-implemented operators stay in the pipeline.
 
-This page covers Spark's `ScalaUDF` (Scala `udf(...)`, `spark.udf.register(...)` over Scala or Java functional interfaces, and SQL `CREATE FUNCTION ... AS 'com.example.MyUDF'`). Other UDF kinds (Python / Pandas, Hive, aggregate) are out of scope and continue to fall back to Spark.
+This page covers Spark's `ScalaUDF` (Scala `udf(...)`, `spark.udf.register(...)` over Scala or Java functional interfaces, and SQL `CREATE FUNCTION ... AS 'com.example.MyUDF'`). Other UDF kinds (Python / Pandas, Hive scalar UDFs, aggregate) are out of scope and continue to fall back to Spark. Hive UDTFs and other generators are covered in [Generators](#generators).
 
 This feature is enabled by default. Set `spark.comet.exec.scalaUDF.codegen.enabled` to `false` to route plans containing a `ScalaUDF` back to Spark for the enclosing operator.
 
@@ -42,8 +42,15 @@ This feature is enabled by default. Set `spark.comet.exec.scalaUDF.codegen.enabl
 ## Not supported
 
 - Aggregate UDFs (`ScalaAggregator`, `TypedImperativeAggregate`, the legacy `UserDefinedAggregateFunction`).
-- Table UDFs and generators.
+- Python UDTFs (`@udtf`).
 - Python `@udf` and Pandas `@pandas_udf`.
+
+## Generators
+
+A deterministic generator that Comet has no native implementation of, such as a Hive `GenericUDTF` registered with `CREATE FUNCTION` or Spark's `json_tuple`, runs inside Comet's native generate operator. The operator calls the generator on the JVM for each batch and keeps one generator instance per partition, so state the generator holds carries across batches. Once a partition's input ends, the rows of the generator's `terminate()` (a Hive UDTF's `close()`) follow, carrying the partition's last input row, as in Spark.
+
+Set `spark.comet.exec.jvmGenerator.enabled` to `false` to run such a generator in Spark instead.
+
 - Hive `GenericUDF` and `SimpleUDF`.
 - `NullType` and `UserDefinedType` arguments and return types. UDT-typed columns fall back to Spark; to keep execution in the Comet pipeline, store and read the underlying representation directly (e.g. write MLlib `Vector` outputs as `Struct<type: Byte, size: Int, indices: Array<Int>, values: Array<Double>>` rather than `VectorUDT`).
 - Trees whose total nested-field count (output plus all input columns the UDF tree references) exceeds `spark.sql.codegen.maxFields` (default 100). Comet refuses these at plan time and the operator falls back to Spark.

docs/source/user-guide/latest/understanding-comet-plans.md

@@ -310,28 +310,28 @@ consecutively in a plan, they execute as a single fused block. The write operato
 (`CometWriteFiles`, `CometNativeWrite` and `CometIcebergWrite`) are the exception: each runs as its
 own native plan over its child's output.
 
-| Node                           | Spark equivalent                                                                  |
-| ------------------------------ | --------------------------------------------------------------------------------- |
-| `CometProject`                 | `ProjectExec`                                                                     |
-| `CometFilter`                  | `FilterExec`                                                                      |
-| `CometSort`                    | `SortExec`                                                                        |
-| `CometLocalLimit`              | `LocalLimitExec`                                                                  |
-| `CometGlobalLimit`             | `GlobalLimitExec`                                                                 |
-| `CometExpand`                  | `ExpandExec`                                                                      |
-| `CometExplode`                 | `GenerateExec` (for `explode`, `explode_outer`, `posexplode`, `posexplode_outer`) |
-| `CometHashAggregate`           | `HashAggregateExec`, `ObjectHashAggregateExec`                                    |
-| `CometHashJoin`                | `ShuffledHashJoinExec`                                                            |
-| `CometBroadcastHashJoin`       | `BroadcastHashJoinExec`                                                           |
-| `CometBroadcastNestedLoopJoin` | `BroadcastNestedLoopJoinExec`                                                     |
-| `CometSortMergeJoin`           | `SortMergeJoinExec`                                                               |
-| `CometMergeRows`               | `MergeRowsExec` (Spark 3.5+; opt-in)                                              |
-| `CometWindowExec`              | `WindowExec`                                                                      |
-| `CometWindowGroupLimitExec`    | `WindowGroupLimitExec` (Spark 3.5 and later)                                      |
-| `CometSample`                  | `SampleExec` (sampling without replacement)                                       |
-| `CometTakeOrderedAndProject`   | `TakeOrderedAndProjectExec`                                                       |
-| `CometWriteFiles`              | `WriteFilesExec` (Spark 4.0 and later, experimental native Parquet writes)        |
-| `CometNativeWrite`             | `DataWritingCommandExec` (Spark 3.x, experimental native Parquet writes)          |
-| `CometIcebergWrite`            | `IcebergWrite` (native Iceberg data-file writes, opt-in)                          |
+| Node                           | Spark equivalent                                                                    |
+| ------------------------------ | ----------------------------------------------------------------------------------- |
+| `CometProject`                 | `ProjectExec`                                                                       |
+| `CometFilter`                  | `FilterExec`                                                                        |
+| `CometSort`                    | `SortExec`                                                                          |
+| `CometLocalLimit`              | `LocalLimitExec`                                                                    |
+| `CometGlobalLimit`             | `GlobalLimitExec`                                                                   |
+| `CometExpand`                  | `ExpandExec`                                                                        |
+| `CometExplode`                 | `GenerateExec`; generators without a native implementation run on the JVM inside it |
+| `CometHashAggregate`           | `HashAggregateExec`, `ObjectHashAggregateExec`                                      |
+| `CometHashJoin`                | `ShuffledHashJoinExec`                                                              |
+| `CometBroadcastHashJoin`       | `BroadcastHashJoinExec`                                                             |
+| `CometBroadcastNestedLoopJoin` | `BroadcastNestedLoopJoinExec`                                                       |
+| `CometSortMergeJoin`           | `SortMergeJoinExec`                                                                 |
+| `CometMergeRows`               | `MergeRowsExec` (Spark 3.5+; opt-in)                                                |
+| `CometWindowExec`              | `WindowExec`                                                                        |
+| `CometWindowGroupLimitExec`    | `WindowGroupLimitExec` (Spark 3.5 and later)                                        |
+| `CometSample`                  | `SampleExec` (sampling without replacement)                                         |
+| `CometTakeOrderedAndProject`   | `TakeOrderedAndProjectExec`                                                         |
+| `CometWriteFiles`              | `WriteFilesExec` (Spark 4.0 and later, experimental native Parquet writes)          |
+| `CometNativeWrite`             | `DataWritingCommandExec` (Spark 3.x, experimental native Parquet writes)            |
+| `CometIcebergWrite`            | `IcebergWrite` (native Iceberg data-file writes, opt-in)                            |
 
 ### JVM-Side Operators
 

native/core/src/execution/planner.rs

@@ -5088,6 +5088,8 @@ mod tests {
                             outer,
                             position,
                             project_list: vec![bound],
+                            inline: false,
+                            terminate: None,
                         })),
                         ..Default::default()
                     };

native/core/src/execution/planner/explode.rs

@@ -15,8 +15,10 @@
 // specific language governing permissions and limitations
 // under the License.
 
-//! Builder for the native `Explode` operator, which plans Spark's `explode`, `posexplode`, and
-//! their `_outer` variants as a projection followed by Comet's `ExplodeExec`.
+//! Builder for the native `Explode` operator, which plans Spark's `explode`, `posexplode`,
+//! `inline`, and their `_outer` variants as a projection followed by Comet's `ExplodeExec`.
+//! Spark's `stack` arrives here as `inline` over an array of structs built by the JVM serde, and
+//! any other generator as `inline` over the rows a JVM `CometUDF` produces for each input row.
 
 use std::sync::Arc;
 
@@ -162,11 +164,30 @@ impl OperatorBuilder for ExplodeBuilder {
             }
         };
 
-        output_fields.push(Field::new(
-            array_field.name(),
-            element_type,
-            true, // Element is nullable after unnesting
-        ));
+        // `inline` flattens each struct element into one column per field. Every field is
+        // nullable after unnesting: a null struct element, or the row `inline_outer` emits for a
+        // NULL or empty array, yields a row of nulls.
+        let struct_column_indices = if explode.inline {
+            let DataType::Struct(struct_fields) = &element_type else {
+                return Err(ExecutionError::GeneralError(format!(
+                    "Expected List of Struct for inline, got {:?}",
+                    array_field.data_type()
+                )));
+            };
+            output_fields.extend(
+                struct_fields
+                    .iter()
+                    .map(|f| f.as_ref().clone().with_nullable(true)),
+            );
+            vec![array_input_index]
+        } else {
+            output_fields.push(Field::new(
+                array_field.name(),
+                element_type,
+                true, // Element is nullable after unnesting
+            ));
+            vec![]
+        };
 
         let output_schema = Arc::new(Schema::new(output_fields));
 
@@ -193,13 +214,18 @@ impl OperatorBuilder for ExplodeBuilder {
         let unnest_options = UnnestOptions::new().with_null_handling(null_handling);
 
         // Comet's specialized fork of `UnnestExec`; see `operators::explode`.
-        let unnest_exec = Arc::new(ExplodeExec::new(
+        let mut unnest_exec = ExplodeExec::new(
             project_exec,
             list_unnests,
-            vec![], // No struct columns to unnest
+            struct_column_indices,
             output_schema,
             unnest_options,
-        )?);
+        )?;
+        if let Some(terminate) = &explode.terminate {
+            let terminate = planner.create_expr(terminate, Arc::new(Schema::empty()))?;
+            unnest_exec = unnest_exec.with_terminate(terminate)?;
+        }
+        let unnest_exec = Arc::new(unnest_exec);
 
         Ok((
             scans,