Panic in Comet shuffle when building structs

[andygrove]

Describe the bug

In PR #1736 there is a panic during Comet shuffle when building structs.

2025-05-30T21:12:35.6022140Z [info] - merge with updates to nested struct fields in MATCHED clauses *** FAILED *** (108 milliseconds)
2025-05-30T21:12:35.6054223Z [info]   org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 338.0 failed 1 times, most recent failure: Lost task 0.0 in stage 338.0 (TID 428) (84023761926b executor driver): org.apache.comet.CometNativeException: StructBuilder (Schema { fields: [Field { name: "a", data_type: List(Field { name: "item", data_type: Int32, nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }), nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }, Field { name: "m", data_type: Map(Field { name: "entries", data_type: Struct([Field { name: "key", data_type: Utf8, nullable: false, dict_id: 0, dict_is_ordered: false, metadata: {} }, Field { name: "value", data_type: Utf8, nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }]), nullable: false, dict_id: 0, dict_is_ordered: false, metadata: {} }, false), nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }], metadata: {} }) and field_builder with index 0 (List(Field { name: "item", data_type: Int32, nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} })) are of unequal lengths: (1 != 0).
2025-05-30T21:12:35.6067600Z [info]         at comet::errors::init::{{closure}}(__internal__:0)
2025-05-30T21:12:35.6068602Z [info]         at std::panicking::rust_panic_with_hook(__internal__:0)
2025-05-30T21:12:35.6069248Z [info]         at std::panicking::begin_panic_handler::{{closure}}(__internal__:0)
2025-05-30T21:12:35.6069913Z [info]         at std::sys::backtrace::__rust_end_short_backtrace(__internal__:0)
2025-05-30T21:12:35.6070581Z [info]         at __rustc::rust_begin_unwind(__internal__:0)
2025-05-30T21:12:35.6071098Z [info]         at core::panicking::panic_fmt(__internal__:0)
2025-05-30T21:12:35.6071664Z [info]         at arrow_array::builder::struct_builder::StructBuilder::validate_content::{{closure}}::panic_cold_display(__internal__:0)
2025-05-30T21:12:35.6072355Z [info]         at arrow_array::builder::struct_builder::StructBuilder::validate_content(__internal__:0)
2025-05-30T21:12:35.6072899Z [info]         at arrow_array::builder::struct_builder::StructBuilder::finish(__internal__:0)
2025-05-30T21:12:35.6073514Z [info]         at <arrow_array::builder::struct_builder::StructBuilder as arrow_array::builder::ArrayBuilder>::finish(__internal__:0)
2025-05-30T21:12:35.6074135Z [info]         at arrow_array::builder::struct_builder::StructBuilder::finish(__internal__:0)
2025-05-30T21:12:35.6074751Z [info]         at <arrow_array::builder::struct_builder::StructBuilder as arrow_array::builder::ArrayBuilder>::finish(__internal__:0)
2025-05-30T21:12:35.6075546Z [info]         at <core::iter::adapters::GenericShunt<I,R> as core::iter::traits::iterator::Iterator>::next(__internal__:0)
2025-05-30T21:12:35.6076196Z [info]         at comet::execution::shuffle::row::process_sorted_row_partition(__internal__:0)
2025-05-30T21:12:35.6076943Z [info]         at comet::execution::jni_api::Java_org_apache_comet_Native_writeSortedFileNative::{{closure}}::{{closure}}(__internal__:0)
2025-05-30T21:12:35.6077543Z [info]         at Java_org_apache_comet_Native_writeSortedFileNative(__internal__:0)
2025-05-30T21:12:35.6077914Z [info]         at <unknown>(__internal__:0)
2025-05-30T21:12:35.6078818Z [info]     at org.apache.comet.Native.writeSortedFileNative(Native Method)
2025-05-30T21:12:35.6079388Z [info]     at org.apache.spark.sql.comet.execution.shuffle.SpillWriter.doSpilling(SpillWriter.java:187)
2025-05-30T21:12:35.6080158Z [info]     at org.apache.spark.sql.comet.execution.shuffle.CometDiskBlockWriter$ArrowIPCWriter.doSpilling(CometDiskBlockWriter.java:401)
2025-05-30T21:12:35.6080975Z [info]     at org.apache.spark.sql.comet.execution.shuffle.CometDiskBlockWriter.close(CometDiskBlockWriter.java:304)
2025-05-30T21:12:35.6081846Z [info]     at org.apache.spark.sql.comet.execution.shuffle.CometBypassMergeSortShuffleWriter.write(CometBypassMergeSortShuffleWriter.java:244)

Formatted version of error:

StructBuilder (Schema { fields: [
  Field { name: "a", data_type: List(Field { name: "item", data_type: Int32, ... }), ... }, 
  Field { name: "m", data_type: Map(Field { name: "entries", data_type: Struct([
    Field { name: "key", data_type: Utf8, ... }, 
    Field { name: "value", data_type: Utf8, ... }]), ... }, false), ... 
  }
], ... }) 

... and field_builder with index 0 ...

 (List(Field { name: "item", data_type: Int32, ... })) are of unequal lengths: (1 != 0).

Steps to reproduce

No response

Expected behavior

No response

Additional context

No response