You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/10/18 15:52:36 UTC
(datafusion-comet) 01/01: fix: handle NullType struct fields in JVM columnar shuffle (#7099)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-7099-d60cc1dd5fe1223309ee1607811951b1dcd775cb
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
commit 4ef103624e6f5f87072ae1e46b85541fb559fda1
Author: Jaden Fix <16...@users.noreply.github.com>
AuthorDate: Sun Oct 18 15:52:09 2026 +0000
fix: handle NullType struct fields in JVM columnar shuffle (#7099)
The field-major struct conversion in the JVM columnar shuffle's
row-to-Arrow path (`append_struct_fields_field_major` for a top-level
struct column and `append_nested_struct_fields_field_major` for a struct
inside a struct) had no `DataType::Null` arm, so a struct such as
`named_struct('x', a, 'n', null)` hit `unreachable!` and failed the
query. `append_field` already handled `Null` by appending a null per
row; both functions now do the same.
Adds Rust unit tests that lay out Spark UnsafeRows for a top-level and a
nested struct with a NullType field, and a Scala regression test in
CometColumnarShuffleSuite covering round-robin and hash partitioning.
---
native/shuffle/src/spark_unsafe/row.rs | 132 +++++++++++++++++++++
.../comet/exec/CometColumnarShuffleSuite.scala | 17 +++
2 files changed, 149 insertions(+)
diff --git a/native/shuffle/src/spark_unsafe/row.rs b/native/shuffle/src/spark_unsafe/row.rs
index 260dca4a2e..219f09131e 100644
--- a/native/shuffle/src/spark_unsafe/row.rs
+++ b/native/shuffle/src/spark_unsafe/row.rs
@@ -565,6 +565,13 @@ fn append_nested_struct_fields_field_major(
nested_fields,
)?;
}
+ // Every value of a NullType field is null, so there is nothing to read from the row.
+ DataType::Null => {
+ let field_builder = get_field_builder!(struct_builder, NullBuilder, field_idx);
+ for _ in 0..num_rows {
+ field_builder.append_null();
+ }
+ }
// For list and map, fall back to append_field since they have variable-length elements
dt @ (DataType::List(_) | DataType::Map(_, _)) => {
for row_idx in 0..num_rows {
@@ -1026,6 +1033,13 @@ fn append_struct_fields_field_major(
nested_fields,
)?;
}
+ // Every value of a NullType field is null, so there is nothing to read from the row.
+ DataType::Null => {
+ let field_builder = get_field_builder!(struct_builder, NullBuilder, field_idx);
+ for _ in row_start..row_end {
+ field_builder.append_null();
+ }
+ }
// For list and map, fall back to append_field since they have variable-length elements
dt @ (DataType::List(_) | DataType::Map(_, _)) => {
for (row_idx, i) in (row_start..row_end).enumerate() {
@@ -1686,4 +1700,122 @@ mod test {
lists.iter().map(|l| l.map(<[_]>::to_vec)).collect();
assert_eq!(appended, expected);
}
+
+ /// Lays out an 8-byte-aligned `UnsafeRow` with one struct field: the null bitset, the field's
+ /// offset and size, then the nested `UnsafeRow` holding `fields`. A one-word field is stored
+ /// in its slot, a longer one in the variable-length region, and `None` is written as null, as
+ /// Spark writes every `NullType` value. `None` for `fields` makes the struct itself null.
+ fn unsafe_row_with_struct(fields: Option<&[Option<Vec<u64>>]>) -> Vec<u64> {
+ let Some(fields) = fields else {
+ return vec![1, 0];
+ };
+ assert!(fields.len() <= 64);
+ let mut nested = vec![0u64; 1 + fields.len()];
+ let mut variable: Vec<u64> = Vec::new();
+ for (i, field) in fields.iter().enumerate() {
+ match field {
+ None => nested[0] |= 1 << i,
+ Some(value) if value.len() == 1 => nested[1 + i] = value[0],
+ Some(value) => {
+ let offset = 8 * (nested.len() + variable.len()) as u64;
+ nested[1 + i] = (offset << 32) | (8 * value.len()) as u64;
+ variable.extend(value);
+ }
+ }
+ }
+ nested.extend(variable);
+ let mut row = vec![0, (16 << 32) | (8 * nested.len()) as u64];
+ row.extend(nested);
+ row
+ }
+
+ fn struct_column(data_type: &DataType, rows: &[Vec<u64>]) -> ArrayRef {
+ let mut row_addresses: Vec<jlong> = rows.iter().map(|r| r.as_ptr() as jlong).collect();
+ let mut row_sizes: Vec<jint> = rows.iter().map(|r| (r.len() * 8) as jint).collect();
+ let schema = [data_type.clone()];
+ let mut builder = make_builders(data_type, rows.len(), 1.0).unwrap();
+ append_columns(
+ row_addresses.as_mut_ptr(),
+ row_sizes.as_mut_ptr(),
+ 0,
+ rows.len(),
+ &schema,
+ 0,
+ &mut builder,
+ 1.0,
+ )
+ .unwrap();
+ builder.finish()
+ }
+
+ // `named_struct('x', a, 'n', null)` reaches the field-major struct conversion with a
+ // `DataType::Null` field, which it has to accept like `append_field` does.
+ #[test]
+ fn struct_column_with_null_type_field() {
+ let fields = Fields::from(vec![
+ Field::new("x", DataType::Int32, true),
+ Field::new("n", DataType::Null, true),
+ ]);
+ let data_type = DataType::Struct(fields.clone());
+ let rows = [
+ unsafe_row_with_struct(Some(&[Some(vec![7]), None])),
+ unsafe_row_with_struct(None),
+ unsafe_row_with_struct(Some(&[None, None])),
+ ];
+
+ let array = struct_column(&data_type, &rows);
+
+ assert_eq!(array.data_type(), &data_type);
+ let st = array.as_struct();
+ assert_eq!(st.len(), 3);
+ assert_eq!(st.null_count(), 1);
+ assert!(st.is_null(1));
+ let x: Vec<Option<i32>> = st.column(0).as_primitive::<Int32Type>().iter().collect();
+ assert_eq!(x, vec![Some(7), None, None]);
+ let n = st.column(1);
+ assert_eq!(n.data_type(), &DataType::Null);
+ assert_eq!(n.len(), 3);
+ assert_eq!(n.logical_null_count(), 3);
+ }
+
+ // `named_struct('s', named_struct('x', a, 'n', null))` reaches the nested variant.
+ #[test]
+ fn nested_struct_column_with_null_type_field() {
+ let inner_fields = Fields::from(vec![
+ Field::new("x", DataType::Int32, true),
+ Field::new("n", DataType::Null, true),
+ ]);
+ let fields = Fields::from(vec![Field::new(
+ "s",
+ DataType::Struct(inner_fields.clone()),
+ true,
+ )]);
+ let data_type = DataType::Struct(fields.clone());
+ let inner = |x: Option<u64>| {
+ // Drop the outer row's bitset and slot: the nested row starts at its own bitset.
+ unsafe_row_with_struct(Some(&[x.map(|x| vec![x]), None]))[2..].to_vec()
+ };
+ let rows = [
+ unsafe_row_with_struct(Some(&[Some(inner(Some(7)))])),
+ unsafe_row_with_struct(Some(&[None])),
+ unsafe_row_with_struct(None),
+ unsafe_row_with_struct(Some(&[Some(inner(None))])),
+ ];
+
+ let array = struct_column(&data_type, &rows);
+
+ assert_eq!(array.data_type(), &data_type);
+ let st = array.as_struct();
+ assert_eq!(st.len(), 4);
+ assert!(st.is_null(2));
+ let s = st.column(0).as_struct();
+ assert_eq!(s.null_count(), 2);
+ assert!(s.is_null(1) && s.is_null(2));
+ let x: Vec<Option<i32>> = s.column(0).as_primitive::<Int32Type>().iter().collect();
+ assert_eq!(x, vec![Some(7), None, None, None]);
+ let n = s.column(1);
+ assert_eq!(n.data_type(), &DataType::Null);
+ assert_eq!(n.len(), 4);
+ assert_eq!(n.logical_null_count(), 4);
+ }
}
diff --git a/spark/src/test/scala/org/apache/comet/exec/CometColumnarShuffleSuite.scala b/spark/src/test/scala/org/apache/comet/exec/CometColumnarShuffleSuite.scala
index 4e776ea206..7f97054f1d 100644
--- a/spark/src/test/scala/org/apache/comet/exec/CometColumnarShuffleSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/exec/CometColumnarShuffleSuite.scala
@@ -142,6 +142,23 @@ abstract class CometColumnarShuffleSuite extends CometTestBase with AdaptiveSpar
checkShuffleAnswer(shuffled, 1)
}
+ // Regression test for https://github.com/apache/datafusion-comet/issues/7062: the native
+ // row-to-Arrow conversion panicked on a struct with a `NullType` field, at the top level and
+ // inside another struct.
+ test("columnar shuffle on struct with NullType field") {
+ withParquetTable((0 until 50).map(i => Tuple1(i)), "tbl") {
+ Seq(
+ "SELECT _1, named_struct('x', _1, 'n', null) AS st FROM tbl",
+ "SELECT _1, named_struct('s', named_struct('x', _1, 'n', null)) AS st FROM tbl").foreach {
+ query =>
+ val roundRobin = sql(query).repartition(4)
+ checkShuffleAnswer(roundRobin, 1)
+ val hash = sql(query).repartition(4, $"_1")
+ checkShuffleAnswer(hash, 1)
+ }
+ }
+ }
+
test("columnar shuffle on nested struct including nulls") {
Seq(10, 201).foreach { numPartitions =>
Seq("1.0", "10.0").foreach { ratio =>
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org