mbutrovich commented on code in PR #5192: URL: https://github.com/apache/datafusion-comet/pull/5192#discussion_r3713126024
########## 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<T>` input as null. +/// Bridges DataFusion's `UnnestExec` (which drops empty rows under +/// `preserve_nulls=true`) to Spark's `explode_outer`/`posexplode_outer` +/// semantics. See <https://github.com/apache/datafusion/issues/19053>. +#[derive(Debug, Clone)] +pub struct ListEmptyToNullExpr { + child: Arc<dyn PhysicalExpr>, +} + +impl ListEmptyToNullExpr { + pub fn new(child: Arc<dyn PhysicalExpr>) -> Self { + Self { child } + } +} + +impl Display for ListEmptyToNullExpr { + fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { + write!(f, "list_empty_to_null({})", self.child) + } +} + +impl PartialEq for ListEmptyToNullExpr { + fn eq(&self, other: &Self) -> bool { + self.child.eq(&other.child) + } +} + +impl Eq for ListEmptyToNullExpr {} + +impl Hash for ListEmptyToNullExpr { + fn hash<H: Hasher>(&self, state: &mut H) { + self.child.hash(state); + } +} + +impl PhysicalExpr for ListEmptyToNullExpr { + fn fmt_sql(&self, f: &mut Formatter<'_>) -> std::fmt::Result { + Display::fmt(self, f) + } + + fn return_field(&self, input_schema: &Schema) -> DataFusionResult<FieldRef> { + // Preserve the child field's name and element type; force the outer + // list to nullable because we mark empty rows as null. + let child_field = self.child.return_field(input_schema)?; + Ok(Arc::new(Field::new( + child_field.name(), + child_field.data_type().clone(), + true, + ))) + } + + fn evaluate(&self, batch: &RecordBatch) -> DataFusionResult<ColumnarValue> { + let value = self.child.evaluate(batch)?; + let array = value.into_array(batch.num_rows())?; + + let Some(list) = array.as_any().downcast_ref::<ListArray>() else { + return exec_err!( + "ListEmptyToNullExpr expected List input, got {}", + array.data_type() + ); + }; + + let offsets = list.offsets(); + let len = list.len(); + let existing_nulls = list.nulls(); + + // Fast path: no currently-valid row is empty, so the input already + // satisfies outer semantics. `is_valid` returns true when `nulls` is + // `None`, so this single scan short-circuits on the first empty + // valid row without allocating. + let has_valid_empty = (0..len) + .any(|i| offsets[i + 1] == offsets[i] && existing_nulls.is_none_or(|n| n.is_valid(i))); + if !has_valid_empty { + return Ok(ColumnarValue::Array(Arc::clone(&array))); + } + + let non_empty = BooleanBuffer::collect_bool(len, |i| offsets[i + 1] > offsets[i]); + let 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), + )?; Review Comment: The slow path here (`non_empty` via `BooleanBuffer::collect_bool`, then `existing.inner() & &non_empty`, then `ListArray::try_new`) is what `arrow::compute::nullif` already does in one pass. I flagged the double allocation in my first-pass review; pointing at the actual kernel now instead of just the diagnosis. `nullif(left: &dyn Array, right: &BooleanArray)` in arrow-select 58.4.0 (the version pinned at native/Cargo.toml:37) computes `left_null_bitmap & !right` with `bitwise_bin_op_helper`, a single word-level pass, then rebuilds the array via `left_data.into_builder().nulls(Some(nulls))` and `build_unchecked()`, reusing the existing offsets and values buffers without a manual `ListArray::try_new`. Source: `arrow-select-58.4.0/src/nullif.rs:44-112`. Sketch: ```rust let is_empty = BooleanArray::from( (0..len).map(|i| offsets[i + 1] == offsets[i]).collect::<Vec<_>>() ); let result = nullif(array.as_ref(), &is_empty)?; Ok(ColumnarValue::Array(result)) ``` That drops the `existing_nulls` match, the `DataType::List` destructure, and `ListArray::try_new` entirely, since `nullif` already folds "already null" and "now empty" into one null-bitmap combination. Worth trying before adding a second manual optimization pass on top of the current approach. ########## native/core/src/execution/planner.rs: ########## @@ -1897,14 +1898,74 @@ 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( "Explode operator requires a child expression".to_string(), )); }; + 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). Review Comment: The `(TODO: link the Comet tracking issue here)` placeholder can point at #5210 now, not a hypothetical future issue. DataFusion closed #19053 via apache/datafusion#22100 ("Add support for `unnest_outer` function for arrays"), merged 2026-07-31, the day before this PR opened. That PR replaces `UnnestOptions.preserve_nulls: bool` with `datafusion::common::unnest::NullHandling { Drop, Preserve, PreserveAndExpandEmpty }`. `NullHandling::PreserveAndExpandEmpty` is this exact bridge: it treats an empty list identically to a NULL list. That release isn't in datafusion 54.1.0 (native/Cargo.toml:41) yet, so it doesn't help this PR today, and you already filed #5210 for the follow-up on 2026-08-02. Once Comet moves to a datafusion release carrying #22100, `ListEmptyToNullExpr`, the whole `(true, true)` / `(true, false)` pre-projection match here, and `preserve_nulls: explode.outer` at line 2051 collapse to: ```rust null_handling: if explode.outer { NullHandling::PreserveAndExpandEmpty } else { NullHandling::Drop }, ``` Linking #5210 here instead of the placeholder answers @andygrove's "does this TODO still need to be addressed" thread directly. ########## native/core/src/execution/expressions/list_positions.rs: ########## @@ -105,10 +116,16 @@ impl PhysicalExpr for ListPositionsExpr { } } Review Comment: Given this morning's benchmark: `explode_outer` is 1.4x-1.9x over Spark, `posexplode_outer` is at parity, and the delta between the two on the same data (122ms vs 169ms mixed, 89-95ms vs 129-133ms dense) points at this loop. Before filing the follow-up issue you and @andygrove discussed, worth a quick check of whether replacing the inner `for i in 0..(end - start) { values.push(i) }` with `values.extend(0..(end - start))` moves anything. `Vec::push` re-checks `len == capacity` every call even though `with_capacity` already reserved the exact size; `extend` from a `Range` doesn't carry that per-element branch. Five minutes to check before opening the issue, and if it's noise, the issue can say so with a number attached instead of a guess. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
