viirya commented on code in PR #6428: URL: https://github.com/apache/datafusion-comet/pull/6428#discussion_r4210414649
########## 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: I can confirm both on the merge base `98662215d` with Spark 4.1. `named_struct('x', CAST(id AS DOUBLE), 'y', 1D) < named_struct('y', 1D, 'x', CAST(id AS DOUBLE))` and the same query with `<=>` both return a wrong row on main. With this PR both match Spark, so adding them to the fixture would lock that in. ########## 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: Could you share a query that fails on main without the `field_nullable` change? All five queries in `create_named_struct_nullability.sql` pass natively on the merge base `98662215d`, including the `array(named_struct(...), NULL)` example from the description. `CometCreateArray` already casts every element to `deepNullable(...)` (#5452), so `make_array` never sees the mismatch. I also ran this PR with `CreateNamedStruct::fields` reverted to `expr.nullable(schema)`. Both new fixtures and a set of comparison, CASE, IF, array and aggregate probes still matched Spark. Only `test_create_struct_preserves_catalyst_nullability_in_array` failed, and it calls `array_array` directly without that cast. The change does add a failure mode. A native child that emits a null where Catalyst proves non-null now fails the task in `StructArray::try_new` with `Found unmasked nulls for non-nullable StructArray field`. Unless there is a query that needs it, could this half be dropped or moved to its own PR with a test that fails without it? If it stays, the `expandComplexLiteral` doc @andygrove mentioned needs updating too. ########## 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: The doc says `apply_cmp_for_nested` requires identical types including nested field names and nullability. In DataFusion 55 it checks `equals_datatype`, which ignores both. The constraints that actually bite are `NestedPredicate`'s `validate_types` for `=`/`<>`, which compares names through `datatype_is_logically_equal`, and the name-matching arrow `CastExpr` in the old fallback. On main, `<` and `<=>` returned wrong rows only because that cast reordered the fields. Could the comment describe those two? Otherwise the next reader will "fix" the wrong thing. ########## 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: The two `CASE` queries here also pass on main, so they don't guard the new `create_case_expr` alignment. This query panics on main with `Found unmasked nulls for non-nullable StructArray field "x"` and matches Spark with this PR. It is valid on Spark 3.x as well with `spark.sql.caseSensitive=false`: ```sql SELECT CASE WHEN id % 2 = 0 THEN map(1, named_struct('x', CAST(id AS DOUBLE), 'X', CAST(NULL AS DOUBLE))) ELSE map(1, named_struct('X', CAST(NULL AS DOUBLE), 'x', CAST(id AS DOUBLE))) END FROM range(4) ``` Could you add it, or something like it, alongside the existing ones? -- 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]
