This is an automated email from the ASF dual-hosted git repository.
andygrove pushed a commit to branch branch-1.0
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
The following commit(s) were added to refs/heads/branch-1.0 by this push:
new 0de93c30c4 fix: apply the parent struct's null mask before hashing its
fields (#5754) (#5823)
0de93c30c4 is described below
commit 0de93c30c4fe04672ba9833e613e19006990b3c0
Author: Andy Grove <[email protected]>
AuthorDate: Wed Sep 9 21:47:17 2026 -0600
fix: apply the parent struct's null mask before hashing its fields (#5754)
(#5823)
---
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 +++++++++++++++
.../apache/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 233097ffc1..a9e5b67aa9 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<struct<..>>`, 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::<i8, Int8Array>(
diff --git a/native/spark-expr/src/hash_funcs/utils.rs
b/native/spark-expr/src/hash_funcs/utils.rs
index 18e5a41bc7..4186751f92 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::<StructArray>().unwrap();
- // Hash each field of the struct - Spark hashes all fields
recursively
- let columns: Vec<ArrayRef> =
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<ArrayRef> = 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 7009fc99c2..93ac9304fc 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::<i8, Int8Array>(
diff --git
a/spark/src/test/scala/org/apache/comet/CometHashExpressionSuite.scala
b/spark/src/test/scala/org/apache/comet/CometHashExpressionSuite.scala
index 23a539fa1e..c997da0012 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<a: INT, b: ARRAY<STRING>>) USING parquet")
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
