sunchao commented on code in PR #6428:
URL: https://github.com/apache/datafusion-comet/pull/6428#discussion_r4212259778
##########
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> {
Review Comment:
Consolidated this in 6d8a0a912. `positional_common_type` and
`cast_to_common_type` now live in `case_when.rs` and serve IF, CASE,
comparisons, and IN. The planner's `create_case_expr` wrapper is gone.
Conditionals retain leaf-type promotion and combine map sort flags, while
comparisons and IN only align compatible metadata. Both use the same
dictionary-target checks and UTC cast options.
Local validation: 20 native conditional tests and four constructor tests
pass; all 27 conditional/struct SQL-file cases pass on Spark 4.0.4. Focused
Clippy, Rust formatting, Spotless, and Scalastyle pass. Local Spark 3.5/4.1
tests are dependency-DNS blocked; hosted CI is pending.
##########
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)
+
+-- Equal nullability must not bypass field-name alignment. Native Range makes
every field
+-- non-nullable; id=1 compares equal, while id=2 detects an accidental
name-based reorder.
+query expect_native(equalto)
Review Comment:
Added `<` and `<=>` alongside `=` in 6d8a0a912, with explicit
native-expression assertions. The computed-struct query includes id=0 to catch
the wrong ordering result you identified. The fixture also covers Parquet
columns, so it exercises the BinaryExpr path as well as nested floating-point
equality.
Both isolated queries return wrong results with the merge-base native
library and match Spark with this revision.
Local validation: 20 native conditional tests and four constructor tests
pass; all 27 conditional/struct SQL-file cases pass on Spark 4.0.4. Focused
Clippy, Rust formatting, Spotless, and Scalastyle pass. Local Spark 3.5/4.1
tests are dependency-DNS blocked; hosted CI is pending.
##########
spark/src/main/scala/org/apache/comet/serde/structs.scala:
##########
@@ -56,6 +56,8 @@ object CometCreateNamedStruct extends
CometExpressionSerde[CreateNamedStruct] {
val structBuilder = ExprOuterClass.CreateNamedStruct.newBuilder()
structBuilder.addAllValues(valExprs.map(_.get).asJava)
structBuilder.addAllNames(expr.names.map(_.toString).asJava)
+ // Native expressions can conservatively report nullable even when
Catalyst proves otherwise.
+ structBuilder.addAllFieldNullable(expr.valExprs.map(v =>
Boolean.box(v.nullable)).asJava)
Review Comment:
You're right about the original example: `CometCreateArray` already
normalizes it, and the direct `array_array` test did not establish a SQL
regression. I replaced those fixtures with this path, which bypasses
CreateArray:
```sql
SELECT array_union(
array_repeat(named_struct('x', CAST(id AS INT)), 2),
array_repeat(named_struct('x', IF(id = 0, 0, 1)), 2))
FROM range(3)
```
On Spark 4.0.4 with ordinary constant folding and native Range, the PR
matches Spark. Changing only `CreateNamedStruct::fields` back to
`expr.nullable(schema)` fails in array_union because the first struct field is
nullable and the second is not. The revised SQL covers this and its
nested-struct variant. 6d8a0a912 retains `field_nullable`, makes the unit test
check the constructor contract directly, corrects the PR rationale, and updates
the literal-expansion docs.
Local validation: 20 native conditional tests and four constructor tests
pass; all 27 conditional/struct SQL-file cases pass on Spark 4.0.4. Focused
Clippy, Rust formatting, Spotless, and Scalastyle pass. Local Spark 3.5/4.1
tests are dependency-DNS blocked; hosted CI is pending.
##########
spark/src/test/resources/sql-tests/expressions/struct/create_named_struct_nullability.sql:
##########
@@ -0,0 +1,55 @@
+-- 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.
+
+-- Config: spark.comet.sparkToColumnar.enabled=true
+-- Config: spark.comet.sparkToColumnar.supportedOperatorList=Range
+-- Config: spark.sql.caseSensitive=false
+
+-- Catalyst proves BIGINT -> DOUBLE nonnullable; native cast inference is
conservative.
+-- The computed struct must have the same field flags as the typed NULL array
element.
+query
+SELECT array(named_struct('score', CAST(id AS DOUBLE),
+ 'amount', CAST(id AS DECIMAL(18,2))), NULL)
+FROM range(8)
+
+-- The field contract must also survive a nested constructor with an actually
nullable child.
+query
+SELECT array(named_struct('nested', named_struct('score', CAST(id AS DOUBLE),
+ 'optional', CASE WHEN id % 2 = 0 THEN id END)), NULL)
+FROM range(8)
+
+-- CASE unions nullability by ordinal, even when case-insensitive names swap
places.
+query
+SELECT CASE WHEN id % 2 = 0
Review Comment:
Added your map-valued CASE reproducer in 6d8a0a912, plus a variant with an
untyped NULL branch. These replace the two weaker struct-only CASE queries.
They remain ungated for Spark 3.x and run with case sensitivity disabled.
The isolated CASE query fails Arrow validation with the merge-base native
library and matches Spark with this revision.
Local validation: 20 native conditional tests and four constructor tests
pass; all 27 conditional/struct SQL-file cases pass on Spark 4.0.4. Focused
Clippy, Rust formatting, Spotless, and Scalastyle pass. Local Spark 3.5/4.1
tests are dependency-DNS blocked; hosted CI is pending.
##########
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
Review Comment:
Corrected the comment in 6d8a0a912 to distinguish the name-sensitive nested
equality validator from DataFusion's comparison kernel and to explain the old
Arrow cast's positional reordering. One detail from checking the pinned Arrow
implementation: `equals_datatype` ignores struct field names but does compare
nested nullability. The new wording preserves that distinction.
##########
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:
Confirmed the failure on both the merge-base native library and the previous
PR head and included the fix in 6d8a0a912. The planner now computes one common
positional type across the IN value and every candidate before calling
`spark_in_list`. A new SQL fixture covers integer and floating-point paths,
multiple candidates, NOT IN, nested structs/lists, and NULL candidates. It is
ungated because Spark 3.x also accepts differently named structs in IN.
Local validation: 20 native conditional tests and four constructor tests
pass; all 27 conditional/struct SQL-file cases pass on Spark 4.0.4. Focused
Clippy, Rust formatting, Spotless, and Scalastyle pass. Local Spark 3.5/4.1
tests are dependency-DNS blocked; hosted CI is pending.
##########
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:
Removed the old Field::try_merge/CastExpr tail and the
LargeList/FixedSizeList admission arms in 6d8a0a912. Spark's supported nested
representations now go through the shared positional helper, with incompatible
physical layouts left unchanged. There is no remaining name-based fallback in
this comparison reconciliation.
--
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]
