From 4ad941120d1cb57be72ef2b5fd7e416df394d32a Mon Sep 17 00:00:00 2001 From: Liang-Chi Hsieh Date: Sun, 6 Sep 2026 20:21:21 -0700 Subject: [PATCH] fix: apply the parent struct's null mask before hashing its fields Arrow keeps a `StructArray`'s children validity independent of the parent's, so at a row where the struct itself is null a child buffer can still hold a value. The struct branch of `create_hashes_internal!` recursed straight into `struct_array.columns()` without consulting the parent's null buffer, so a null struct hashed whatever happened to sit in the child slot instead of leaving the seed alone as Spark does. Two rows with the same logical key then hash differently. What the tests here demonstrate directly is a wrong answer from `hash` and `xxhash64` in a Spark query. The same hash decides shuffle partition assignment, so equal keys reaching different partitions follows from it, but that path is not exercised by these tests. The `List` and `Map` branches already guard on `is_null`; only `Struct` did not. Uses `StructArray::flatten`, which unions the parent's nulls into each child exactly the way #4432 fixed the same null-mask propagation problem in `GetStructField`, and does it without revalidating the child data buffers: the union only ever adds nulls, so the buffers are unchanged. Going through the checked `ArrayData` builder instead would rescan every child buffer on each call -- for a string child, the whole UTF-8 values buffer -- and this branch runs once per element when hashing a list of structs. `flatten` is called only when the parent actually has a null, because it also builds a fresh `Fields` with every non-nullable field re-marked nullable, which this call site discards. A struct carrying an all-valid null buffer, which is what slicing leaves behind, would otherwise pay a `Vec` and an `Arc<[FieldRef]>` for nothing. `NullBuffer` stores its null count, so the test itself is free. The branch is shared by murmur3 and xxhash64 through the macro, so both are fixed and both get a regression test. Each test fails without the change: the null row hashes 3319311472 rather than the 42 seed under murmur3. The unit tests hand-build the array, so they do not show that a query can reach this shape. It can: a nullable struct whose child field is REQUIRED is written as `optional group c { required int32 a; }`, and on read the child leaf has nowhere to record a null of its own, so its buffer holds a value at exactly the rows where the struct is null. Two end-to-end tests in `CometHashExpressionSuite` cover that, one with a scalar child and one where the child is itself a struct so the union has to recurse, and both disagree with Spark before the change. The per-element route through `hash_list_array!` gets an end-to-end test too. Using the struct as the list element directly does not reproduce, because a null element is rebuilt on the way into the array and the hidden child values go with it. Wrapping it does: `array(named_struct('tag', 1, 'b', c))` yields an element that is itself valid, so it is copied rather than rebuilt, and the null `c` inside keeps the values Parquet wrote under it. Covered with one element and with two, so the chaining between elements is exercised, and both disagree with Spark before the change. Co-authored-by: Claude Code --- native/spark-expr/src/hash_funcs/murmur3.rs | 85 ++++++++++++++++ native/spark-expr/src/hash_funcs/utils.rs | 26 ++++- native/spark-expr/src/hash_funcs/xxhash64.rs | 66 +++++++++++++ .../comet/CometHashExpressionSuite.scala | 97 ++++++++++++++++++- 4 files changed, 271 insertions(+), 3 deletions(-) diff --git a/native/spark-expr/src/hash_funcs/murmur3.rs b/native/spark-expr/src/hash_funcs/murmur3.rs index 233097ffc15..a9e5b67aa9f 100644 --- a/native/spark-expr/src/hash_funcs/murmur3.rs +++ b/native/spark-expr/src/hash_funcs/murmur3.rs @@ -302,6 +302,91 @@ mod tests { ); } + /// Arrow lets a `StructArray`'s children carry their own validity, so at a row where the struct + /// itself is null the child buffer can still hold a value. Spark hashes a null struct as the + /// seed, so those hidden child values must not reach the hash. This is the same null-mask + /// propagation problem that #4432 fixed for `GetStructField`. + #[test] + fn test_null_struct_ignores_hidden_child_values() { + use arrow::array::{Int32Array, StructArray}; + use arrow::buffer::NullBuffer; + use arrow::datatypes::{DataType, Field, Fields}; + + let fields: Fields = vec![Arc::new(Field::new("a", DataType::Int32, true))].into(); + // Row 1 is a null struct whose child still holds 999. + let child: ArrayRef = Arc::new(Int32Array::from(vec![Some(1), Some(999)])); + let nulls = NullBuffer::from(vec![true, false]); + let with_hidden: ArrayRef = Arc::new(StructArray::new( + fields.clone(), + vec![Arc::clone(&child)], + Some(nulls.clone()), + )); + // Same shape, but the hidden slot is null too. + let child_null: ArrayRef = Arc::new(Int32Array::from(vec![Some(1), None])); + let without_hidden: ArrayRef = + Arc::new(StructArray::new(fields, vec![child_null], Some(nulls))); + + let mut a = vec![42u32; 2]; + create_murmur3_hashes(&[with_hidden], &mut a).unwrap(); + let mut b = vec![42u32; 2]; + create_murmur3_hashes(&[without_hidden], &mut b).unwrap(); + + assert_eq!( + a, b, + "a null struct must hash the same regardless of what its child buffer holds" + ); + assert_eq!(a[1], 42, "a null struct must leave the seed untouched"); + } + + /// The struct branch is also reached once per element when hashing `array>`, which + /// is the path #5567 made usable as a shuffle partitioning key. The test above hashes a struct + /// directly, so it does not cover that route. + /// + /// Here the null is the list *element* itself, with valid elements either side so the chaining + /// is exercised. The end-to-end test in `CometHashExpressionSuite` covers the other shape, a + /// valid element wrapping a null struct, which is the one a query can produce. + #[test] + fn test_null_struct_element_of_list_ignores_hidden_child_values() { + use arrow::array::{Int32Array, ListArray, StructArray}; + use arrow::buffer::{NullBuffer, OffsetBuffer}; + use arrow::datatypes::{DataType, Field, Fields}; + + let fields: Fields = vec![Arc::new(Field::new("a", DataType::Int32, true))].into(); + // Element 1 is a null struct whose child still holds 999; elements 0 and 2 are valid. + let element_nulls = NullBuffer::from(vec![true, false, true]); + let with_hidden: ArrayRef = Arc::new(StructArray::new( + fields.clone(), + vec![Arc::new(Int32Array::from(vec![Some(1), Some(999), Some(3)])) as ArrayRef], + Some(element_nulls.clone()), + )); + // The same shape with the hidden slot null as well. + let without_hidden: ArrayRef = Arc::new(StructArray::new( + fields, + vec![Arc::new(Int32Array::from(vec![Some(1), None, Some(3)])) as ArrayRef], + Some(element_nulls), + )); + + // One row holding all three elements, so element hashes chain in order. + let as_list = |elements: ArrayRef| -> ArrayRef { + Arc::new(ListArray::new( + Arc::new(Field::new("item", elements.data_type().clone(), true)), + OffsetBuffer::new(vec![0i32, 3].into()), + elements, + None, + )) + }; + + let mut from_hidden = vec![42u32; 1]; + create_murmur3_hashes(&[as_list(with_hidden)], &mut from_hidden).unwrap(); + let mut from_null = vec![42u32; 1]; + create_murmur3_hashes(&[as_list(without_hidden)], &mut from_null).unwrap(); + + assert_eq!( + from_hidden, from_null, + "a null struct element must hash the same regardless of its child buffer" + ); + } + #[test] fn test_i8() { test_murmur3_hash::( diff --git a/native/spark-expr/src/hash_funcs/utils.rs b/native/spark-expr/src/hash_funcs/utils.rs index 18e5a41bc75..4186751f921 100644 --- a/native/spark-expr/src/hash_funcs/utils.rs +++ b/native/spark-expr/src/hash_funcs/utils.rs @@ -828,8 +828,30 @@ macro_rules! create_hashes_internal { } DataType::Struct(_) => { let struct_array = col.as_any().downcast_ref::().unwrap(); - // Hash each field of the struct - Spark hashes all fields recursively - let columns: Vec = struct_array.columns().to_vec(); + // Hash each field of the struct - Spark hashes all fields recursively. + // + // Arrow keeps a struct's children validity independent of the parent's, so at a + // row where the struct is null a child buffer can still hold a value. Spark + // hashes a null struct as the seed, so the parent's nulls have to be pushed + // into each child before recursing, the same way #4432 fixed `GetStructField`. + // Without it a null struct hashes whatever happens to sit in the child slot. + // `flatten` does exactly this union, and skips revalidating the child data + // buffers: it only ever adds nulls, so the buffers themselves are unchanged. + // Rebuilding them through the checked builder would rescan every child buffer + // (for a string child, the whole UTF-8 values buffer) on each call, and this + // branch is reached once per element when hashing a list of structs. + // + // Only call it when there is actually a null to push down. `flatten` returns + // early when there is no null buffer at all, but with a buffer present it + // builds a fresh `Fields` with every non-nullable field re-marked nullable, + // which this call site discards. So the case worth skipping is a buffer that + // is present and all-valid -- what slicing leaves behind -- which would + // otherwise pay a `Vec` and an `Arc<[FieldRef]>` for nothing. `NullBuffer` + // caches its null count, so the test itself is O(1). + let columns: Vec = match struct_array.nulls() { + Some(nulls) if nulls.null_count() > 0 => struct_array.flatten().1, + _ => struct_array.columns().to_vec(), + }; if !columns.is_empty() { $recursive_hash_method(&columns, $hashes_buffer)?; } diff --git a/native/spark-expr/src/hash_funcs/xxhash64.rs b/native/spark-expr/src/hash_funcs/xxhash64.rs index 7009fc99c2c..93ac9304fc8 100644 --- a/native/spark-expr/src/hash_funcs/xxhash64.rs +++ b/native/spark-expr/src/hash_funcs/xxhash64.rs @@ -208,6 +208,72 @@ mod tests { assert_eq!(from_dict, from_decoded); } + /// The struct branch is shared with murmur3 through `create_hashes_internal!`, so the parent + /// null mask has to reach xxhash64's children too. See #4432 for the same problem in + /// `GetStructField`. + #[test] + fn test_null_struct_ignores_hidden_child_values() { + use arrow::array::StructArray; + use arrow::buffer::NullBuffer; + use arrow::datatypes::{DataType, Field, Fields}; + + let fields: Fields = vec![Arc::new(Field::new("a", DataType::Int32, true))].into(); + let nulls = NullBuffer::from(vec![true, false]); + let hidden: ArrayRef = Arc::new(StructArray::new( + fields.clone(), + vec![Arc::new(Int32Array::from(vec![Some(1), Some(999)])) as ArrayRef], + Some(nulls.clone()), + )); + let plain: ArrayRef = Arc::new(StructArray::new( + fields, + vec![Arc::new(Int32Array::from(vec![Some(1), None])) as ArrayRef], + Some(nulls), + )); + + let mut a = vec![42u64; 2]; + create_xxhash64_hashes(&[hidden], &mut a).unwrap(); + let mut b = vec![42u64; 2]; + create_xxhash64_hashes(&[plain], &mut b).unwrap(); + assert_eq!(a, b, "a null struct must hash the same either way"); + assert_eq!(a[1], 42, "a null struct must leave the seed untouched"); + } + + /// Companion to the murmur3 case: the struct branch is shared through the macro, so the + /// per-element route needs covering here too. + #[test] + fn test_null_struct_element_of_list_ignores_hidden_child_values() { + use arrow::array::{ListArray, StructArray}; + use arrow::buffer::{NullBuffer, OffsetBuffer}; + use arrow::datatypes::{DataType, Field, Fields}; + + let fields: Fields = vec![Arc::new(Field::new("a", DataType::Int32, true))].into(); + let element_nulls = NullBuffer::from(vec![true, false, true]); + let with_hidden: ArrayRef = Arc::new(StructArray::new( + fields.clone(), + vec![Arc::new(Int32Array::from(vec![Some(1), Some(999), Some(3)])) as ArrayRef], + Some(element_nulls.clone()), + )); + let without_hidden: ArrayRef = Arc::new(StructArray::new( + fields, + vec![Arc::new(Int32Array::from(vec![Some(1), None, Some(3)])) as ArrayRef], + Some(element_nulls), + )); + let as_list = |elements: ArrayRef| -> ArrayRef { + Arc::new(ListArray::new( + Arc::new(Field::new("item", elements.data_type().clone(), true)), + OffsetBuffer::new(vec![0i32, 3].into()), + elements, + None, + )) + }; + + let mut from_hidden = vec![42u64; 1]; + create_xxhash64_hashes(&[as_list(with_hidden)], &mut from_hidden).unwrap(); + let mut from_null = vec![42u64; 1]; + create_xxhash64_hashes(&[as_list(without_hidden)], &mut from_null).unwrap(); + assert_eq!(from_hidden, from_null); + } + #[test] fn test_i8() { test_xxhash64_hash::( diff --git a/spark/src/test/scala/org/apache/comet/CometHashExpressionSuite.scala b/spark/src/test/scala/org/apache/comet/CometHashExpressionSuite.scala index 23a539fa1eb..c997da0012f 100644 --- a/spark/src/test/scala/org/apache/comet/CometHashExpressionSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometHashExpressionSuite.scala @@ -21,8 +21,9 @@ package org.apache.comet import scala.util.Random -import org.apache.spark.sql.CometTestBase +import org.apache.spark.sql.{CometTestBase, Row} import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper +import org.apache.spark.sql.types.{IntegerType, StructField, StructType} import org.apache.comet.testing.{DataGenOptions, FuzzDataGenerator, ParquetGenerator, SchemaGenOptions} @@ -256,6 +257,100 @@ class CometHashExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelpe } } + test("hash - null struct with a required child field") { + // `c` is nullable but its child is REQUIRED, so Spark writes + // `optional group c { required int32 a; }`. On read the child leaf has nowhere to record a + // null of its own, so its buffer holds a value at exactly the rows where the struct is null. + // That is the shape where the parent's null mask has to reach the children; a struct built in + // the plan, or one whose child is also nullable, has null children there and hides the bug. + withTempPath { dir => + val schema = StructType( + Seq( + StructField( + "c", + StructType(Seq(StructField("a", IntegerType, nullable = false))), + nullable = true))) + withSQLConf(CometConf.COMET_ENABLED.key -> "false") { + val rows = Seq(Row(Row(1)), Row(null), Row(Row(3)), Row(null)) + spark + .createDataFrame(spark.sparkContext.parallelize(rows), schema) + .coalesce(1) + .write + .parquet(dir.toString) + } + spark.read.parquet(dir.toString).createOrReplaceTempView("null_struct_t") + checkSparkAnswerAndOperator("SELECT hash(c), xxhash64(c) FROM null_struct_t ORDER BY 1, 2") + } + } + + test("hash - null struct whose child is itself a struct") { + // The union has to recurse: the outer struct's nulls reach the inner struct, whose own + // children are required and so carry values under the null. + withTempPath { dir => + val inner = StructType(Seq(StructField("x", IntegerType, nullable = false))) + val schema = StructType( + Seq( + StructField( + "c", + StructType(Seq(StructField("b", inner, nullable = false))), + nullable = true))) + withSQLConf(CometConf.COMET_ENABLED.key -> "false") { + val rows = Seq(Row(Row(Row(1))), Row(null), Row(Row(Row(3))), Row(null)) + spark + .createDataFrame(spark.sparkContext.parallelize(rows), schema) + .coalesce(1) + .write + .parquet(dir.toString) + } + spark.read.parquet(dir.toString).createOrReplaceTempView("null_nested_struct_t") + checkSparkAnswerAndOperator( + "SELECT hash(c), xxhash64(c) FROM null_nested_struct_t ORDER BY 1, 2") + } + } + + test("hash - list element wrapping a null struct with a required child field") { + // The per-element path in `hash_list_array!`, which #5567 made usable as a shuffle key. + // + // Wrapping the struct rather than using it as the element directly is what makes this + // reachable: `array(named_struct('tag', 1, 'b', c))` produces an element that is itself + // valid, so it is copied rather than rebuilt, and the null `c` inside it keeps the child + // values Parquet wrote under it. Using `c` as the element directly does not reproduce, + // because a null element is rebuilt on the way in and the hidden values go with it. + // + // So this covers a null struct nested inside a valid element. The unit test in murmur3.rs + // covers the complementary shape, where the element itself is the null struct. + withTempPath { dir => + val schema = StructType( + Seq( + StructField( + "c", + StructType(Seq(StructField("a", IntegerType, nullable = false))), + nullable = true))) + withSQLConf(CometConf.COMET_ENABLED.key -> "false") { + val rows = Seq(Row(Row(1)), Row(null), Row(Row(3)), Row(null)) + spark + .createDataFrame(spark.sparkContext.parallelize(rows), schema) + .coalesce(1) + .write + .parquet(dir.toString) + } + spark.read.parquet(dir.toString).createOrReplaceTempView("wrapped_null_struct_t") + + checkSparkAnswerAndOperator(""" + SELECT + hash(array(named_struct('tag', 1, 'b', c))), + xxhash64(array(named_struct('tag', 1, 'b', c))) + FROM wrapped_null_struct_t ORDER BY 1, 2""") + + // Two elements, so the hash of the second chains onto the first. + checkSparkAnswerAndOperator(""" + SELECT + hash(array(named_struct('tag', 1, 'b', c), named_struct('tag', 2, 'b', c))), + xxhash64(array(named_struct('tag', 1, 'b', c), named_struct('tag', 2, 'b', c))) + FROM wrapped_null_struct_t ORDER BY 1, 2""") + } + } + test("hash - struct with array field") { withTable("t") { sql("CREATE TABLE t(c STRUCT>) USING parquet")