feat: support explode_outer

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

@@ -27,14 +27,14 @@
 
 ## explode_outer
 
-- Same `CometExplodeExec` path as `explode`, but the `outer=true` case is `Incompatible` (empty arrays are not preserved as null outputs) and falls back unless `spark.comet.operator.GenerateExec.allowIncompatible=true` ([datafusion#19053](https://github.com/apache/datafusion/issues/19053)).
+- Same `CometExplodeExec` path as `explode`. Compatible for array inputs; empty and NULL arrays both emit one null-valued row per Spark's `outer` semantics via the `ListEmptyToNullExpr` planner bridge (works around [datafusion#19053](https://github.com/apache/datafusion/issues/19053)). Map inputs fall back.
 
 ## posexplode
 
 - Handled at the operator level as a `GenerateExec` (`CometExplodeExec`), like `explode`. Compatible for array inputs; map inputs fall back ([#2837](https://github.com/apache/datafusion-comet/issues/2837)).
 
 ## posexplode_outer
 
-- Same `CometExplodeExec` path as `posexplode`, but the `outer=true` case is `Incompatible` and falls back unless `spark.comet.operator.GenerateExec.allowIncompatible=true` ([datafusion#19053](https://github.com/apache/datafusion/issues/19053)).
+- 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 via the `ListEmptyToNullExpr` planner bridge (works around [datafusion#19053](https://github.com/apache/datafusion/issues/19053)).
 
 [Spark Expression Support]: ../../user-guide/latest/expressions.md

docs/source/contributor-guide/native_shuffle.md

@@ -49,7 +49,6 @@ Native shuffle (`CometExchange`) is selected when all of the following condition
    columnar output. Row-based Spark operators require JVM shuffle.
 
 3. **Supported partitioning type**: Native shuffle supports:
-
    - `HashPartitioning`
    - `RangePartitioning`
    - `SinglePartition`
@@ -129,7 +128,6 @@ Native shuffle (`CometExchange`) is selected when all of the following condition
 1. **Plan construction**: `CometNativeShuffleWriter` builds a protobuf operator tree with a
    `ShuffleWriter` operator at the root and `childNativeOp` as its child. `childNativeOp` takes
    one of two shapes:
-
    - The child plan's `nativeOp` directly, when `CometShuffleExchangeExec`'s child is a
      `CometNativeExec` subtree. The upstream operators run inside the same `CometExecIterator`
      as the writer, with no JVM-to-native batch boundary between them.
@@ -142,21 +140,18 @@ Native shuffle (`CometExchange`) is selected when all of the following condition
 2. **Native execution**: A single `CometExecIterator` per partition runs the unified plan.
 
 3. **Partitioning**: `ShuffleWriterExec` receives batches and routes to the appropriate partitioner:
-
    - `MultiPartitionShuffleRepartitioner`: For hash/range/round-robin partitioning
    - `SinglePartitionShufflePartitioner`: For single partition (simpler path)
 
 4. **Buffering and spilling**: The partitioner buffers rows per partition. When memory pressure
    exceeds the threshold, partitions spill to temporary files.
 
 5. **Encoding**: `ShuffleBlockWriter` encodes each partition's data as compressed Arrow IPC:
-
    - Writes compression type header
    - Writes field count header
    - Writes compressed IPC stream
 
 6. **Output files**: Two files are produced:
-
    - **Data file**: Concatenated partition data
    - **Index file**: Array of 8-byte little-endian offsets marking partition boundaries
 
@@ -168,7 +163,6 @@ Native shuffle (`CometExchange`) is selected when all of the following condition
 1. `CometBlockStoreShuffleReader` fetches shuffle blocks via `ShuffleBlockFetcherIterator`.
 
 2. For each block, `NativeBatchDecoderIterator`:
-
    - Reads the 8-byte compressed length header
    - Reads the 8-byte field count header
    - Reads the compressed IPC data

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

@@ -325,18 +325,18 @@ The type-name conversion functions (`bigint`, `binary`, `boolean`, `date`, `deci
 
 ## generator_funcs
 
-`explode` and `posexplode` are supported via `CometExplodeExec` (operator-level, not
-expression-level). The `outer` variants are wired but marked `Incompatible`; they require
-`spark.comet.exec.explode.enabled=true` and `allowIncompatible`.
+`explode`, `explode_outer`, `posexplode`, and `posexplode_outer` are supported via
+`CometExplodeExec` (operator-level, not expression-level). Requires
+`spark.comet.exec.explode.enabled=true`.
 
 | Function | Status | Implementation | Notes |
 | --- | --- | --- | --- |
 | `explode` | ✅ | — | via `CometExplodeExec` |
-| `explode_outer` | ✅ | — | outer=true falls back (Incompatible) ([audit](../../contributor-guide/expression-audits/generator_funcs.md#explode_outer)) |
+| `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`) |
 | `posexplode` | ✅ | — | via `CometExplodeExec` |
-| `posexplode_outer` | ✅ | — | outer=true falls back (Incompatible) ([audit](../../contributor-guide/expression-audits/generator_funcs.md#posexplode_outer)) |
+| `posexplode_outer` | ✅ | — | via `CometExplodeExec` ([audit](../../contributor-guide/expression-audits/generator_funcs.md#posexplode_outer)) |
 | `stack` | 🔜 | — | Operator-level generator |
 
 ---

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

@@ -107,12 +107,12 @@ omitted from the tables below and may be reconsidered based on demand:
 
 ## Generators and set operations
 
-| Operator       | Status | Notes                                                                                                                      |
-| -------------- | ------ | -------------------------------------------------------------------------------------------------------------------------- |
-| `GenerateExec` | ✅     | Supports `explode` and `posexplode` over arrays. The `_outer` variants are incompatible, and `inline` / `stack` fall back. |
-| `ExpandExec`   | ✅     |                                                                                                                            |
-| `UnionExec`    | ✅     |                                                                                                                            |
-| `CoalesceExec` | ✅     |                                                                                                                            |
+| Operator       | Status | Notes                                                                                                            |
+| -------------- | ------ | ---------------------------------------------------------------------------------------------------------------- |
+| `GenerateExec` | ✅     | Supports `explode`, `explode_outer`, `posexplode`, `posexplode_outer` over arrays. `inline` / `stack` fall back. |
+| `ExpandExec`   | ✅     |                                                                                                                  |
+| `UnionExec`    | ✅     |                                                                                                                  |
+| `CoalesceExec` | ✅     |                                                                                                                  |
 
 ## Writes
 

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

@@ -246,22 +246,22 @@ by role. Names match what is shown in the plan output.
 These are implemented in Rust and run in DataFusion. When several appear
 consecutively in a plan, they execute as a single fused block.
 
-| Node                           | Spark equivalent                                |
-| ------------------------------ | ----------------------------------------------- |
-| `CometProject`                 | `ProjectExec`                                   |
-| `CometFilter`                  | `FilterExec`                                    |
-| `CometSort`                    | `SortExec`                                      |
-| `CometLocalLimit`              | `LocalLimitExec`                                |
-| `CometGlobalLimit`             | `GlobalLimitExec`                               |
-| `CometExpand`                  | `ExpandExec`                                    |
-| `CometExplode`                 | `GenerateExec` (for `explode` and `posexplode`) |
-| `CometHashAggregate`           | `HashAggregateExec`, `ObjectHashAggregateExec`  |
-| `CometHashJoin`                | `ShuffledHashJoinExec`                          |
-| `CometBroadcastHashJoin`       | `BroadcastHashJoinExec`                         |
-| `CometBroadcastNestedLoopJoin` | `BroadcastNestedLoopJoinExec`                   |
-| `CometSortMergeJoin`           | `SortMergeJoinExec`                             |
-| `CometWindow`                  | `WindowExec`                                    |
-| `CometTakeOrderedAndProject`   | `TakeOrderedAndProjectExec`                     |
+| 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`                                                               |
+| `CometWindow`                  | `WindowExec`                                                                      |
+| `CometTakeOrderedAndProject`   | `TakeOrderedAndProjectExec`                                                       |
 
 ### JVM-Side Operators
 

native/core/src/execution/expressions/list_empty_to_null.rs

@@ -0,0 +1,146 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::fmt::{Display, Formatter};
+use std::hash::{Hash, Hasher};
+use std::sync::Arc;
+
+use arrow::array::{Array, ArrayRef, ListArray, RecordBatch};
+use arrow::buffer::{BooleanBuffer, NullBuffer};
+use arrow::datatypes::{DataType, Field, FieldRef, Schema};
+use datafusion::common::{exec_err, Result as DataFusionResult};
+use datafusion::physical_expr::PhysicalExpr;
+use datafusion::physical_plan::ColumnarValue;
+
+/// A `PhysicalExpr` that marks every empty row of a `List<T>` input as null.
+/// Bridges DataFusion's `UnnestExec` (which drops empty rows under
+/// `preserve_nulls=true`) to Spark's `explode_outer`/`posexplode_outer`
+/// semantics. See <https://github.com/apache/datafusion/issues/19053>.
+#[derive(Debug, Clone)]
+pub struct ListEmptyToNullExpr {
+    child: Arc<dyn PhysicalExpr>,
+}
+
+impl ListEmptyToNullExpr {
+    pub fn new(child: Arc<dyn PhysicalExpr>) -> Self {
+        Self { child }
+    }
+}
+
+impl Display for ListEmptyToNullExpr {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        write!(f, "list_empty_to_null({})", self.child)
+    }
+}
+
+impl PartialEq for ListEmptyToNullExpr {
+    fn eq(&self, other: &Self) -> bool {
+        self.child.eq(&other.child)
+    }
+}
+
+impl Eq for ListEmptyToNullExpr {}
+
+impl Hash for ListEmptyToNullExpr {
+    fn hash<H: Hasher>(&self, state: &mut H) {
+        self.child.hash(state);
+    }
+}
+
+impl PhysicalExpr for ListEmptyToNullExpr {
+    fn fmt_sql(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        Display::fmt(self, f)
+    }
+
+    fn return_field(&self, input_schema: &Schema) -> DataFusionResult<FieldRef> {
+        // Preserve the child field's name and element type; force the outer
+        // list to nullable because we mark empty rows as null.
+        let child_field = self.child.return_field(input_schema)?;
+        Ok(Arc::new(Field::new(
+            child_field.name(),
+            child_field.data_type().clone(),
+            true,
+        )))
+    }
+
+    fn evaluate(&self, batch: &RecordBatch) -> DataFusionResult<ColumnarValue> {
+        let value = self.child.evaluate(batch)?;
+        let array = value.into_array(batch.num_rows())?;
+
+        let Some(list) = array.as_any().downcast_ref::<ListArray>() else {
+            return exec_err!(
+                "ListEmptyToNullExpr expected List input, got {}",
+                array.data_type()
+            );
+        };
+
+        let offsets = list.offsets();
+        let len = list.len();
+        let existing_nulls = list.nulls();
+
+        // Fast path: no currently-valid row is empty, so the input already
+        // satisfies outer semantics. `is_valid` returns true when `nulls` is
+        // `None`, so this single scan short-circuits on the first empty
+        // valid row without allocating.
+        let has_valid_empty = (0..len)
+            .any(|i| offsets[i + 1] == offsets[i] && existing_nulls.is_none_or(|n| n.is_valid(i)));
+        if !has_valid_empty {
+            return Ok(ColumnarValue::Array(Arc::clone(&array)));
+        }
+
+        let non_empty = BooleanBuffer::collect_bool(len, |i| offsets[i + 1] > offsets[i]);
+        let new_nulls = match existing_nulls {
+            None => NullBuffer::new(non_empty),
+            Some(existing) => {
+                let combined = existing.inner() & &non_empty;
+                let null_count = len - combined.count_set_bits();
+                // SAFETY: null_count was just derived as len - popcount(combined).
+                unsafe { NullBuffer::new_unchecked(combined, null_count) }
+            }
+        };
+
+        let DataType::List(element_field) = list.data_type() else {
+            unreachable!("ListArray downcast guarantees DataType::List");
+        };
+
+        let result = ListArray::try_new(
+            Arc::clone(element_field),
+            offsets.clone(),
+            Arc::clone(list.values()),
+            Some(new_nulls),
+        )?;
+
+        Ok(ColumnarValue::Array(Arc::new(result) as ArrayRef))
+    }
+
+    fn children(&self) -> Vec<&Arc<dyn PhysicalExpr>> {
+        vec![&self.child]
+    }
+
+    fn with_new_children(
+        self: Arc<Self>,
+        children: Vec<Arc<dyn PhysicalExpr>>,
+    ) -> DataFusionResult<Arc<dyn PhysicalExpr>> {
+        if children.len() != 1 {
+            return exec_err!(
+                "ListEmptyToNullExpr expects exactly 1 child, got {}",
+                children.len()
+            );
+        }
+        Ok(Arc::new(ListEmptyToNullExpr::new(Arc::clone(&children[0]))))
+    }
+}

native/core/src/execution/expressions/mod.rs

@@ -20,6 +20,7 @@
 pub mod arithmetic;
 pub mod bitwise;
 pub mod comparison;
+pub mod list_empty_to_null;
 pub mod list_positions;
 pub mod logical;
 pub mod nullcheck;