You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by ag...@apache.org on 2026/08/04 17:37:32 UTC
(datafusion-comet) branch main updated: feat: support `explode_outer` (#5192)
This is an automated email from the ASF dual-hosted git repository.
andygrove pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
The following commit(s) were added to refs/heads/main by this push:
new 2af9cec38b feat: support `explode_outer` (#5192)
2af9cec38b is described below
commit 2af9cec38b1f68550c3e13c1109f3c7502081e84
Author: Oleks V <co...@users.noreply.github.com>
AuthorDate: Tue Aug 4 10:37:27 2026 -0700
feat: support `explode_outer` (#5192)
---
.../expression-audits/generator_funcs.md | 4 +-
docs/source/user-guide/latest/expressions.md | 10 +-
docs/source/user-guide/latest/operators.md | 12 +-
.../user-guide/latest/understanding-comet-plans.md | 32 +-
.../execution/expressions/list_empty_to_null.rs | 311 +++++++++++++
.../src/execution/expressions/list_positions.rs | 172 ++++++-
native/core/src/execution/expressions/mod.rs | 1 +
native/core/src/execution/planner.rs | 80 +++-
.../org/apache/spark/sql/comet/operators.scala | 7 +-
.../sql-tests/expressions/array/explode.sql | 500 +++++++++++++++++++++
.../sql-tests/expressions/array/posexplode.sql | 213 ++++++++-
.../apache/comet/exec/CometGenerateExecSuite.scala | 211 ++++++++-
12 files changed, 1481 insertions(+), 72 deletions(-)
diff --git a/docs/source/contributor-guide/expression-audits/generator_funcs.md b/docs/source/contributor-guide/expression-audits/generator_funcs.md
index 1b7c2dca03..4d160c6ac3 100644
--- a/docs/source/contributor-guide/expression-audits/generator_funcs.md
+++ b/docs/source/contributor-guide/expression-audits/generator_funcs.md
@@ -27,7 +27,7 @@
## 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
@@ -35,6 +35,6 @@
## 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
diff --git a/docs/source/user-guide/latest/expressions.md b/docs/source/user-guide/latest/expressions.md
index 70efdf8559..189033f3f7 100644
--- a/docs/source/user-guide/latest/expressions.md
+++ b/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). Enabled by default via
+`spark.comet.exec.explode.enabled`.
| 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 |
---
diff --git a/docs/source/user-guide/latest/operators.md b/docs/source/user-guide/latest/operators.md
index 76ee09cc12..42235b1367 100644
--- a/docs/source/user-guide/latest/operators.md
+++ b/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
diff --git a/docs/source/user-guide/latest/understanding-comet-plans.md b/docs/source/user-guide/latest/understanding-comet-plans.md
index ea32e37b8f..a600615b4b 100644
--- a/docs/source/user-guide/latest/understanding-comet-plans.md
+++ b/docs/source/user-guide/latest/understanding-comet-plans.md
@@ -282,22 +282,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
diff --git a/native/core/src/execution/expressions/list_empty_to_null.rs b/native/core/src/execution/expressions/list_empty_to_null.rs
new file mode 100644
index 0000000000..ad6d9452ac
--- /dev/null
+++ b/native/core/src/execution/expressions/list_empty_to_null.rs
@@ -0,0 +1,311 @@
+// 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 combined = BooleanBuffer::collect_bool(len, |i| {
+ offsets[i + 1] > offsets[i] && existing_nulls.is_none_or(|n| n.is_valid(i))
+ });
+ let new_nulls = NullBuffer::new(combined);
+
+ 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]))))
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use arrow::array::{Int32Array, ListArray};
+ use arrow::buffer::{NullBuffer, OffsetBuffer};
+ use arrow::datatypes::{DataType, Field, Schema};
+ use datafusion::physical_expr::expressions::Column;
+
+ fn element_field() -> Arc<Field> {
+ Arc::new(Field::new("item", DataType::Int32, true))
+ }
+
+ fn list_field(nullable: bool) -> Arc<Field> {
+ Arc::new(Field::new("arr", DataType::List(element_field()), nullable))
+ }
+
+ fn build_batch(list: ArrayRef) -> RecordBatch {
+ let schema = Schema::new(vec![Field::new("arr", list.data_type().clone(), true)]);
+ RecordBatch::try_new(Arc::new(schema), vec![list]).unwrap()
+ }
+
+ fn evaluate_ref(list: ArrayRef) -> ArrayRef {
+ let batch = build_batch(list);
+ let expr = ListEmptyToNullExpr::new(Arc::new(Column::new("arr", 0)));
+ let result = expr.evaluate(&batch).unwrap();
+ result.into_array(batch.num_rows()).unwrap()
+ }
+
+ fn evaluate(list: ListArray) -> ListArray {
+ let array: ArrayRef = Arc::new(list);
+ let out = evaluate_ref(array);
+ out.as_any().downcast_ref::<ListArray>().unwrap().clone()
+ }
+
+ #[test]
+ fn fast_path_no_empty_rows_returns_input_untouched() {
+ // Rows: [1,2,3], [4], [5,6] -- no empty rows, so the fast path should
+ // return the original array pointer without allocating a new bitmap.
+ let values = Int32Array::from(vec![1, 2, 3, 4, 5, 6]);
+ let offsets = OffsetBuffer::new(vec![0, 3, 4, 6].into());
+ let input: ArrayRef = Arc::new(ListArray::new(
+ element_field(),
+ offsets,
+ Arc::new(values),
+ None,
+ ));
+
+ let out = evaluate_ref(Arc::clone(&input));
+ assert!(
+ Arc::ptr_eq(&out, &input),
+ "fast path should return the same ArrayRef"
+ );
+ let list = out.as_any().downcast_ref::<ListArray>().unwrap();
+ assert!(list.nulls().is_none());
+ assert_eq!(list.len(), 3);
+ }
+
+ #[test]
+ fn fast_path_when_only_empty_row_is_already_null() {
+ // Rows: [1,2], NULL (offsets 2..2 -- looks empty), [3]. The middle row
+ // is already null so the fast path applies without materializing a new
+ // bitmap.
+ let values = Int32Array::from(vec![1, 2, 3]);
+ let offsets = OffsetBuffer::new(vec![0, 2, 2, 3].into());
+ let nulls = NullBuffer::from(vec![true, false, true]);
+ let input: ArrayRef = Arc::new(ListArray::new(
+ element_field(),
+ offsets,
+ Arc::new(values),
+ Some(nulls),
+ ));
+
+ let out = evaluate_ref(Arc::clone(&input));
+ assert!(
+ Arc::ptr_eq(&out, &input),
+ "fast path should return the same ArrayRef when only empty rows are already null"
+ );
+ }
+
+ #[test]
+ fn mixed_empty_null_and_non_empty_rows() {
+ // Rows: [10, 20], [], NULL, [30]. The empty row (index 1) must become
+ // null; the already-null row (index 2) must stay null; the two data
+ // rows must survive intact.
+ let values = Int32Array::from(vec![10, 20, 30]);
+ let offsets = OffsetBuffer::new(vec![0, 2, 2, 2, 3].into());
+ let nulls = NullBuffer::from(vec![true, true, false, true]);
+ let input = ListArray::new(element_field(), offsets, Arc::new(values), Some(nulls));
+ let output = evaluate(input);
+ let out_nulls = output.nulls().expect("nulls buffer must be present");
+ assert!(out_nulls.is_valid(0));
+ assert!(!out_nulls.is_valid(1), "empty row must be marked null");
+ assert!(!out_nulls.is_valid(2), "already-null row must stay null");
+ assert!(out_nulls.is_valid(3));
+ // Offsets and values must be preserved so downstream unnest still
+ // reads the original element slices for the valid rows.
+ assert_eq!(output.value(0).len(), 2);
+ assert_eq!(output.value(3).len(), 1);
+ }
+
+ #[test]
+ fn empty_row_without_prior_null_bitmap() {
+ // Fresh (no nulls) input containing one empty row must materialize a
+ // null bitmap with only that row cleared.
+ let values = Int32Array::from(vec![1, 2, 3]);
+ let offsets = OffsetBuffer::new(vec![0, 2, 2, 3].into());
+ let input = ListArray::new(element_field(), offsets, Arc::new(values), None);
+ let output = evaluate(input);
+ let out_nulls = output.nulls().expect("nulls buffer must be materialized");
+ assert!(out_nulls.is_valid(0));
+ assert!(!out_nulls.is_valid(1));
+ assert!(out_nulls.is_valid(2));
+ }
+
+ #[test]
+ fn zero_row_batch_takes_fast_path() {
+ // A zero-row batch has no rows to inspect, so the fast path returns
+ // the input untouched.
+ let values = Int32Array::from(Vec::<i32>::new());
+ let offsets = OffsetBuffer::new(vec![0].into());
+ let input: ArrayRef = Arc::new(ListArray::new(
+ element_field(),
+ offsets,
+ Arc::new(values),
+ None,
+ ));
+
+ let out = evaluate_ref(Arc::clone(&input));
+ assert_eq!(out.len(), 0);
+ assert!(
+ Arc::ptr_eq(&out, &input),
+ "zero-row batch should take the fast path"
+ );
+ }
+
+ #[test]
+ fn sliced_input_with_non_zero_offset() {
+ // Build a 5-row list, then slice out rows 1..4 -- exposes non-zero
+ // logical offset while `values()` stays unsliced. The empty row that
+ // now lives at logical index 1 (was index 2 in the underlying array)
+ // must be flipped to null.
+ let values = Int32Array::from(vec![1, 2, 3, 4, 5, 6]);
+ let offsets = OffsetBuffer::new(vec![0, 2, 3, 3, 5, 6].into());
+ let input = ListArray::new(element_field(), offsets, Arc::new(values), None);
+ let sliced = input.slice(1, 3);
+ let output = evaluate(sliced);
+ assert_eq!(output.len(), 3);
+ let out_nulls = output.nulls().expect("nulls buffer must be materialized");
+ // Logical row 0: was [3] -- non-empty
+ assert!(out_nulls.is_valid(0));
+ // Logical row 1: was [] -- must be null
+ assert!(!out_nulls.is_valid(1));
+ // Logical row 2: was [4, 5] -- non-empty
+ assert!(out_nulls.is_valid(2));
+ // The non-empty rows must still expose their original elements.
+ assert_eq!(output.value(0).len(), 1);
+ assert_eq!(output.value(2).len(), 2);
+ }
+
+ #[test]
+ fn return_field_forces_nullable() {
+ // The output field must be nullable regardless of the child's
+ // nullability, because empty rows are marked null downstream.
+ let schema = Schema::new(vec![list_field(false).as_ref().clone()]);
+ let expr = ListEmptyToNullExpr::new(Arc::new(Column::new("arr", 0)));
+ let field = expr.return_field(&schema).unwrap();
+ assert!(field.is_nullable());
+ assert_eq!(field.name(), "arr");
+ }
+}
diff --git a/native/core/src/execution/expressions/list_positions.rs b/native/core/src/execution/expressions/list_positions.rs
index 11643db4ce..083f88a523 100644
--- a/native/core/src/execution/expressions/list_positions.rs
+++ b/native/core/src/execution/expressions/list_positions.rs
@@ -20,15 +20,20 @@ use std::hash::{Hash, Hasher};
use std::sync::Arc;
use arrow::array::{Array, ArrayRef, Int32Array, ListArray, RecordBatch};
+use arrow::buffer::OffsetBuffer;
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 takes a `List<T>` input and produces a `List<Int32>` where each row's
-/// values are `[0, 1, ..., len - 1]`. Offsets and the null bitmap are inherited from the input,
-/// so when the resulting list is unnested in parallel with the original list it produces the
-/// `pos` column expected by Spark's `posexplode`.
+/// values are `[0, 1, ..., len - 1]`. Per-row lengths and the null bitmap are inherited from the
+/// input, so when the resulting list is unnested in parallel with the original list it produces
+/// the `pos` column expected by Spark's `posexplode`.
+///
+/// Offsets are rebased to zero so a sliced input (whose `offsets.first()` may be non-zero while
+/// `values()` stays unsliced) lines up with the freshly built values array. See
+/// <https://github.com/apache/datafusion-comet/issues/5224>.
#[derive(Debug, Clone)]
pub struct ListPositionsExpr {
child: Arc<dyn PhysicalExpr>,
@@ -94,21 +99,31 @@ impl PhysicalExpr for ListPositionsExpr {
};
let offsets = list.offsets();
- let total_len = *offsets.last().unwrap() as usize;
+ // `ListArray::slice` slices `value_offsets` and `nulls` but leaves `values` unsliced, so
+ // a sliced input has `offsets.first() > 0` while the underlying values run to the full
+ // original length. We build a fresh values array of length
+ // `offsets.last() - offsets.first()`, so we must rebase offsets to zero to line them up.
+ // See https://github.com/apache/datafusion-comet/issues/5224.
+ let base = offsets.first().copied().unwrap_or(0);
+ let total_len = (*offsets.last().unwrap() - base) as usize;
let mut values: Vec<i32> = Vec::with_capacity(total_len);
for window in offsets.windows(2) {
let start = window[0];
let end = window[1];
- for i in 0..(end - start) {
- values.push(i);
- }
+ values.extend(0..(end - start));
}
+ let rebased_offsets = if base == 0 {
+ offsets.clone()
+ } else {
+ OffsetBuffer::new(offsets.iter().map(|o| o - base).collect::<Vec<_>>().into())
+ };
+
let element_field = Arc::new(Field::new("item", DataType::Int32, true));
let result = ListArray::new(
element_field,
- offsets.clone(),
+ rebased_offsets,
Arc::new(Int32Array::from(values)),
list.nulls().cloned(),
);
@@ -133,3 +148,144 @@ impl PhysicalExpr for ListPositionsExpr {
Ok(Arc::new(ListPositionsExpr::new(Arc::clone(&children[0]))))
}
}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use arrow::array::{Int32Array, ListArray};
+ use arrow::buffer::{NullBuffer, OffsetBuffer};
+ use arrow::datatypes::{DataType, Field, Schema};
+ use datafusion::physical_expr::expressions::Column;
+
+ fn element_field() -> Arc<Field> {
+ Arc::new(Field::new("item", DataType::Int32, true))
+ }
+
+ fn evaluate_ref(list: ArrayRef) -> ArrayRef {
+ let schema = Schema::new(vec![Field::new("arr", list.data_type().clone(), true)]);
+ let batch = RecordBatch::try_new(Arc::new(schema), vec![list]).unwrap();
+ let expr = ListPositionsExpr::new(Arc::new(Column::new("arr", 0)));
+ let result = expr.evaluate(&batch).unwrap();
+ result.into_array(batch.num_rows()).unwrap()
+ }
+
+ fn evaluate(list: ListArray) -> ListArray {
+ let array: ArrayRef = Arc::new(list);
+ let out = evaluate_ref(array);
+ out.as_any().downcast_ref::<ListArray>().unwrap().clone()
+ }
+
+ fn row_values(list: &ListArray, row: usize) -> Vec<i32> {
+ let values = list.value(row);
+ let ints = values.as_any().downcast_ref::<Int32Array>().unwrap();
+ (0..ints.len()).map(|i| ints.value(i)).collect()
+ }
+
+ #[test]
+ fn non_sliced_input_produces_position_per_row() {
+ // Rows: [10, 20, 30], [40], [50, 60]. Positions must be [0,1,2],[0],[0,1].
+ let values = Int32Array::from(vec![10, 20, 30, 40, 50, 60]);
+ let offsets = OffsetBuffer::new(vec![0, 3, 4, 6].into());
+ let input = ListArray::new(element_field(), offsets, Arc::new(values), None);
+ let output = evaluate(input);
+
+ assert_eq!(output.len(), 3);
+ assert!(output.nulls().is_none());
+ assert_eq!(row_values(&output, 0), vec![0, 1, 2]);
+ assert_eq!(row_values(&output, 1), vec![0]);
+ assert_eq!(row_values(&output, 2), vec![0, 1]);
+ }
+
+ #[test]
+ fn empty_rows_produce_empty_position_rows() {
+ // Rows: [], [10], []. Positions must be [], [0], [].
+ let values = Int32Array::from(vec![10]);
+ let offsets = OffsetBuffer::new(vec![0, 0, 1, 1].into());
+ let input = ListArray::new(element_field(), offsets, Arc::new(values), None);
+ let output = evaluate(input);
+
+ assert_eq!(output.len(), 3);
+ assert_eq!(output.value(0).len(), 0);
+ assert_eq!(row_values(&output, 1), vec![0]);
+ assert_eq!(output.value(2).len(), 0);
+ }
+
+ #[test]
+ fn null_rows_are_inherited() {
+ // Row 1 is null; positions must inherit the null bit so downstream unnest
+ // aligns with the value array.
+ let values = Int32Array::from(vec![10, 20, 30]);
+ let offsets = OffsetBuffer::new(vec![0, 2, 2, 3].into());
+ let nulls = NullBuffer::from(vec![true, false, true]);
+ let input = ListArray::new(element_field(), offsets, Arc::new(values), Some(nulls));
+ let output = evaluate(input);
+
+ let out_nulls = output.nulls().expect("null bitmap must be inherited");
+ assert!(out_nulls.is_valid(0));
+ assert!(!out_nulls.is_valid(1));
+ assert!(out_nulls.is_valid(2));
+ assert_eq!(row_values(&output, 0), vec![0, 1]);
+ assert_eq!(row_values(&output, 2), vec![0]);
+ }
+
+ #[test]
+ fn zero_row_batch() {
+ let values = Int32Array::from(Vec::<i32>::new());
+ let offsets = OffsetBuffer::new(vec![0].into());
+ let input = ListArray::new(element_field(), offsets, Arc::new(values), None);
+ let output = evaluate(input);
+ assert_eq!(output.len(), 0);
+ }
+
+ #[test]
+ fn sliced_input_with_non_zero_offset_base() {
+ // 5-row list, slice out rows 1..4. `offsets()` becomes [2, 3, 3, 5]
+ // (base 2) while `values()` stays unsliced at length 6. Before the
+ // rebase fix, `ListArray::new` panics with
+ // `InvalidArgumentError("Max offset of 5 exceeds length of values 3")`.
+ let values = Int32Array::from(vec![1, 2, 3, 4, 5, 6]);
+ let offsets = OffsetBuffer::new(vec![0, 2, 3, 3, 5, 6].into());
+ let input = ListArray::new(element_field(), offsets, Arc::new(values), None);
+ let sliced: ArrayRef = Arc::new(input.slice(1, 3));
+
+ let out = evaluate_ref(sliced);
+ let list = out.as_any().downcast_ref::<ListArray>().unwrap();
+
+ assert_eq!(list.len(), 3);
+ // Logical row 0: was [3] -> [0]
+ assert_eq!(row_values(list, 0), vec![0]);
+ // Logical row 1: was [] -> []
+ assert_eq!(list.value(1).len(), 0);
+ // Logical row 2: was [4, 5] -> [0, 1]
+ assert_eq!(row_values(list, 2), vec![0, 1]);
+ }
+
+ #[test]
+ fn sliced_input_with_nulls_and_empty_rows() {
+ // Underlying: [1, 2], NULL, [], [3, 4, 5], [6]. Slice rows 1..5,
+ // so the logical view has a non-zero offset base and mixes null,
+ // empty, and non-empty rows.
+ let values = Int32Array::from(vec![1, 2, 3, 4, 5, 6]);
+ let offsets = OffsetBuffer::new(vec![0, 2, 2, 2, 5, 6].into());
+ let nulls = NullBuffer::from(vec![true, false, true, true, true]);
+ let input = ListArray::new(element_field(), offsets, Arc::new(values), Some(nulls));
+ let sliced: ArrayRef = Arc::new(input.slice(1, 4));
+
+ let out = evaluate_ref(sliced);
+ let list = out.as_any().downcast_ref::<ListArray>().unwrap();
+ assert_eq!(list.len(), 4);
+
+ let out_nulls = list.nulls().expect("null bitmap must be inherited");
+ // Logical row 0: NULL
+ assert!(!out_nulls.is_valid(0));
+ // Logical row 1: []
+ assert!(out_nulls.is_valid(1));
+ assert_eq!(list.value(1).len(), 0);
+ // Logical row 2: [3, 4, 5]
+ assert!(out_nulls.is_valid(2));
+ assert_eq!(row_values(list, 2), vec![0, 1, 2]);
+ // Logical row 3: [6]
+ assert!(out_nulls.is_valid(3));
+ assert_eq!(row_values(list, 3), vec![0]);
+ }
+}
diff --git a/native/core/src/execution/expressions/mod.rs b/native/core/src/execution/expressions/mod.rs
index e174bd3747..4c25109af6 100644
--- a/native/core/src/execution/expressions/mod.rs
+++ b/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;
diff --git a/native/core/src/execution/planner.rs b/native/core/src/execution/planner.rs
index f20dadf7f3..d4c97c509d 100644
--- a/native/core/src/execution/planner.rs
+++ b/native/core/src/execution/planner.rs
@@ -25,6 +25,7 @@ use crate::execution::operators::init_csv_datasource_exec;
use crate::execution::operators::AlignedArrowStreamReader;
use crate::execution::operators::IcebergScanExec;
use crate::execution::{
+ expressions::list_empty_to_null::ListEmptyToNullExpr,
expressions::list_positions::ListPositionsExpr,
expressions::subquery::Subquery,
operators::{
@@ -1897,7 +1898,7 @@ impl PhysicalPlanner {
self.create_plan(&children[0], inputs, partition_count)?;
// Create the expression for the array to explode
- let child_expr = if let Some(child_expr) = &explode.child {
+ let raw_child_expr = if let Some(child_expr) = &explode.child {
self.create_expr(child_expr, child.schema())?
} else {
return Err(ExecutionError::GeneralError(
@@ -1905,6 +1906,69 @@ impl PhysicalPlanner {
));
};
+ let child_schema = child.schema();
+ let child_field_name = raw_child_expr
+ .return_field(&child_schema)
+ .expect("Failed to get field from child expression")
+ .name()
+ .to_string();
+
+ // Bridge Spark's outer semantics: DataFusion's `UnnestExec` with
+ // `preserve_nulls = true` emits one null row for a NULL list but drops rows
+ // whose list is empty. Spark's `explode_outer`/`posexplode_outer` must emit
+ // exactly one null row in both cases, so we mark empty rows as null before
+ // unnesting. See https://github.com/apache/datafusion/issues/19053. Once
+ // Comet moves to a DataFusion release carrying
+ // https://github.com/apache/datafusion/pull/22100, `ListEmptyToNullExpr`
+ // and the pre-projection below can be removed in favor of
+ // `NullHandling::PreserveAndExpandEmpty`. See
+ // https://github.com/apache/datafusion-comet/issues/5210.
+ //
+ // For `posexplode_outer` the wrapped array is materialized in a
+ // pre-projection so `ListPositionsExpr` and the array passthrough share
+ // a single evaluation of `ListEmptyToNullExpr` instead of re-running it
+ // per branch. Plain `explode_outer` references the wrapped array exactly
+ // once, so no pre-projection is needed there.
+ let (child_expr, child_native_plan): (Arc<dyn PhysicalExpr>, _) =
+ match (explode.outer, explode.position) {
+ (true, true) => {
+ let wrapped: Arc<dyn PhysicalExpr> =
+ Arc::new(ListEmptyToNullExpr::new(raw_child_expr));
+ let reserved_name =
+ format!("__comet_explode_outer_{}", child_field_name);
+
+ let mut pre_exprs: Vec<(Arc<dyn PhysicalExpr>, String)> = child_schema
+ .fields()
+ .iter()
+ .enumerate()
+ .map(|(i, f)| {
+ (
+ Arc::new(Column::new(f.name(), i)) as Arc<dyn PhysicalExpr>,
+ f.name().to_string(),
+ )
+ })
+ .collect();
+ let wrapped_idx = pre_exprs.len();
+ pre_exprs.push((wrapped, reserved_name.clone()));
+
+ let pre_exec = Arc::new(ProjectionExec::try_new(
+ pre_exprs,
+ Arc::clone(&child.native_plan),
+ )?);
+ (
+ Arc::new(Column::new(&reserved_name, wrapped_idx))
+ as Arc<dyn PhysicalExpr>,
+ pre_exec as Arc<dyn ExecutionPlan>,
+ )
+ }
+ (true, false) => (
+ Arc::new(ListEmptyToNullExpr::new(raw_child_expr))
+ as Arc<dyn PhysicalExpr>,
+ Arc::clone(&child.native_plan),
+ ),
+ (false, _) => (raw_child_expr, Arc::clone(&child.native_plan)),
+ };
+
// Create projection expressions for other columns
let projections: Vec<Arc<dyn PhysicalExpr>> = explode
.project_list
@@ -1914,7 +1978,6 @@ impl PhysicalPlanner {
// For posexplode, a parallel List<Int32> positions column is added before the
// array column so UnnestExec can unnest both in parallel.
- let child_schema = child.schema();
let mut project_exprs: Vec<(Arc<dyn PhysicalExpr>, String)> = projections
.iter()
.map(|expr| {
@@ -1926,22 +1989,15 @@ impl PhysicalPlanner {
})
.collect();
- let array_field = child_expr
- .return_field(&child_schema)
- .expect("Failed to get field from array expression");
- let array_col_name = array_field.name().to_string();
-
if explode.position {
let positions_expr: Arc<dyn PhysicalExpr> =
Arc::new(ListPositionsExpr::new(Arc::clone(&child_expr)));
project_exprs.push((positions_expr, "pos".to_string()));
}
- project_exprs.push((Arc::clone(&child_expr), array_col_name.clone()));
+ project_exprs.push((Arc::clone(&child_expr), child_field_name.clone()));
- let project_exec = Arc::new(ProjectionExec::try_new(
- project_exprs,
- Arc::clone(&child.native_plan),
- )?);
+ let project_exec =
+ Arc::new(ProjectionExec::try_new(project_exprs, child_native_plan)?);
let project_schema = project_exec.schema();
diff --git a/spark/src/main/scala/org/apache/spark/sql/comet/operators.scala b/spark/src/main/scala/org/apache/spark/sql/comet/operators.scala
index 8e1efa0087..05439a16e3 100644
--- a/spark/src/main/scala/org/apache/spark/sql/comet/operators.scala
+++ b/spark/src/main/scala/org/apache/spark/sql/comet/operators.scala
@@ -58,7 +58,7 @@ import org.apache.comet.{CometConf, CometExecIterator, CometRuntimeException, Co
import org.apache.comet.CometSparkSessionExtensions.{isCometShuffleEnabled, withFallbackReason}
import org.apache.comet.parquet.CometParquetUtils
import org.apache.comet.rules.CometExecRule
-import org.apache.comet.serde.{CometOperatorSerde, Compatible, Incompatible, OperatorOuterClass, QueryContextInterner, SupportLevel, Unsupported}
+import org.apache.comet.serde.{CometOperatorSerde, Compatible, OperatorOuterClass, QueryContextInterner, SupportLevel, Unsupported}
import org.apache.comet.serde.OperatorOuterClass.{AggregateMode => CometAggregateMode, Operator}
import org.apache.comet.serde.QueryPlanSerde
import org.apache.comet.serde.QueryPlanSerde.{aggExprToProto, exprToProto, isStringCollationType, supportedSortType}
@@ -1452,11 +1452,6 @@ object CometExplodeExec extends CometOperatorSerde[GenerateExec] {
if (nodeName != "explode" && nodeName != "posexplode") {
return Unsupported(Some(s"Unsupported generator: ${op.generator.nodeName}"))
}
- if (op.outer) {
- // DataFusion UnnestExec has different semantics to Spark for this case
- // https://github.com/apache/datafusion/issues/19053
- return Incompatible(Some("Empty arrays are not preserved as null outputs when outer=true"))
- }
op.generator.children.head.dataType match {
case _: ArrayType =>
Compatible()
diff --git a/spark/src/test/resources/sql-tests/expressions/array/explode.sql b/spark/src/test/resources/sql-tests/expressions/array/explode.sql
new file mode 100644
index 0000000000..68a2bcc943
--- /dev/null
+++ b/spark/src/test/resources/sql-tests/expressions/array/explode.sql
@@ -0,0 +1,500 @@
+-- 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.
+
+-- Exercises explode and explode_outer across every primitive data type Spark
+-- supports for array element input, including null arrays, empty arrays, null
+-- array elements, floating-point specials, integer/decimal boundaries, nested
+-- arrays, and arrays of structs. Comet's explode_outer must emit exactly one
+-- row (with a null value) for both null and empty inputs, matching Spark's
+-- GenerateExec `outer` semantics.
+
+-- ===== INT arrays: null id, null / empty / non-empty arrays =====
+
+-- MinSparkVersion: 3.5
+
+statement
+CREATE TABLE test_explode_int(id int, arr array<int>) USING parquet
+
+statement
+INSERT INTO test_explode_int VALUES
+ (1, array(10, 20, 30)),
+ (2, array(40)),
+ (3, array()),
+ (4, NULL),
+ (5, array(NULL, 50, NULL)),
+ (NULL, array(60, 70)),
+ (NULL, array()),
+ (NULL, NULL)
+
+query
+SELECT id, explode(arr) AS v FROM test_explode_int
+
+query
+SELECT id, explode_outer(arr) AS v FROM test_explode_int
+
+-- LATERAL VIEW forms round-trip through the same Generate operator
+query
+SELECT id, v FROM test_explode_int LATERAL VIEW explode(arr) t AS v
+
+query
+SELECT id, v FROM test_explode_int LATERAL VIEW OUTER explode(arr) t AS v
+
+-- explode / explode_outer of a literal array (constant folding is disabled by
+-- the test runner, so the array reaches the operator as an expression)
+query
+SELECT id, explode_outer(array(100, 200, 300)) FROM test_explode_int WHERE id = 1
+
+query
+SELECT id, explode_outer(cast(NULL as array<int>)) FROM test_explode_int WHERE id = 1
+
+query
+SELECT id, explode_outer(array()) FROM test_explode_int WHERE id = 1
+
+-- ===== BOOLEAN =====
+
+statement
+CREATE TABLE test_explode_bool(id int, arr array<boolean>) USING parquet
+
+statement
+INSERT INTO test_explode_bool VALUES
+ (1, array(true, false, true)),
+ (2, array(NULL, true)),
+ (3, array()),
+ (4, NULL)
+
+query
+SELECT id, explode_outer(arr) FROM test_explode_bool
+
+-- ===== TINYINT / SMALLINT / BIGINT with min/max boundaries =====
+
+statement
+CREATE TABLE test_explode_tinyint(id int, arr array<tinyint>) USING parquet
+
+statement
+INSERT INTO test_explode_tinyint VALUES
+ (1, array(cast(-128 as tinyint), cast(0 as tinyint), cast(127 as tinyint))),
+ (2, array(cast(NULL as tinyint))),
+ (3, array()),
+ (4, NULL)
+
+query
+SELECT id, explode_outer(arr) FROM test_explode_tinyint
+
+statement
+CREATE TABLE test_explode_smallint(id int, arr array<smallint>) USING parquet
+
+statement
+INSERT INTO test_explode_smallint VALUES
+ (1, array(cast(-32768 as smallint), cast(0 as smallint), cast(32767 as smallint))),
+ (2, array()),
+ (3, NULL)
+
+query
+SELECT id, explode_outer(arr) FROM test_explode_smallint
+
+statement
+CREATE TABLE test_explode_bigint(id int, arr array<bigint>) USING parquet
+
+statement
+INSERT INTO test_explode_bigint VALUES
+ (1, array(-9223372036854775808L, 0L, 9223372036854775807L)),
+ (2, array(NULL, 1L)),
+ (3, array()),
+ (4, NULL)
+
+query
+SELECT id, explode_outer(arr) FROM test_explode_bigint
+
+-- Integer min/max boundaries for the default INT case
+statement
+CREATE TABLE test_explode_int_bounds(id int, arr array<int>) USING parquet
+
+statement
+INSERT INTO test_explode_int_bounds VALUES
+ (1, array(-2147483648, 0, 2147483647)),
+ (2, array()),
+ (3, NULL)
+
+query
+SELECT id, explode_outer(arr) FROM test_explode_int_bounds
+
+-- ===== FLOAT / DOUBLE: NaN, +/-Infinity, +/-0.0, NULL, empty, null-array =====
+
+statement
+CREATE TABLE test_explode_float(id int, arr array<float>) USING parquet
+
+statement
+INSERT INTO test_explode_float VALUES
+ (1, array(cast('NaN' as float), cast(0.0 as float), cast(-0.0 as float))),
+ (2, array(cast('Infinity' as float), cast('-Infinity' as float))),
+ (3, array(cast(1.5 as float), NULL, cast(-1.5 as float))),
+ (4, array()),
+ (5, NULL)
+
+query
+SELECT id, explode_outer(arr) FROM test_explode_float
+
+statement
+CREATE TABLE test_explode_double(id int, arr array<double>) USING parquet
+
+statement
+INSERT INTO test_explode_double VALUES
+ (1, array(cast('NaN' as double), cast(0.0 as double), cast(-0.0 as double))),
+ (2, array(cast('Infinity' as double), cast('-Infinity' as double))),
+ (3, array(1.5, NULL, -1.5)),
+ (4, array()),
+ (5, NULL)
+
+query
+SELECT id, explode_outer(arr) FROM test_explode_double
+
+-- ===== DECIMAL boundaries =====
+
+statement
+CREATE TABLE test_explode_decimal(id int, arr array<decimal(18,4)>) USING parquet
+
+statement
+INSERT INTO test_explode_decimal VALUES
+ (1, array(cast('99999999999999.9999' as decimal(18,4)),
+ cast('-99999999999999.9999' as decimal(18,4)),
+ cast('0.0000' as decimal(18,4)))),
+ (2, array(cast(NULL as decimal(18,4)), cast('1.2345' as decimal(18,4)))),
+ (3, array()),
+ (4, NULL)
+
+query
+SELECT id, explode_outer(arr) FROM test_explode_decimal
+
+-- Large-precision decimal
+statement
+CREATE TABLE test_explode_decimal38(id int, arr array<decimal(38,10)>) USING parquet
+
+statement
+INSERT INTO test_explode_decimal38 VALUES
+ (1, array(cast('9999999999999999999999999999.9999999999' as decimal(38,10)),
+ cast('-9999999999999999999999999999.9999999999' as decimal(38,10)))),
+ (2, array()),
+ (3, NULL)
+
+query
+SELECT id, explode_outer(arr) FROM test_explode_decimal38
+
+-- ===== STRING =====
+
+statement
+CREATE TABLE test_explode_string(id int, arr array<string>) USING parquet
+
+statement
+INSERT INTO test_explode_string VALUES
+ (1, array('a', 'bb', 'ccc')),
+ (2, array('', ' ', 'unicode: éü')),
+ (3, array(NULL, 'x')),
+ (4, array()),
+ (5, NULL)
+
+query
+SELECT id, explode_outer(arr) FROM test_explode_string
+
+-- ===== BINARY =====
+
+statement
+CREATE TABLE test_explode_binary(id int, arr array<binary>) USING parquet
+
+statement
+INSERT INTO test_explode_binary VALUES
+ (1, array(cast('a' as binary), cast('bc' as binary))),
+ (2, array(NULL, cast('x' as binary))),
+ (3, array()),
+ (4, NULL)
+
+query
+SELECT id, explode_outer(arr) FROM test_explode_binary
+
+-- ===== DATE / TIMESTAMP =====
+
+statement
+CREATE TABLE test_explode_date(id int, arr array<date>) USING parquet
+
+statement
+INSERT INTO test_explode_date VALUES
+ (1, array(date '1970-01-01', date '9999-12-31', date '2024-06-15')),
+ (2, array(NULL, date '2024-02-29')),
+ (3, array()),
+ (4, NULL)
+
+query
+SELECT id, explode_outer(arr) FROM test_explode_date
+
+statement
+CREATE TABLE test_explode_ts(id int, arr array<timestamp>) USING parquet
+
+statement
+INSERT INTO test_explode_ts VALUES
+ (1, array(timestamp '1970-01-01 00:00:00', timestamp '2024-06-15 12:34:56.789')),
+ (2, array(NULL, timestamp '9999-12-31 23:59:59.999999')),
+ (3, array()),
+ (4, NULL)
+
+query
+SELECT id, explode_outer(arr) FROM test_explode_ts
+
+-- ===== Nested arrays: array<array<int>> =====
+
+statement
+CREATE TABLE test_explode_nested(id int, arr array<array<int>>) USING parquet
+
+statement
+INSERT INTO test_explode_nested VALUES
+ (1, array(array(1, 2), array(3))),
+ (2, array(array(), array(NULL))),
+ (3, array(NULL, array(4, 5))),
+ (4, array()),
+ (5, NULL)
+
+-- explode_outer of the outer array; inner arrays are emitted verbatim
+query
+SELECT id, explode_outer(arr) FROM test_explode_nested
+
+-- ===== Array of structs =====
+
+statement
+CREATE TABLE test_explode_struct(id int, arr array<struct<a: int, b: string>>) USING parquet
+
+statement
+INSERT INTO test_explode_struct VALUES
+ (1, array(named_struct('a', 10, 'b', 'x'), named_struct('a', 20, 'b', NULL))),
+ (2, array(named_struct('a', cast(NULL as int), 'b', 'y'))),
+ (3, array()),
+ (4, NULL)
+
+query
+SELECT id, explode_outer(arr) FROM test_explode_struct
+
+-- Access struct fields after exploding
+query
+SELECT id, v.a AS a, v.b AS b FROM test_explode_struct LATERAL VIEW OUTER explode(arr) t AS v
+
+-- ===== Multiple projected columns, several of them nullable =====
+
+statement
+CREATE TABLE test_explode_multi(id int, name string, extra bigint, arr array<int>) USING parquet
+
+statement
+INSERT INTO test_explode_multi VALUES
+ (1, 'A', 100L, array(1, 2, 3)),
+ (2, NULL, 200L, array()),
+ (3, 'C', NULL, array(4)),
+ (4, NULL, NULL, NULL),
+ (NULL, 'E', 500L, array(5, 6))
+
+query
+SELECT id, name, extra, explode_outer(arr) AS v FROM test_explode_multi
+
+-- ===== Pre-projection wiring: carry the array column through alongside its
+-- explosion. The passthrough `arr` shows the original array (empty rows stay
+-- []) while the exploded value is NULL for empty rows, so this is the only
+-- shape where the difference between the original array and the null-marked
+-- copy is observable at the query level.
+
+query
+SELECT id, arr, explode_outer(arr) FROM test_explode_int
+
+-- ===== Pre-projection wiring: no passthrough columns. This drives the
+-- planner's `project_list` to empty (no columns carried through) and covers
+-- the codepath where the second projection contains only the exploded array.
+
+query
+SELECT explode_outer(arr) FROM test_explode_int
+
+-- ===== Empty table (zero input rows) =====
+
+statement
+CREATE TABLE test_explode_empty(id int, arr array<int>) USING parquet
+
+query
+SELECT id, explode_outer(arr) FROM test_explode_empty
+
+-- ===== Map falls back to Spark: outer form must also fall back =====
+
+statement
+CREATE TABLE test_explode_map(id int, m map<string, int>) USING parquet
+
+statement
+INSERT INTO test_explode_map VALUES
+ (1, map('k1', 1, 'k2', 2)),
+ (2, map()),
+ (3, NULL)
+
+query expect_fallback(Comet only supports explode/explode_outer for arrays, not maps)
+SELECT id, explode_outer(m) FROM test_explode_map
+
+-- ===== TIMESTAMP_NTZ (Spark 3.4+): distinct Arrow layout (no tz) from LTZ =====
+
+statement
+CREATE TABLE test_explode_ts_ntz(id int, arr array<timestamp_ntz>) USING parquet
+
+statement
+INSERT INTO test_explode_ts_ntz VALUES
+ (1, array(timestamp_ntz '1970-01-01 00:00:00', timestamp_ntz '2024-06-15 12:34:56.789')),
+ (2, array(NULL, timestamp_ntz '9999-12-31 23:59:59.999999')),
+ (3, array()),
+ (4, NULL)
+
+query
+SELECT id, explode_outer(arr) FROM test_explode_ts_ntz
+
+-- ===== array<map<...>> element type: distinct Arrow layout from array<array>/array<struct> =====
+
+statement
+CREATE TABLE test_explode_arr_map(id int, arr array<map<string, int>>) USING parquet
+
+statement
+INSERT INTO test_explode_arr_map VALUES
+ (1, array(map('a', 1, 'b', 2), map('c', 3))),
+ (2, array(map(), NULL)),
+ (3, array()),
+ (4, NULL)
+
+query
+SELECT id, explode_outer(arr) FROM test_explode_arr_map
+
+-- ===== Batches of uniformly empty or uniformly null arrays =====
+
+statement
+CREATE TABLE test_explode_all_empty(id int, arr array<int>) USING parquet
+
+statement
+INSERT INTO test_explode_all_empty VALUES (1, array()), (2, array()), (3, array())
+
+query
+SELECT id, explode_outer(arr) FROM test_explode_all_empty
+
+statement
+CREATE TABLE test_explode_all_null(id int, arr array<int>) USING parquet
+
+statement
+INSERT INTO test_explode_all_null VALUES (1, NULL), (2, NULL), (3, NULL)
+
+query
+SELECT id, explode_outer(arr) FROM test_explode_all_null
+
+-- ===== Stacked LATERAL VIEW OUTER: composition of two GenerateExec outer paths =====
+
+statement
+CREATE TABLE test_explode_stacked(id int, arrs array<array<int>>) USING parquet
+
+statement
+INSERT INTO test_explode_stacked VALUES
+ (1, array(array(1, 2), array())),
+ (2, array(array(), NULL)),
+ (3, array()),
+ (4, NULL)
+
+query
+SELECT id, x, y
+FROM test_explode_stacked
+LATERAL VIEW OUTER explode(arrs) t1 AS x
+LATERAL VIEW OUTER explode(x) t2 AS y
+
+-- ===== Downstream consumers of the exploded (nullable) value column =====
+
+query
+SELECT v, count(*) AS c
+FROM test_explode_int LATERAL VIEW OUTER explode(arr) t AS v
+GROUP BY v
+
+query
+SELECT id, v
+FROM test_explode_int LATERAL VIEW OUTER explode(arr) t AS v
+WHERE v IS NULL
+
+query
+SELECT a.id, b.v
+FROM test_explode_int a
+JOIN (SELECT id, explode_outer(arr) AS v FROM test_explode_int) b
+ON a.id = b.v
+
+-- ===== Interval-typed arrays: expected to fall back at the scan gate =====
+
+statement
+CREATE TABLE test_explode_ym(id int, arr array<interval year to month>) USING parquet
+
+statement
+INSERT INTO test_explode_ym VALUES
+ (1, array(interval '1-2' year to month, interval '-3-4' year to month, NULL)),
+ (2, array()),
+ (3, NULL)
+
+query spark_answer_only
+SELECT id, explode_outer(arr) FROM test_explode_ym
+
+statement
+CREATE TABLE test_explode_dt(id int, arr array<interval day to second>) USING parquet
+
+statement
+INSERT INTO test_explode_dt VALUES
+ (1, array(interval '1 02:03:04' day to second, interval '-5 06:07:08.9' day to second, NULL)),
+ (2, array()),
+ (3, NULL)
+
+query spark_answer_only
+SELECT id, explode_outer(arr) FROM test_explode_dt
+
+-- ===== Non-attribute generator child: the array expression is computed from
+-- input columns instead of read directly. `CometExplodeExec.convert` routes the
+-- child through `exprToProto` and falls back only when the child fails to
+-- convert. Existing tests only pass bare attributes or a pure-literal array, so
+-- this section pins the column-referencing expression-child path.
+
+statement
+CREATE TABLE test_explode_expr_child(id int, a array<int>, b array<int>) USING parquet
+
+statement
+INSERT INTO test_explode_expr_child VALUES
+ (1, array(1, 2), array(3, 4)),
+ (2, array(), array(5)),
+ (3, NULL, array(6, 7)),
+ (4, array(8), NULL)
+
+query
+SELECT id, explode(concat(a, b)) AS v FROM test_explode_expr_child
+
+query
+SELECT id, explode_outer(concat(a, b)) AS v FROM test_explode_expr_child
+
+query
+SELECT id, explode(slice(concat(a, b), 1, 2)) AS v FROM test_explode_expr_child
+
+statement
+CREATE TABLE test_explode_array_ctor(id int, x int, y int, z int) USING parquet
+
+statement
+INSERT INTO test_explode_array_ctor VALUES
+ (1, 10, 20, 30),
+ (2, NULL, 40, 50),
+ (3, 60, NULL, 70)
+
+query
+SELECT id, explode(array(x, y, z)) AS v FROM test_explode_array_ctor
+
+query
+SELECT id, explode_outer(array(x, y, z)) AS v FROM test_explode_array_ctor
+
+-- Spark 3.5+ supports non determenistic functions in generators
+-- https://github.com/apache/datafusion-comet/issues/5238
+query expect_fallback(Only deterministic generators are supported)
+SELECT id, explode(array(rand(0))) FROM (SELECT 1 id) WHERE id = 1
diff --git a/spark/src/test/resources/sql-tests/expressions/array/posexplode.sql b/spark/src/test/resources/sql-tests/expressions/array/posexplode.sql
index b7ba70d45a..cca5650b05 100644
--- a/spark/src/test/resources/sql-tests/expressions/array/posexplode.sql
+++ b/spark/src/test/resources/sql-tests/expressions/array/posexplode.sql
@@ -15,9 +15,8 @@
-- specific language governing permissions and limitations
-- under the License.
--- posexplode_outer is gated behind allowIncompatible=true (DataFusion #19053).
--- Setting it at file scope does not affect the non-outer cases.
--- Config: spark.comet.operator.GenerateExec.allowIncompatible=true
+-- posexplode_outer is now supported natively; see DataFusion #19053 handling
+-- via ListEmptyToNullExpr in the planner.
statement
CREATE TABLE test_posexplode_int(id int, arr array<int>) USING parquet
@@ -38,11 +37,10 @@ SELECT id, posexplode(arr) FROM test_posexplode_int
query
SELECT id, pos, value FROM test_posexplode_int LATERAL VIEW posexplode(arr) p AS pos, value
--- posexplode_outer keeps rows whose array is NULL. Empty arrays are excluded by
--- the WHERE clause because DataFusion #19053 drops empty arrays under
--- preserve_nulls=true; that is the documented Incompatible behavior.
+-- posexplode_outer keeps rows whose array is NULL or empty. Comet emits one
+-- row with a NULL position and NULL value for each such input row.
query
-SELECT id, posexplode_outer(arr) FROM test_posexplode_int WHERE id != 4
+SELECT id, posexplode_outer(arr) FROM test_posexplode_int
-- posexplode of a literal array (constant folding is disabled by the test runner)
query
@@ -98,3 +96,204 @@ INSERT INTO test_posexplode_map VALUES
-- posexplode over a map falls back to Spark (Comet only supports array inputs, not maps)
query expect_fallback(Comet only supports explode/explode_outer for arrays, not maps)
SELECT id, posexplode(m) FROM test_posexplode_map
+
+-- ===== posexplode_outer across non-int element types =====
+
+-- posexplode_outer with nullable int elements (in-array NULLs plus outer null row)
+query
+SELECT id, posexplode_outer(arr) FROM test_posexplode_nullable
+
+-- posexplode_outer over an array of strings
+query
+SELECT id, posexplode_outer(arr) FROM test_posexplode_str
+
+-- posexplode_outer over an array of structs via LATERAL VIEW OUTER, then project fields
+query
+SELECT id, pos, value.v1 AS v1, value.v2 AS v2
+FROM test_posexplode_struct LATERAL VIEW OUTER posexplode(arr) p AS pos, value
+
+statement
+CREATE TABLE test_posexplode_outer_types(
+ id int,
+ arr_bool array<boolean>,
+ arr_bi array<bigint>,
+ arr_dbl array<double>,
+ arr_dec array<decimal(18,4)>,
+ arr_bin array<binary>,
+ arr_dt array<date>,
+ arr_ts array<timestamp>) USING parquet
+
+statement
+INSERT INTO test_posexplode_outer_types VALUES
+ (1,
+ array(true, false, NULL),
+ array(-9223372036854775808L, 0L, 9223372036854775807L),
+ array(cast('NaN' as double), cast(0.0 as double), cast(-0.0 as double), cast('Infinity' as double), cast('-Infinity' as double)),
+ array(cast('99999999999999.9999' as decimal(18,4)), cast('-99999999999999.9999' as decimal(18,4)), cast(NULL as decimal(18,4))),
+ array(cast('a' as binary), NULL, cast('bc' as binary)),
+ array(date '1970-01-01', date '9999-12-31', NULL),
+ array(timestamp '1970-01-01 00:00:00', timestamp '2024-06-15 12:34:56.789', NULL)),
+ (2, array(), array(), array(), array(), array(), array(), array()),
+ (3, NULL, NULL, NULL, NULL, NULL, NULL, NULL)
+
+query
+SELECT id, posexplode_outer(arr_bool) FROM test_posexplode_outer_types
+
+query
+SELECT id, posexplode_outer(arr_bi) FROM test_posexplode_outer_types
+
+query
+SELECT id, posexplode_outer(arr_dbl) FROM test_posexplode_outer_types
+
+query
+SELECT id, posexplode_outer(arr_dec) FROM test_posexplode_outer_types
+
+query
+SELECT id, posexplode_outer(arr_bin) FROM test_posexplode_outer_types
+
+query
+SELECT id, posexplode_outer(arr_dt) FROM test_posexplode_outer_types
+
+query
+SELECT id, posexplode_outer(arr_ts) FROM test_posexplode_outer_types
+
+-- posexplode_outer over an array of structs (explicit projection form)
+query
+SELECT id, posexplode_outer(arr) FROM test_posexplode_struct
+
+-- LATERAL VIEW OUTER posexplode over int arrays: covers the analyzer's
+-- `MultiAlias(GeneratorOuter(g), names)` branch alongside the direct
+-- `posexplode_outer(...)` projection form above.
+query
+SELECT id, pos, value
+FROM test_posexplode_int LATERAL VIEW OUTER posexplode(arr) p AS pos, value
+
+-- ===== Pre-projection wiring for posexplode_outer =====
+
+-- Carry the array column through alongside posexplode_outer. The passthrough
+-- `arr` shows the original array (empty rows stay []) while the exploded pos
+-- and value are NULL for empty rows, so this is the only shape where the
+-- difference between the original array and the null-marked copy is
+-- observable at the query level.
+query
+SELECT id, arr, posexplode_outer(arr) FROM test_posexplode_int
+
+-- No passthrough columns. This drives `project_list` to empty in the planner
+-- and covers the codepath where the second projection contains only the
+-- positions column and the exploded array.
+query
+SELECT posexplode_outer(arr) FROM test_posexplode_int
+
+-- posexplode_outer batch of only-empty arrays exercises the slow path with an
+-- all-zeros non-empty bitmap; only-null exercises the fast-path passthrough.
+statement
+CREATE TABLE test_posexplode_all_empty(id int, arr array<int>) USING parquet
+
+statement
+INSERT INTO test_posexplode_all_empty VALUES (1, array()), (2, array()), (3, array())
+
+query
+SELECT id, posexplode_outer(arr) FROM test_posexplode_all_empty
+
+statement
+CREATE TABLE test_posexplode_all_null(id int, arr array<int>) USING parquet
+
+statement
+INSERT INTO test_posexplode_all_null VALUES (1, NULL), (2, NULL), (3, NULL)
+
+query
+SELECT id, posexplode_outer(arr) FROM test_posexplode_all_null
+
+-- ===== `pos` column value validation. Existing queries project `pos`
+-- alongside `id`/`value` but never filter, group, or aggregate on it, so a
+-- native `ListPositionsExpr` regression that emitted always-zero, off-by-one,
+-- or NULL-swapped-with-zero indices could still match on golden row order.
+-- These queries force value-level assertions on the position column.
+
+query
+SELECT pos, count(*) AS c
+FROM test_posexplode_int LATERAL VIEW posexplode(arr) p AS pos, value
+GROUP BY pos
+ORDER BY pos
+
+query
+SELECT id, pos, value
+FROM test_posexplode_int LATERAL VIEW posexplode(arr) p AS pos, value
+WHERE pos = 0
+ORDER BY id
+
+query
+SELECT id, pos, value
+FROM test_posexplode_int LATERAL VIEW OUTER posexplode(arr) p AS pos, value
+WHERE pos IS NULL
+ORDER BY id
+
+query
+SELECT id, pos, value
+FROM test_posexplode_int LATERAL VIEW OUTER posexplode(arr) p AS pos, value
+WHERE pos IS NOT NULL
+ORDER BY id, pos
+
+query
+SELECT id, MAX(pos) AS max_pos
+FROM test_posexplode_int LATERAL VIEW posexplode(arr) p AS pos, value
+GROUP BY id
+ORDER BY id
+
+-- Reconstruct the original array by ordering `value` by `pos`: a strong
+-- regression net against pos/value branch misalignment (see #5224).
+query
+SELECT id, collect_list(value) AS values_ordered_by_pos
+FROM (
+ SELECT id, pos, value
+ FROM test_posexplode_int LATERAL VIEW posexplode(arr) p AS pos, value
+ ORDER BY id, pos
+)
+GROUP BY id
+ORDER BY id
+
+-- ===== Stacked LATERAL VIEW OUTER posexplode: two GenerateExec outer paths
+-- composed back-to-back, where the inner `pos` must reset to 0 per outer
+-- element and empty / NULL inner arrays yield (NULL, NULL) against the
+-- correct outer pos. explode.sql covers the plain-explode counterpart; this
+-- exercises the parallel positions branch through two nested unnests.
+
+statement
+CREATE TABLE test_posexplode_stacked(id int, arrs array<array<int>>) USING parquet
+
+statement
+INSERT INTO test_posexplode_stacked VALUES
+ (1, array(array(10, 20), array(), array(30))),
+ (2, array(array(), NULL)),
+ (3, array()),
+ (4, NULL)
+
+query
+SELECT id, p1, p2, y
+FROM test_posexplode_stacked
+LATERAL VIEW OUTER posexplode(arrs) t1 AS p1, x
+LATERAL VIEW OUTER posexplode(x) t2 AS p2, y
+
+-- ===== Non-attribute generator child for posexplode: the array expression is
+-- computed from input columns. Explode has its own coverage of this shape
+-- (see explode.sql); posexplode must exercise it separately because the
+-- serde builds a distinct native operator with the parallel positions branch.
+
+statement
+CREATE TABLE test_posexplode_expr_child(id int, a array<int>, b array<int>) USING parquet
+
+statement
+INSERT INTO test_posexplode_expr_child VALUES
+ (1, array(10, 20), array(30, 40)),
+ (2, array(), array(50)),
+ (3, NULL, array(60, 70)),
+ (4, array(80), NULL)
+
+query
+SELECT id, posexplode(concat(a, b)) FROM test_posexplode_expr_child
+
+query
+SELECT id, posexplode_outer(concat(a, b)) FROM test_posexplode_expr_child
+
+query
+SELECT id, posexplode(slice(concat(a, b), 1, 2)) FROM test_posexplode_expr_child
diff --git a/spark/src/test/scala/org/apache/comet/exec/CometGenerateExecSuite.scala b/spark/src/test/scala/org/apache/comet/exec/CometGenerateExecSuite.scala
index d03522afe1..7fe827fa40 100644
--- a/spark/src/test/scala/org/apache/comet/exec/CometGenerateExecSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/exec/CometGenerateExecSuite.scala
@@ -20,7 +20,6 @@
package org.apache.comet.exec
import org.apache.spark.sql.CometTestBase
-import org.apache.spark.sql.execution.GenerateExec
import org.apache.spark.sql.functions.col
import org.apache.comet.CometConf
@@ -65,7 +64,6 @@ class CometGenerateExecSuite extends CometTestBase {
test("explode_outer with simple array") {
withSQLConf(
CometConf.COMET_EXEC_LOCAL_TABLE_SCAN_ENABLED.key -> "true",
- CometConf.getOperatorAllowIncompatConfigKey(classOf[GenerateExec]) -> "true",
CometConf.COMET_EXEC_EXPLODE_ENABLED.key -> "true") {
val df = Seq((1, Array(1, 2, 3)), (2, Array(4, 5)), (3, Array(6)))
.toDF("id", "arr")
@@ -74,8 +72,7 @@ class CometGenerateExecSuite extends CometTestBase {
}
}
- // https://github.com/apache/datafusion-comet/issues/2838
- ignore("explode_outer with empty array") {
+ test("explode_outer with empty array") {
withSQLConf(
CometConf.COMET_EXEC_LOCAL_TABLE_SCAN_ENABLED.key -> "true",
CometConf.COMET_EXEC_EXPLODE_ENABLED.key -> "true") {
@@ -89,7 +86,6 @@ class CometGenerateExecSuite extends CometTestBase {
test("explode_outer with null array") {
withSQLConf(
CometConf.COMET_EXEC_LOCAL_TABLE_SCAN_ENABLED.key -> "true",
- CometConf.getOperatorAllowIncompatConfigKey(classOf[GenerateExec]) -> "true",
CometConf.COMET_EXEC_EXPLODE_ENABLED.key -> "true") {
val df = Seq((1, Some(Array(1, 2))), (2, None), (3, Some(Array(3))))
.toDF("id", "arr")
@@ -169,8 +165,7 @@ class CometGenerateExecSuite extends CometTestBase {
}
}
- // https://github.com/apache/datafusion-comet/issues/2838
- ignore("explode_outer with nullable projected column") {
+ test("explode_outer with nullable projected column") {
withSQLConf(
CometConf.COMET_EXEC_LOCAL_TABLE_SCAN_ENABLED.key -> "true",
CometConf.COMET_EXEC_EXPLODE_ENABLED.key -> "true") {
@@ -199,8 +194,7 @@ class CometGenerateExecSuite extends CometTestBase {
}
}
- // https://github.com/apache/datafusion-comet/issues/2838
- ignore("explode_outer with mixed null, empty, and non-empty arrays") {
+ test("explode_outer with mixed null, empty, and non-empty arrays") {
withSQLConf(
CometConf.COMET_EXEC_LOCAL_TABLE_SCAN_ENABLED.key -> "true",
CometConf.COMET_EXEC_EXPLODE_ENABLED.key -> "true") {
@@ -268,7 +262,6 @@ class CometGenerateExecSuite extends CometTestBase {
test("posexplode_outer with simple array") {
withSQLConf(
CometConf.COMET_EXEC_LOCAL_TABLE_SCAN_ENABLED.key -> "true",
- CometConf.getOperatorAllowIncompatConfigKey(classOf[GenerateExec]) -> "true",
CometConf.COMET_EXEC_EXPLODE_ENABLED.key -> "true") {
val df = Seq((1, Array(10, 20, 30)), (2, Array(40, 50)), (3, Array(60)))
.toDF("id", "arr")
@@ -386,4 +379,202 @@ class CometGenerateExecSuite extends CometTestBase {
}
}
+ test("explode_outer across batch boundary with mixed empty/null rows") {
+ // Mix null, empty, and non-empty rows and force multiple small batches so that
+ // `ListEmptyToNullExpr` runs on each batch and its fast/slow path split is exercised
+ // more than once with different offset patterns.
+ withSQLConf(
+ CometConf.COMET_EXEC_LOCAL_TABLE_SCAN_ENABLED.key -> "true",
+ CometConf.COMET_EXEC_EXPLODE_ENABLED.key -> "true",
+ CometConf.COMET_BATCH_SIZE.key -> "4") {
+ val rows: Seq[(Int, Option[Array[Int]])] = (1 to 40).map { i =>
+ val arr = i % 5 match {
+ case 0 => None
+ case 1 => Some(Array.empty[Int])
+ case _ => Some((0 until (i % 5)).map(j => i * 100 + j).toArray)
+ }
+ (i, arr)
+ }
+ val df = rows
+ .toDF("id", "arr")
+ .selectExpr("id", "explode_outer(arr) as value")
+ checkSparkAnswerAndOperator(df)
+ }
+ }
+
+ test("posexplode_outer across batch boundary with mixed empty/null rows") {
+ // Same shape as the explode_outer counterpart but exercises the parallel positions
+ // branch. With the pre-projection introduced for outer, `ListEmptyToNullExpr` runs once
+ // per batch and both branches share the same materialized array.
+ withSQLConf(
+ CometConf.COMET_EXEC_LOCAL_TABLE_SCAN_ENABLED.key -> "true",
+ CometConf.COMET_EXEC_EXPLODE_ENABLED.key -> "true",
+ CometConf.COMET_BATCH_SIZE.key -> "4") {
+ val rows: Seq[(Int, Option[Array[Int]])] = (1 to 40).map { i =>
+ val arr = i % 5 match {
+ case 0 => None
+ case 1 => Some(Array.empty[Int])
+ case _ => Some((0 until (i % 5)).map(j => i * 100 + j).toArray)
+ }
+ (i, arr)
+ }
+ val df = rows
+ .toDF("id", "arr")
+ .selectExpr("id", "posexplode_outer(arr) as (pos, value)")
+ checkSparkAnswerAndOperator(df)
+ }
+ }
+
+ // Regression tests for https://github.com/apache/datafusion-comet/issues/5224.
+ //
+ // A native limit with a non-zero offset produces a batch whose `ListArray` has a non-zero
+ // offset base (`LimitStream::poll_and_skip` does `batch.slice(self.skip, ...)`). Before the
+ // fix, `ListPositionsExpr` rebuilt a fresh values array numbered from zero but reused the
+ // input's original offset buffer, so `ListArray::new` panicked with
+ // "Max offset of N exceeds length of values M".
+ //
+ // `spark.sql.leafNodeDefaultParallelism = 1` is required rather than cosmetic: with the
+ // default parallelism each partition produces a one-row batch, `LimitStream` discards whole
+ // batches instead of slicing, and the bug is masked. AQE off just keeps the plan readable.
+
+ test("posexplode over limit with offset") {
+ withSQLConf(
+ "spark.sql.adaptive.enabled" -> "false",
+ "spark.sql.leafNodeDefaultParallelism" -> "1",
+ CometConf.COMET_EXEC_LOCAL_TABLE_SCAN_ENABLED.key -> "true",
+ CometConf.COMET_EXEC_EXPLODE_ENABLED.key -> "true") {
+ withTempView("t") {
+ Seq((1, Array(1, 2, 3)), (2, Array(4, 5)), (3, Array(6)), (4, Array(7, 8)), (5, Array(9)))
+ .toDF("id", "arr")
+ .createOrReplaceTempView("t")
+ val df = sql("SELECT id, posexplode(arr) FROM (SELECT id, arr FROM t LIMIT 4 OFFSET 1)")
+ checkSparkAnswerAndOperator(df)
+ }
+ }
+ }
+
+ test("posexplode_outer over limit with offset") {
+ withSQLConf(
+ "spark.sql.adaptive.enabled" -> "false",
+ "spark.sql.leafNodeDefaultParallelism" -> "1",
+ CometConf.COMET_EXEC_LOCAL_TABLE_SCAN_ENABLED.key -> "true",
+ CometConf.COMET_EXEC_EXPLODE_ENABLED.key -> "true") {
+ withTempView("t") {
+ Seq(
+ (1, Some(Array(1, 2, 3))),
+ (2, None),
+ (3, Some(Array.empty[Int])),
+ (4, Some(Array(7, 8))),
+ (5, None))
+ .toDF("id", "arr")
+ .createOrReplaceTempView("t")
+ val df =
+ sql("SELECT id, posexplode_outer(arr) FROM (SELECT id, arr FROM t LIMIT 4 OFFSET 1)")
+ checkSparkAnswerAndOperator(df)
+ }
+ }
+ }
+
+ test("explode over limit with offset") {
+ // Plain `explode` does not build `ListPositionsExpr`, so it did not trigger #5224.
+ // Guarded here so that a future regression in the values path is caught alongside the
+ // posexplode fix.
+ withSQLConf(
+ "spark.sql.adaptive.enabled" -> "false",
+ "spark.sql.leafNodeDefaultParallelism" -> "1",
+ CometConf.COMET_EXEC_LOCAL_TABLE_SCAN_ENABLED.key -> "true",
+ CometConf.COMET_EXEC_EXPLODE_ENABLED.key -> "true") {
+ withTempView("t") {
+ Seq((1, Array(1, 2, 3)), (2, Array(4, 5)), (3, Array(6)), (4, Array(7, 8)), (5, Array(9)))
+ .toDF("id", "arr")
+ .createOrReplaceTempView("t")
+ val df = sql("SELECT id, explode(arr) FROM (SELECT id, arr FROM t LIMIT 4 OFFSET 1)")
+ checkSparkAnswerAndOperator(df)
+ }
+ }
+ }
+
+ test("explode_outer over limit with offset") {
+ // Exercises `ListEmptyToNullExpr` on a sliced input with a non-zero offset base. The
+ // helper preserves the base offset and passes it through unchanged, so the fix in
+ // `ListPositionsExpr` is what actually keeps the parallel `pos` branch safe. This test
+ // covers the `explode_outer` shape without the `pos` branch.
+ withSQLConf(
+ "spark.sql.adaptive.enabled" -> "false",
+ "spark.sql.leafNodeDefaultParallelism" -> "1",
+ CometConf.COMET_EXEC_LOCAL_TABLE_SCAN_ENABLED.key -> "true",
+ CometConf.COMET_EXEC_EXPLODE_ENABLED.key -> "true") {
+ withTempView("t") {
+ Seq(
+ (1, Some(Array(1, 2, 3))),
+ (2, None),
+ (3, Some(Array.empty[Int])),
+ (4, Some(Array(7, 8))),
+ (5, None))
+ .toDF("id", "arr")
+ .createOrReplaceTempView("t")
+ val df =
+ sql("SELECT id, explode_outer(arr) FROM (SELECT id, arr FROM t LIMIT 4 OFFSET 1)")
+ checkSparkAnswerAndOperator(df)
+ }
+ }
+ }
+
+ // A single input row whose flattened output exceeds COMET_BATCH_SIZE forces
+ // UnnestExec to emit multiple output batches from one input row. This is a
+ // distinct axis from cross-batch input slicing (covered above): pos values
+ // must remain contiguous across the split, and `explode` / `explode_outer`
+ // must preserve element order. Existing batch-boundary tests use arrays of
+ // at most 5 elements against COMET_BATCH_SIZE=4, so no test forced this
+ // one-input-many-output-batches shape until now.
+
+ test("posexplode single row exceeds batch size") {
+ withSQLConf(
+ CometConf.COMET_EXEC_LOCAL_TABLE_SCAN_ENABLED.key -> "true",
+ CometConf.COMET_EXEC_EXPLODE_ENABLED.key -> "true",
+ CometConf.COMET_BATCH_SIZE.key -> "8") {
+ val df = Seq(
+ (1, (0 until 30).toArray),
+ (2, Array(100, 101)),
+ (3, Array.empty[Int]),
+ (4, (0 until 20).map(i => 200 + i).toArray))
+ .toDF("id", "arr")
+ .selectExpr("id", "posexplode(arr) as (pos, value)")
+ checkSparkAnswerAndOperator(df)
+ }
+ }
+
+ test("posexplode_outer single row exceeds batch size") {
+ withSQLConf(
+ CometConf.COMET_EXEC_LOCAL_TABLE_SCAN_ENABLED.key -> "true",
+ CometConf.COMET_EXEC_EXPLODE_ENABLED.key -> "true",
+ CometConf.COMET_BATCH_SIZE.key -> "8") {
+ val df = Seq(
+ (1, Some((0 until 30).toArray)),
+ (2, Some(Array(100, 101))),
+ (3, None),
+ (4, Some(Array.empty[Int])),
+ (5, Some((0 until 20).map(i => 200 + i).toArray)))
+ .toDF("id", "arr")
+ .selectExpr("id", "posexplode_outer(arr) as (pos, value)")
+ checkSparkAnswerAndOperator(df)
+ }
+ }
+
+ test("explode single row exceeds batch size") {
+ withSQLConf(
+ CometConf.COMET_EXEC_LOCAL_TABLE_SCAN_ENABLED.key -> "true",
+ CometConf.COMET_EXEC_EXPLODE_ENABLED.key -> "true",
+ CometConf.COMET_BATCH_SIZE.key -> "8") {
+ val df = Seq(
+ (1, (0 until 30).toArray),
+ (2, Array(100, 101)),
+ (3, Array.empty[Int]),
+ (4, (0 until 20).map(i => 200 + i).toArray))
+ .toDF("id", "arr")
+ .selectExpr("id", "explode(arr) as value")
+ checkSparkAnswerAndOperator(df)
+ }
+ }
+
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org