comphead commented on code in PR #6428:
URL: https://github.com/apache/datafusion-comet/pull/6428#discussion_r4211438638
##########
native/core/src/execution/planner.rs:
##########
@@ -1254,13 +1257,83 @@ impl PhysicalPlanner {
}
}
+ /// Merge compatible nested types by ordinal, retaining the left field
names and physical
+ /// leaf types. Return None when a metadata-only Spark cast cannot safely
align the layouts.
+ fn positional_nullability_union(left: &DataType, right: &DataType) ->
Option<DataType> {
+ fn merge_field(left: &FieldRef, right: &FieldRef) -> Option<FieldRef> {
+ Some(Arc::new(
+ left.as_ref()
+ .clone()
+ .with_data_type(merge_type(left.data_type(),
right.data_type())?)
+ .with_nullable(left.is_nullable() || right.is_nullable()),
+ ))
+ }
+ fn merge_type(left: &DataType, right: &DataType) -> Option<DataType> {
+ // Comet's Spark cast cannot target a dictionary, even for an
unchanged child.
+ if matches!(left, DataType::Dictionary(_, _))
+ || matches!(right, DataType::Dictionary(_, _))
+ {
+ return None;
+ }
+ if matches!(left, DataType::Null) {
+ return Some(right.clone());
+ }
+ if matches!(right, DataType::Null) {
+ return Some(left.clone());
+ }
+ if left == right {
+ return Some(left.clone());
+ }
+ Some(match (left, right) {
+ (DataType::Struct(left), DataType::Struct(right)) if
left.len() == right.len() => {
+ DataType::Struct(
+ left.iter()
+ .zip(right)
+ .map(|(left, right)| merge_field(left, right))
+ .collect::<Option<_>>()?,
+ )
+ }
+ (DataType::List(left), DataType::List(right)) => {
+ DataType::List(merge_field(left, right)?)
+ }
+ (DataType::Map(left, sorted), DataType::Map(right,
right_sorted))
+ if sorted == right_sorted =>
+ {
+ DataType::Map(merge_field(left, right)?, *sorted)
+ }
+ _ => return None,
+ })
+ }
+ merge_type(left, right)
+ }
+
+ /// Cast nested metadata by ordinal. Arrow's native struct cast matches
field names and can
+ /// silently reorder values when Spark considers differently named structs
compatible.
+ fn cast_to_positional_nested_type(
+ expr: Arc<dyn PhysicalExpr>,
+ from_type: &DataType,
+ target: &DataType,
+ ) -> Arc<dyn PhysicalExpr> {
+ if from_type == target {
+ expr
+ } else {
+ Arc::new(Cast::new(
+ expr,
+ target.clone(),
+ SparkCastOptions::new_without_timezone(EvalMode::Legacy,
false),
+ None,
+ None,
+ ))
+ }
+ }
+
/// DataFusion's nested comparison kernel (`apply_cmp_for_nested`)
requires both operands to
- /// have identical data types, including nested field nullability, whereas
Spark comparisons
- /// ignore nullability. When a comparison's operands are nested types that
differ only in
- /// nullability (e.g. a higher-order `transform` produces `List(non-null
Struct)` while the
- /// other side is `List(nullable Struct)`), cast both to their
nullability-union type so the
- /// kernel accepts them. Non-comparison ops and non-nested or
already-matching types are left
- /// untouched.
+ /// have identical data types, including nested field names and
nullability, whereas Spark
+ /// compares compatible struct fields by ordinal. When operands differ in
nested metadata,
+ /// cast both to the left field names and their nullability-union type so
the kernel accepts
+ /// them without changing field positions. Exact equality is required
here: Arrow's
+ /// `equals_datatype` ignores nested field names, even when nullability
already matches.
+ /// Non-comparison ops and non-nested or already-matching types are left
untouched.
pub fn reconcile_nested_comparison_types(
Review Comment:
`IN` has the same shape and is not covered here. Spark leaves
`named_struct('x', id) IN (named_struct('y', id))` uncast, because
`findTypeForComplex` only unifies struct names that match under the resolver,
and `In.checkInputDataTypes` accepts it in 3.5.9 and 4.0.2 through
`equalsStructurally(..., ignoreNullability = true)`, comparing the fields by
position. Comet sends it to `spark_in_list` without this reconcile step. From
reading the code I expect it to fail, in `validate_types` when a leaf is a
float and otherwise in DataFusion's `InListExpr` type check. Both use
`datatype_is_logically_equal`, which compares struct field names. I have not
run it. Should this PR route `IN` through the same alignment, or should a
follow-up issue track it?
##########
native/core/src/execution/planner.rs:
##########
@@ -4858,6 +5004,135 @@ mod tests {
};
use datafusion_comet_spark_expr::EvalMode;
+ #[test]
+ fn test_if_reconciles_nested_branch_nullability() {
Review Comment:
This test only goes through `ExprStruct::If` and `create_if_expr`, which
this PR does not change, and it feeds bound columns directly, so none of this
PR's code is on its path. #6458 already covers these shapes:
`if_reconciles_struct_field_nullability_and_names` and
`if_reconciles_case_variant_fields_positionally` in `case_when.rs`, and
`conditional/if_nested_nullability.sql`, whose INSERT batches give uniform
THEN, uniform ELSE, NULL and mixed predicates over struct, array, map and
nested containers, including the swapped `x`/`X` case through `to_json`. I
expect this test passes on the PR's base too, but I have not run it. Could it
be dropped (about 130 lines)? The last query in
`create_named_struct_nullability.sql` (three `IF`s over a typed NULL branch) is
on the same unchanged path. Any missing shape would fit in the existing fixture.
##########
spark/src/test/resources/sql-tests/expressions/struct/create_named_struct_positional_comparison.sql:
##########
@@ -0,0 +1,38 @@
+-- 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.
+
+-- MinSparkVersion: 4.0
+
+-- Config: spark.comet.exec.range.enabled=true
+-- Config: spark.comet.sparkToColumnar.enabled=true
+-- Config: spark.comet.sparkToColumnar.supportedOperatorList=Range
+-- Config: spark.sql.caseSensitive=false
Review Comment:
`exec.range.enabled=true` already makes `CometRangeExec` the leaf here.
`CometExecRule` prefers a leaf's enabled Comet operator and keeps the
Spark-to-Arrow conversion only as the fallback for a range the operator
declines, so the two `sparkToColumnar` lines do nothing for these ranges. No
query in this file uses case-variant names, so `caseSensitive=false` does
nothing either. Dropping all three would also make the file fail loudly if
native Range ever declined these ranges, instead of silently running the
equal-nullability premise on a different leaf. In
`create_named_struct_nullability.sql`, `supportedOperatorList=Range` repeats
the default (`Range,InMemoryTableScan,RDDScan,OneRowRelation`).
##########
native/core/src/execution/planner.rs:
##########
@@ -1288,7 +1361,21 @@ impl PhysicalPlanner {
| DataType::Struct(_)
| DataType::Map(_, _)
);
- if !nested || lt.equals_datatype(&rt) {
+ if !nested || lt == rt {
+ return (left, right);
+ }
+ // Catalyst compares struct values by ordinal, even when their field
names differ.
+ // A name-based Arrow cast would change the values being compared.
+ if let Some(target) = Self::positional_nullability_union(<, &rt) {
+ return (
+ Self::cast_to_positional_nested_type(left, <, &target),
+ Self::cast_to_positional_nested_type(right, &rt, &target),
+ );
+ }
+ if matches!(
+ lt,
+ DataType::Struct(_) | DataType::List(_) | DataType::Map(_, _)
+ ) {
return (left, right);
}
Review Comment:
With this early return, every `Struct`, `List` and `Map` pair is either
aligned above or left alone, so the `Field::try_merge` and `CastExpr` code
below only serves `LargeList` and `FixedSizeList`. Comet maps Spark arrays to
`List` only (`serde.rs`), and the list-building kernels I checked (`flatten`,
`shuffle`, `array_insert`) keep their input's offset width, so I could not find
a path from a Spark array to either type. If there is none, the tail and the
two matching arms of `nested` can go, leaving one reconcile path. If there is
one, the tail's `CastExpr` matches struct fields by name, which is the behavior
this PR replaces, so it would need the positional merge too.
##########
spark/src/test/resources/sql-tests/expressions/struct/create_named_struct_positional_comparison.sql:
##########
@@ -0,0 +1,38 @@
+-- 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.
+
+-- MinSparkVersion: 4.0
+
+-- Config: spark.comet.exec.range.enabled=true
+-- Config: spark.comet.sparkToColumnar.enabled=true
+-- Config: spark.comet.sparkToColumnar.supportedOperatorList=Range
+-- Config: spark.sql.caseSensitive=false
+
+-- Spark 4.x compares struct fields by position, irrespective of field names.
+query
+SELECT named_struct('x', CAST(id AS DOUBLE), 'y', CAST(NULL AS DOUBLE)) =
+ named_struct('y', CAST(NULL AS DOUBLE), 'x', CAST(id AS DOUBLE))
+FROM range(8)
Review Comment:
Every operand in this file comes from `range` or a literal, so no compared
struct is NULL, no leaf is NaN or -0.0, and nothing is read from a Parquet
table. A table with `a struct<x: double, y: double>` and `b struct<y: double,
x: double>` holding a NULL struct, a NULL leaf, NaN and -0.0 would cover the
equal-nullability, swapped-name case from the last commit on scan-produced
types, and nested `=` on float leaves has its own path that this would reach.
The `<` and `<=>` queries requested on this file could run over the same rows.
--
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]
