[Variant] VariantArrayBuilder tracks only offsets (apache#8193)

# Which issue does this PR close?

- Closes apache#8192

# Rationale for this change

Tracking lengths is redundant because that's just the difference between
adjacent offsets. It becomes even easier if we eventually start adding
the views eagerly instead of storing up offsets, because we only need to
remember the difference between starting and final offset.

# What changes are included in this PR?

Update the algorithm that creates the binary view to compute the
offsets.

# Are these changes tested?

Existing tests cover this change.

# Are there any user-facing changes?

No.

Author: scovich

parquet-variant-compute/src/variant_array_builder.rs

@@ -74,12 +74,12 @@ pub struct VariantArrayBuilder {
     nulls: NullBufferBuilder,
     /// buffer for all the metadata
     metadata_buffer: Vec<u8>,
-    /// (offset, len) pairs for locations of metadata in the buffer
-    metadata_locations: Vec<(usize, usize)>,
+    /// ending offset for each serialized metadata dictionary in the buffer
+    metadata_offsets: Vec<usize>,
     /// buffer for values
     value_buffer: Vec<u8>,
-    /// (offset, len) pairs for locations of values in the buffer
-    value_locations: Vec<(usize, usize)>,
+    /// ending offset for each serialized variant value in the buffer
+    value_offsets: Vec<usize>,
     /// The fields of the final `StructArray`
     ///
     /// TODO: 1) Add extension type metadata
@@ -96,9 +96,9 @@ impl VariantArrayBuilder {
         Self {
             nulls: NullBufferBuilder::new(row_capacity),
             metadata_buffer: Vec::new(), // todo allocation capacity
-            metadata_locations: Vec::with_capacity(row_capacity),
+            metadata_offsets: Vec::with_capacity(row_capacity),
             value_buffer: Vec::new(),
-            value_locations: Vec::with_capacity(row_capacity),
+            value_offsets: Vec::with_capacity(row_capacity),
             fields: Fields::from(vec![metadata_field, value_field]),
         }
     }
@@ -108,15 +108,15 @@ impl VariantArrayBuilder {
         let Self {
             mut nulls,
             metadata_buffer,
-            metadata_locations,
+            metadata_offsets,
             value_buffer,
-            value_locations,
+            value_offsets,
             fields,
         } = self;
 
-        let metadata_array = binary_view_array_from_buffers(metadata_buffer, metadata_locations);
+        let metadata_array = binary_view_array_from_buffers(metadata_buffer, metadata_offsets);
 
-        let value_array = binary_view_array_from_buffers(value_buffer, value_locations);
+        let value_array = binary_view_array_from_buffers(value_buffer, value_offsets);
 
         // The build the final struct array
         let inner = StructArray::new(
@@ -136,13 +136,8 @@ impl VariantArrayBuilder {
     pub fn append_null(&mut self) {
         self.nulls.append_null();
         // The subfields are expected to be non-nullable according to the parquet variant spec.
-        let metadata_offset = self.metadata_buffer.len();
-        let metadata_length = 0;
-        self.metadata_locations
-            .push((metadata_offset, metadata_length));
-        let value_offset = self.value_buffer.len();
-        let value_length = 0;
-        self.value_locations.push((value_offset, value_length));
+        self.metadata_offsets.push(self.metadata_buffer.len());
+        self.value_offsets.push(self.value_buffer.len());
     }
 
     /// Append the [`Variant`] to the builder as the next row
@@ -186,10 +181,7 @@ impl VariantArrayBuilder {
     /// assert!(variant_array.value(1).as_object().is_some());
     ///  ```
     pub fn variant_builder(&mut self) -> VariantArrayVariantBuilder<'_> {
-        // append directly into the metadata and value buffers
-        let metadata_buffer = std::mem::take(&mut self.metadata_buffer);
-        let value_buffer = std::mem::take(&mut self.value_buffer);
-        VariantArrayVariantBuilder::new(self, metadata_buffer, value_buffer)
+        VariantArrayVariantBuilder::new(self)
     }
 }
 
@@ -236,11 +228,10 @@ impl<'a> VariantArrayVariantBuilder<'a> {
     ///
     /// Note this is not public as this is a structure that is logically
     /// part of the [`VariantArrayBuilder`] and relies on its internal structure
-    fn new(
-        array_builder: &'a mut VariantArrayBuilder,
-        metadata_buffer: Vec<u8>,
-        value_buffer: Vec<u8>,
-    ) -> Self {
+    fn new(array_builder: &'a mut VariantArrayBuilder) -> Self {
+        // append directly into the metadata and value buffers
+        let metadata_buffer = std::mem::take(&mut array_builder.metadata_buffer);
+        let value_buffer = std::mem::take(&mut array_builder.value_buffer);
         let metadata_offset = metadata_buffer.len();
         let value_offset = value_buffer.len();
         VariantArrayVariantBuilder {
@@ -276,27 +267,25 @@ impl<'a> VariantArrayVariantBuilder<'a> {
         let (metadata_buffer, value_buffer) = std::mem::take(&mut self.variant_builder).finish();
 
         // Sanity Check: if the buffers got smaller, something went wrong (previous data was lost)
-        let metadata_len = metadata_buffer
-            .len()
-            .checked_sub(metadata_offset)
-            .expect("metadata length decreased unexpectedly");
-        let value_len = value_buffer
-            .len()
-            .checked_sub(value_offset)
-            .expect("value length decreased unexpectedly");
+        assert!(
+            metadata_offset <= metadata_buffer.len(),
+            "metadata length decreased unexpectedly"
+        );
+        assert!(
+            value_offset <= value_buffer.len(),
+            "value length decreased unexpectedly"
+        );
 
         // commit the changes by putting the
-        // offsets and lengths into the parent array builder.
-        self.array_builder
-            .metadata_locations
-            .push((metadata_offset, metadata_len));
-        self.array_builder
-            .value_locations
-            .push((value_offset, value_len));
-        self.array_builder.nulls.append_non_null();
+        // ending offsets into the parent array builder.
+        let builder = &mut self.array_builder;
+        builder.metadata_offsets.push(metadata_buffer.len());
+        builder.value_offsets.push(value_buffer.len());
+        builder.nulls.append_non_null();
+
         // put the buffers back into the array builder
-        self.array_builder.metadata_buffer = metadata_buffer;
-        self.array_builder.value_buffer = value_buffer;
+        builder.metadata_buffer = metadata_buffer;
+        builder.value_buffer = value_buffer;
     }
 }
 
@@ -338,19 +327,21 @@ impl Drop for VariantArrayVariantBuilder<'_> {
     }
 }
 
-fn binary_view_array_from_buffers(
-    buffer: Vec<u8>,
-    locations: Vec<(usize, usize)>,
-) -> BinaryViewArray {
-    let mut builder = BinaryViewBuilder::with_capacity(locations.len());
+fn binary_view_array_from_buffers(buffer: Vec<u8>, offsets: Vec<usize>) -> BinaryViewArray {
+    // All offsets are less than or equal to the buffer length, so we can safely cast all offsets
+    // inside the loop below, as long as the buffer length fits in u32.
+    u32::try_from(buffer.len()).expect("buffer length should fit in u32");
+
+    let mut builder = BinaryViewBuilder::with_capacity(offsets.len());
     let block = builder.append_block(buffer.into());
     // TODO this can be much faster if it creates the views directly during append
-    for (offset, length) in locations {
-        let offset = offset.try_into().expect("offset should fit in u32");
-        let length = length.try_into().expect("length should fit in u32");
+    let mut start = 0;
+    for end in offsets {
+        let end = end as u32; // Safe cast: validated max offset fits in u32 above
         builder
-            .try_append_view(block, offset, length)
+            .try_append_view(block, start, end - start)
             .expect("Failed to append view");
+        start = end;
     }
     builder.finish()
 }