mbutrovich commented on code in PR #5192:
URL: https://github.com/apache/datafusion-comet/pull/5192#discussion_r3705582648
##########
native/core/src/execution/planner.rs:
##########
@@ -1926,22 +1986,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)));
Review Comment:
This is the call site that runs `ListPositionsExpr` on the wrapped array.
Per the discussion on #5224, `ListPositionsExpr::evaluate` reuses the input's
original offset buffer against a freshly zero based values array, and that
panics when the array has a non zero base offset, for example after
`GlobalLimitExec` with a non zero skip.
Before this PR `posexplode_outer` fell back to Spark, so that path was
unreachable by default. This PR removes the `allowIncompatible` gate, so it
becomes reachable by default here.
Can we either pull in the offset rebasing fix from #5224, or add a
regression test marked `ignore` referencing #5224, before this merges?
##########
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:
This TODO is still a placeholder: `(TODO: link the Comet tracking issue
here)`. Can you file the tracking issue for removing `ListEmptyToNullExpr` and
this pre-projection once datafusion#19053 lands, and link it here?
##########
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,
Review Comment:
When `existing_nulls` is `Some`, this allocates `non_empty` and then
allocates a second buffer for `existing.inner() & &non_empty`. Both conditions
are already known per row from the `has_valid_empty` scan above. Can this
collapse into one `collect_bool` pass:
```rust
let combined = BooleanBuffer::collect_bool(len, |i| {
offsets[i + 1] > offsets[i] && existing_nulls.is_none_or(|n|
n.is_valid(i))
});
```
That drops one allocation on the path where the input already carries a null
bitmap.
I also looked at `NullBuffer::union`/`union_many` (arrow-rs #9692) as an
alternative. They don't help here, since one side of the combination is a
predicate rather than an existing null buffer, so materializing it first just
to union it costs the same allocation this avoids.
--
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]
