perf: extend field-major processing to nested struct fields

This extends the field-major optimization (from commit 471fb2a) to
recursively handle nested Struct fields. Previously, nested structs fell
back to row-major processing via `append_field`, losing the benefit of
field-major processing at each nesting level.

Changes:
- Add `append_nested_struct_fields_field_major` helper function that
  recursively processes nested struct fields using field-major order
- Update `append_struct_fields_field_major` to use field-major processing
  for nested Struct fields instead of falling back to `append_field`
- Add benchmarks for 2-level and 3-level nested structs

The optimization:
1. Gets the nested StructBuilder once per field
2. Builds nested struct validity in one pass
3. Recursively applies field-major processing to nested struct fields

List and Map fields continue to fall back to `append_field` since they
have variable-length elements that are harder to optimize.

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

Author: andygrove

native/core/benches/struct_conversion.rs

@@ -156,13 +156,316 @@ fn benchmark_struct_conversion(c: &mut Criterion) {
     group.finish();
 }
 
+/// Create a schema with nested structs: Struct<Struct<int64 fields>>
+fn make_nested_struct_schema(num_fields: usize) -> DataType {
+    let inner_fields: Vec<Field> = (0..num_fields)
+        .map(|i| Field::new(format!("inner_f{}", i), DataType::Int64, true))
+        .collect();
+    let inner_struct = DataType::Struct(Fields::from(inner_fields));
+    let outer_fields = vec![Field::new("nested", inner_struct, true)];
+    DataType::Struct(Fields::from(outer_fields))
+}
+
+/// Create a schema with deeply nested structs (3 levels): Struct<Struct<Struct<int64 fields>>>
+fn make_deeply_nested_struct_schema(num_fields: usize) -> DataType {
+    let inner_fields: Vec<Field> = (0..num_fields)
+        .map(|i| Field::new(format!("inner_f{}", i), DataType::Int64, true))
+        .collect();
+    let inner_struct = DataType::Struct(Fields::from(inner_fields));
+    let middle_fields = vec![Field::new("level2", inner_struct, true)];
+    let middle_struct = DataType::Struct(Fields::from(middle_fields));
+    let outer_fields = vec![Field::new("level1", middle_struct, true)];
+    DataType::Struct(Fields::from(outer_fields))
+}
+
+/// Calculate row size for nested struct: Struct<Struct<int64 fields>>
+fn get_nested_row_size(num_inner_fields: usize) -> usize {
+    // Top-level row has 1 column (the outer struct)
+    let top_level_bitset_width = SparkUnsafeRow::get_row_bitset_width(1);
+    let struct_pointer_size = 8;
+
+    // Outer struct has 1 field (the inner struct)
+    let outer_bitset_width = SparkUnsafeRow::get_row_bitset_width(1);
+    let outer_struct_size = outer_bitset_width + 8; // pointer to inner struct
+
+    // Inner struct has num_inner_fields int64 fields
+    let inner_bitset_width = SparkUnsafeRow::get_row_bitset_width(num_inner_fields);
+    let inner_data_size = num_inner_fields * 8;
+    let inner_struct_size = inner_bitset_width + inner_data_size;
+
+    top_level_bitset_width + struct_pointer_size + outer_struct_size + inner_struct_size
+}
+
+/// Calculate row size for deeply nested struct: Struct<Struct<Struct<int64 fields>>>
+fn get_deeply_nested_row_size(num_inner_fields: usize) -> usize {
+    // Top-level row has 1 column (the level1 struct)
+    let top_level_bitset_width = SparkUnsafeRow::get_row_bitset_width(1);
+    let struct_pointer_size = 8;
+
+    // Level 1 struct has 1 field (the level2 struct)
+    let level1_bitset_width = SparkUnsafeRow::get_row_bitset_width(1);
+    let level1_struct_size = level1_bitset_width + 8;
+
+    // Level 2 struct has 1 field (the inner struct)
+    let level2_bitset_width = SparkUnsafeRow::get_row_bitset_width(1);
+    let level2_struct_size = level2_bitset_width + 8;
+
+    // Inner struct has num_inner_fields int64 fields
+    let inner_bitset_width = SparkUnsafeRow::get_row_bitset_width(num_inner_fields);
+    let inner_data_size = num_inner_fields * 8;
+    let inner_struct_size = inner_bitset_width + inner_data_size;
+
+    top_level_bitset_width
+        + struct_pointer_size
+        + level1_struct_size
+        + level2_struct_size
+        + inner_struct_size
+}
+
+struct NestedRowData {
+    data: Vec<u8>,
+}
+
+impl NestedRowData {
+    fn new(num_inner_fields: usize) -> Self {
+        let row_size = get_nested_row_size(num_inner_fields);
+        let mut data = vec![0u8; row_size];
+
+        let top_level_bitset_width = SparkUnsafeRow::get_row_bitset_width(1);
+        let outer_bitset_width = SparkUnsafeRow::get_row_bitset_width(1);
+        let inner_bitset_width = SparkUnsafeRow::get_row_bitset_width(num_inner_fields);
+
+        // Calculate offsets
+        let outer_struct_start = top_level_bitset_width + 8;
+        let outer_struct_size = outer_bitset_width + 8;
+        let inner_struct_start = outer_struct_start + outer_struct_size;
+        let inner_struct_size = inner_bitset_width + num_inner_fields * 8;
+
+        // Write top-level struct pointer (points to outer struct)
+        let outer_offset_and_size =
+            ((outer_struct_start as i64) << 32) | (outer_struct_size as i64);
+        data[top_level_bitset_width..top_level_bitset_width + 8]
+            .copy_from_slice(&outer_offset_and_size.to_le_bytes());
+
+        // Write outer struct pointer (points to inner struct)
+        // Offset is relative to outer struct start
+        let inner_relative_offset = inner_struct_start - outer_struct_start;
+        let inner_offset_and_size =
+            ((inner_relative_offset as i64) << 32) | (inner_struct_size as i64);
+        data[outer_struct_start + outer_bitset_width..outer_struct_start + outer_bitset_width + 8]
+            .copy_from_slice(&inner_offset_and_size.to_le_bytes());
+
+        // Fill inner struct with some data
+        for i in 0..num_inner_fields {
+            let value_offset = inner_struct_start + inner_bitset_width + i * 8;
+            let value = (i as i64) * 100;
+            data[value_offset..value_offset + 8].copy_from_slice(&value.to_le_bytes());
+        }
+
+        NestedRowData { data }
+    }
+
+    fn to_spark_row(&self, spark_row: &mut SparkUnsafeRow) {
+        spark_row.point_to_slice(&self.data);
+    }
+}
+
+struct DeeplyNestedRowData {
+    data: Vec<u8>,
+}
+
+impl DeeplyNestedRowData {
+    fn new(num_inner_fields: usize) -> Self {
+        let row_size = get_deeply_nested_row_size(num_inner_fields);
+        let mut data = vec![0u8; row_size];
+
+        let top_level_bitset_width = SparkUnsafeRow::get_row_bitset_width(1);
+        let level1_bitset_width = SparkUnsafeRow::get_row_bitset_width(1);
+        let level2_bitset_width = SparkUnsafeRow::get_row_bitset_width(1);
+        let inner_bitset_width = SparkUnsafeRow::get_row_bitset_width(num_inner_fields);
+
+        // Calculate offsets
+        let level1_struct_start = top_level_bitset_width + 8;
+        let level1_struct_size = level1_bitset_width + 8;
+        let level2_struct_start = level1_struct_start + level1_struct_size;
+        let level2_struct_size = level2_bitset_width + 8;
+        let inner_struct_start = level2_struct_start + level2_struct_size;
+        let inner_struct_size = inner_bitset_width + num_inner_fields * 8;
+
+        // Write top-level struct pointer (points to level1 struct)
+        let level1_offset_and_size =
+            ((level1_struct_start as i64) << 32) | (level1_struct_size as i64);
+        data[top_level_bitset_width..top_level_bitset_width + 8]
+            .copy_from_slice(&level1_offset_and_size.to_le_bytes());
+
+        // Write level1 struct pointer (points to level2 struct)
+        let level2_relative_offset = level2_struct_start - level1_struct_start;
+        let level2_offset_and_size =
+            ((level2_relative_offset as i64) << 32) | (level2_struct_size as i64);
+        data[level1_struct_start + level1_bitset_width
+            ..level1_struct_start + level1_bitset_width + 8]
+            .copy_from_slice(&level2_offset_and_size.to_le_bytes());
+
+        // Write level2 struct pointer (points to inner struct)
+        let inner_relative_offset = inner_struct_start - level2_struct_start;
+        let inner_offset_and_size =
+            ((inner_relative_offset as i64) << 32) | (inner_struct_size as i64);
+        data[level2_struct_start + level2_bitset_width
+            ..level2_struct_start + level2_bitset_width + 8]
+            .copy_from_slice(&inner_offset_and_size.to_le_bytes());
+
+        // Fill inner struct with some data
+        for i in 0..num_inner_fields {
+            let value_offset = inner_struct_start + inner_bitset_width + i * 8;
+            let value = (i as i64) * 100;
+            data[value_offset..value_offset + 8].copy_from_slice(&value.to_le_bytes());
+        }
+
+        DeeplyNestedRowData { data }
+    }
+
+    fn to_spark_row(&self, spark_row: &mut SparkUnsafeRow) {
+        spark_row.point_to_slice(&self.data);
+    }
+}
+
+fn benchmark_nested_struct_conversion(c: &mut Criterion) {
+    let mut group = c.benchmark_group("nested_struct_conversion");
+
+    // Test nested structs with different inner field counts
+    for num_fields in [5, 10, 20] {
+        for num_rows in [1000, 10000] {
+            let schema = vec![make_nested_struct_schema(num_fields)];
+
+            // Create row data
+            let rows: Vec<NestedRowData> = (0..num_rows)
+                .map(|_| NestedRowData::new(num_fields))
+                .collect();
+
+            let spark_rows: Vec<SparkUnsafeRow> = rows
+                .iter()
+                .map(|row_data| {
+                    let mut spark_row = SparkUnsafeRow::new_with_num_fields(1);
+                    row_data.to_spark_row(&mut spark_row);
+                    spark_row.set_not_null_at(0);
+                    spark_row
+                })
+                .collect();
+
+            let mut row_addresses: Vec<i64> =
+                spark_rows.iter().map(|row| row.get_row_addr()).collect();
+            let mut row_sizes: Vec<i32> = spark_rows.iter().map(|row| row.get_row_size()).collect();
+
+            let row_address_ptr = row_addresses.as_mut_ptr();
+            let row_size_ptr = row_sizes.as_mut_ptr();
+
+            group.bench_with_input(
+                BenchmarkId::new(
+                    format!("inner_fields_{}", num_fields),
+                    format!("rows_{}", num_rows),
+                ),
+                &(num_rows, &schema),
+                |b, (num_rows, schema)| {
+                    b.iter(|| {
+                        let tempfile = Builder::new().tempfile().unwrap();
+
+                        process_sorted_row_partition(
+                            *num_rows,
+                            BATCH_SIZE,
+                            row_address_ptr,
+                            row_size_ptr,
+                            schema,
+                            tempfile.path().to_str().unwrap().to_string(),
+                            1.0,
+                            false,
+                            0,
+                            None,
+                            &CompressionCodec::Zstd(1),
+                        )
+                        .unwrap();
+                    });
+                },
+            );
+
+            std::mem::drop(spark_rows);
+        }
+    }
+
+    group.finish();
+}
+
+fn benchmark_deeply_nested_struct_conversion(c: &mut Criterion) {
+    let mut group = c.benchmark_group("deeply_nested_struct_conversion");
+
+    // Test deeply nested structs (3 levels) with different inner field counts
+    for num_fields in [5, 10, 20] {
+        for num_rows in [1000, 10000] {
+            let schema = vec![make_deeply_nested_struct_schema(num_fields)];
+
+            // Create row data
+            let rows: Vec<DeeplyNestedRowData> = (0..num_rows)
+                .map(|_| DeeplyNestedRowData::new(num_fields))
+                .collect();
+
+            let spark_rows: Vec<SparkUnsafeRow> = rows
+                .iter()
+                .map(|row_data| {
+                    let mut spark_row = SparkUnsafeRow::new_with_num_fields(1);
+                    row_data.to_spark_row(&mut spark_row);
+                    spark_row.set_not_null_at(0);
+                    spark_row
+                })
+                .collect();
+
+            let mut row_addresses: Vec<i64> =
+                spark_rows.iter().map(|row| row.get_row_addr()).collect();
+            let mut row_sizes: Vec<i32> = spark_rows.iter().map(|row| row.get_row_size()).collect();
+
+            let row_address_ptr = row_addresses.as_mut_ptr();
+            let row_size_ptr = row_sizes.as_mut_ptr();
+
+            group.bench_with_input(
+                BenchmarkId::new(
+                    format!("inner_fields_{}", num_fields),
+                    format!("rows_{}", num_rows),
+                ),
+                &(num_rows, &schema),
+                |b, (num_rows, schema)| {
+                    b.iter(|| {
+                        let tempfile = Builder::new().tempfile().unwrap();
+
+                        process_sorted_row_partition(
+                            *num_rows,
+                            BATCH_SIZE,
+                            row_address_ptr,
+                            row_size_ptr,
+                            schema,
+                            tempfile.path().to_str().unwrap().to_string(),
+                            1.0,
+                            false,
+                            0,
+                            None,
+                            &CompressionCodec::Zstd(1),
+                        )
+                        .unwrap();
+                    });
+                },
+            );
+
+            std::mem::drop(spark_rows);
+        }
+    }
+
+    group.finish();
+}
+
 fn config() -> Criterion {
     Criterion::default()
 }
 
 criterion_group! {
     name = benches;
     config = config();
-    targets = benchmark_struct_conversion
+    targets = benchmark_struct_conversion, benchmark_nested_struct_conversion, benchmark_deeply_nested_struct_conversion
 }
 criterion_main!(benches);