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/contributor-guide/native_shuffle.md b/docs/source/contributor-guide/native_shuffle.md index 1e98bc196c..18e80a90c8 100644 --- a/docs/source/contributor-guide/native_shuffle.md +++ b/docs/source/contributor-guide/native_shuffle.md @@ -43,13 +43,12 @@ Comet Native (columnar) → ColumnarToRowExec → rows → JVM Shuffle → Arrow Native shuffle (`CometExchange`) is selected when all of the following conditions are met: -1. **Shuffle mode allows native**: `spark.comet.shuffle.mode` is `native` or `auto`. +1. **Shuffle mode allows native**: `spark.comet.exec.shuffle.mode` is `native` or `auto`. 2. **Child plan is a Comet native operator**: The child must be a `CometPlan` that produces columnar output. Row-based Spark operators require JVM shuffle. 3. **Supported partitioning type**: Native shuffle supports: - - `HashPartitioning` - `RangePartitioning` - `SinglePartition` @@ -70,9 +69,8 @@ Native shuffle (`CometExchange`) is selected when all of the following condition ▼ ┌─────────────────────────────────────────────────────────────────────────────┐ │ CometNativeShuffleWriter │ -│ - Builds protobuf operator plan: ShuffleWriter(child = childNativeOp) │ -│ - Reads per-partition leaf iterators from CometNativeShuffleInputIterator │ -│ - Drives one CometExecIterator per partition │ +│ - Constructs protobuf operator plan │ +│ - Invokes native execution via CometExec.getCometIterator() │ └─────────────────────────────────────────────────────────────────────────────┘ │ ▼ (JNI) @@ -105,14 +103,13 @@ Native shuffle (`CometExchange`) is selected when all of the following condition ### Scala Side -| Class | Location | Description | -| ------------------------------ | ------------------------------------------------ | --------------------------------------------------------------------------------------------------------------------------------------------------- | -| `CometShuffleExchangeExec` | `.../shuffle/CometShuffleExchangeExec.scala` | Physical plan node. Validates types and partitioning, creates `CometShuffleDependency`. | -| `CometNativeShuffleWriter` | `.../shuffle/CometNativeShuffleWriter.scala` | Implements `ShuffleWriter`. Builds the unified `ShuffleWriter(child = childNativeOp)` plan and runs it in one `CometExecIterator` per partition. | -| `CometShuffleDependency` | `.../shuffle/CometShuffleDependency.scala` | Extends `ShuffleDependency`. Holds shuffle type, schema, range partition bounds, and (native shuffle only) a `NativeShuffleSpec`. | -| `CometNativeShuffleInputRDD` | `.../shuffle/CometNativeShuffleInputRDD.scala` | Thin scheduling-anchor RDD on the native-shuffle path. `compute` returns a `CometNativeShuffleInputIterator` carrying per-partition leaf iterators. | -| `CometBlockStoreShuffleReader` | `.../shuffle/CometBlockStoreShuffleReader.scala` | Reads shuffle blocks via `ShuffleBlockFetcherIterator`. Decodes Arrow IPC to `ColumnarBatch`. | -| `NativeBatchDecoderIterator` | `.../shuffle/NativeBatchDecoderIterator.scala` | Reads compressed Arrow IPC from input stream. Calls native decode via JNI. | +| Class | Location | Description | +| ------------------------------ | ------------------------------------------------ | --------------------------------------------------------------------------------------------- | +| `CometShuffleExchangeExec` | `.../shuffle/CometShuffleExchangeExec.scala` | Physical plan node. Validates types and partitioning, creates `CometShuffleDependency`. | +| `CometNativeShuffleWriter` | `.../shuffle/CometNativeShuffleWriter.scala` | Implements `ShuffleWriter`. Builds protobuf plan and invokes native execution. | +| `CometShuffleDependency` | `.../shuffle/CometShuffleDependency.scala` | Extends `ShuffleDependency`. Holds shuffle type, schema, and range partition bounds. | +| `CometBlockStoreShuffleReader` | `.../shuffle/CometBlockStoreShuffleReader.scala` | Reads shuffle blocks via `ShuffleBlockFetcherIterator`. Decodes Arrow IPC to `ColumnarBatch`. | +| `NativeBatchDecoderIterator` | `.../shuffle/NativeBatchDecoderIterator.scala` | Reads compressed Arrow IPC from input stream. Calls native decode via JNI. | ### Rust Side @@ -126,23 +123,13 @@ Native shuffle (`CometExchange`) is selected when all of the following condition ### Write Path -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. - - A synthetic `Scan("ShuffleWriterInput")` placeholder, when the dep was built via the - convenience `prepareShuffleDependency(rdd, ...)` overload (used by - `CometCollectLimitExec` and `CometTakeOrderedAndProjectExec`, or when the - exchange's child is a non-native `CometPlan` such as `CometSparkToColumnarExec`). Native - code reads `ColumnarBatch`es from the JVM input iterator via Arrow C Stream Interface. +1. **Plan construction**: `CometNativeShuffleWriter` builds a protobuf operator plan containing: + - A scan operator reading from the input iterator + - A `ShuffleWriter` operator with partitioning config and compression codec -2. **Native execution**: A single `CometExecIterator` per partition runs the unified plan. +2. **Native execution**: `CometExec.getCometIterator()` executes the plan in Rust. 3. **Partitioning**: `ShuffleWriterExec` receives batches and routes to the appropriate partitioner: - - `MultiPartitionShuffleRepartitioner`: For hash/range/round-robin partitioning - `SinglePartitionShufflePartitioner`: For single partition (simpler path) @@ -150,13 +137,11 @@ Native shuffle (`CometExchange`) is selected when all of the following condition 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 +153,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 @@ -199,10 +183,8 @@ For range partitioning: ### Single Partition -The simplest case: all rows go to partition 0. Uses `SinglePartitionShufflePartitioner`, which -streams each batch straight to the writer, whose `BatchCoalescer` combines small batches up to the -configured batch size. Batches already at least that size pass through unchanged, so a large input -batch is written as a single block that may exceed the batch size. +The simplest case: all rows go to partition 0. Uses `SinglePartitionShufflePartitioner` which +simply concatenates batches to reach the configured batch size. ### Round Robin Partitioning @@ -222,9 +204,7 @@ sizes. Native shuffle uses DataFusion's memory management with spilling support: - **Memory pool**: Tracks memory usage across the shuffle operation. -- **Spill triggers**: Partitions spill to disk when the memory pool denies an allocation, or - when the buffered bytes reach `spark.comet.shuffle.native.maxBufferBytes`. That config defaults to - 0, which disables the fixed limit and leaves memory pressure as the only trigger. +- **Spill threshold**: When buffered data exceeds the threshold, partitions spill to disk. - **Per-partition spilling**: Each partition has its own spill file. Multiple spills for a partition are concatenated when writing the final output. - **Scratch space**: Reusable buffers for partition ID computation to reduce allocations. @@ -238,7 +218,7 @@ The `MultiPartitionShuffleRepartitioner` manages: ## Compression Native shuffle supports multiple compression codecs configured via -`spark.comet.shuffle.compression.codec`: +`spark.comet.exec.shuffle.compression.codec`: | Codec | Description | | -------- | ------------------------------------------------------ | @@ -252,14 +232,14 @@ independently compressed, allowing parallel decompression during reads. ## Configuration -| Config | Default | Description | -| -------------------------------------------- | ------- | ---------------------------------------- | -| `spark.comet.shuffle.enabled` | `true` | Enable Comet shuffle | -| `spark.comet.shuffle.mode` | `auto` | Shuffle mode: `native`, `jvm`, or `auto` | -| `spark.comet.shuffle.compression.codec` | `zstd` | Compression codec | -| `spark.comet.shuffle.compression.zstd.level` | `1` | Zstd compression level | -| `spark.comet.shuffle.native.writeBufferSize` | `1MB` | Write buffer size | -| `spark.comet.shuffle.jvm.batchSize` | `8192` | Target rows per batch | +| Config | Default | Description | +| ------------------------------------------------- | ------- | ---------------------------------------- | +| `spark.comet.exec.shuffle.enabled` | `true` | Enable Comet shuffle | +| `spark.comet.exec.shuffle.mode` | `auto` | Shuffle mode: `native`, `jvm`, or `auto` | +| `spark.comet.exec.shuffle.compression.codec` | `zstd` | Compression codec | +| `spark.comet.exec.shuffle.compression.zstd.level` | `1` | Zstd compression level | +| `spark.comet.shuffle.write.buffer.size` | `1MB` | Write buffer size | +| `spark.comet.columnar.shuffle.batch.size` | `8192` | Target rows per batch | ## Comparison with JVM Shuffle 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 130762d560..b7a79ab8a0 100644 --- a/docs/source/user-guide/latest/understanding-comet-plans.md +++ b/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 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..a2a02eece1 --- /dev/null +++ b/native/core/src/execution/expressions/list_empty_to_null.rs @@ -0,0 +1,313 @@ +// 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` input as null. +/// Bridges DataFusion's `UnnestExec` (which drops empty rows under +/// `preserve_nulls=true`) to Spark's `explode_outer`/`posexplode_outer` +/// semantics. See . +#[derive(Debug, Clone)] +pub struct ListEmptyToNullExpr { + child: Arc, +} + +impl ListEmptyToNullExpr { + pub fn new(child: Arc) -> 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(&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 { + // 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 { + let value = self.child.evaluate(batch)?; + let array = value.into_array(batch.num_rows())?; + + let Some(list) = array.as_any().downcast_ref::() 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 combined = match existing_nulls { + None => non_empty, + Some(existing) => existing.inner() & &non_empty, + }; + 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> { + vec![&self.child] + } + + fn with_new_children( + self: Arc, + children: Vec>, + ) -> DataFusionResult> { + 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 { + Arc::new(Field::new("item", DataType::Int32, true)) + } + + fn list_field(nullable: bool) -> Arc { + 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::().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::().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::::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/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 39e7ccc2e5..d6923f7285 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::{ @@ -1838,7 +1839,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( @@ -1846,6 +1847,66 @@ 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 + // that upstream fix lands, `ListEmptyToNullExpr` and the pre-projection + // below can be removed (TODO: link the Comet tracking issue here). + // + // 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, _) = + match (explode.outer, explode.position) { + (true, true) => { + let wrapped: Arc = + Arc::new(ListEmptyToNullExpr::new(raw_child_expr)); + let reserved_name = + format!("__comet_explode_outer_{}", child_field_name); + + let mut pre_exprs: Vec<(Arc, String)> = child_schema + .fields() + .iter() + .enumerate() + .map(|(i, f)| { + ( + Arc::new(Column::new(f.name(), i)) as Arc, + 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, + pre_exec as Arc, + ) + } + (true, false) => ( + Arc::new(ListEmptyToNullExpr::new(raw_child_expr)) + as Arc, + Arc::clone(&child.native_plan), + ), + (false, _) => (raw_child_expr, Arc::clone(&child.native_plan)), + }; + // Create projection expressions for other columns let projections: Vec> = explode .project_list @@ -1855,7 +1916,6 @@ impl PhysicalPlanner { // For posexplode, a parallel List 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, String)> = projections .iter() .map(|expr| { @@ -1867,22 +1927,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 = 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 b422d60e67..17aa3ce4bb 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, SupportLevel, Unsupported} +import org.apache.comet.serde.{CometOperatorSerde, Compatible, OperatorOuterClass, 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} @@ -1426,11 +1426,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..527e15e9da --- /dev/null +++ b/spark/src/test/resources/sql-tests/expressions/array/explode.sql @@ -0,0 +1,453 @@ +-- 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 ===== + +statement +CREATE TABLE test_explode_int(id int, arr array) 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)) 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) 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) 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) 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) 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) 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) 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) 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) 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) 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) 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) 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) 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) 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> ===== + +statement +CREATE TABLE test_explode_nested(id int, arr array>) 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>) 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) 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) 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) 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) 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> element type: distinct Arrow layout from array/array ===== + +statement +CREATE TABLE test_explode_arr_map(id int, arr array>) 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) 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) 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>) 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) 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) 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 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..ce582ebe64 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) 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,110 @@ 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, + arr_bi array, + arr_dbl array, + arr_dec array, + arr_bin array, + arr_dt array, + arr_ts array) 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) 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) 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 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..63a8baeb2e 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,50 @@ 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) + } + } + }