Add proptest fuzzing to parquet-variant and implement fixes for findings

parquet-variant/Cargo.toml

@@ -47,6 +47,7 @@ name = "parquet_variant"
 bench = false
 
 [dev-dependencies]
+proptest = { version = "1.0", default-features = false, features = ["std"] }
 criterion = { workspace = true, default-features = false }
 rand = { version = "0.9", default-features = false, features = [
     "std",

parquet-variant/src/decoder.rs

@@ -278,8 +278,14 @@ pub(crate) fn decode_double(data: &[u8]) -> Result<f64, ArrowError> {
 /// Decodes a Date from the value section of a variant.
 pub(crate) fn decode_date(data: &[u8]) -> Result<NaiveDate, ArrowError> {
     let days_since_epoch = i32::from_le_bytes(array_from_slice(data, 0)?);
-    let value = DateTime::UNIX_EPOCH + Duration::days(i64::from(days_since_epoch));
-    Ok(value.date_naive())
+    DateTime::UNIX_EPOCH
+        .checked_add_signed(Duration::days(i64::from(days_since_epoch)))
+        .map(|value| value.date_naive())
+        .ok_or_else(|| {
+            ArrowError::CastError(format!(
+                "Could not cast `{days_since_epoch}` days into a NaiveDate"
+            ))
+        })
 }
 
 /// Decodes a TimestampMicros from the value section of a variant.
@@ -338,8 +344,8 @@ pub(crate) fn decode_timestampntz_nanos(data: &[u8]) -> Result<NaiveDateTime, Ar
 
 /// Decodes a UUID from the value section of a variant.
 pub(crate) fn decode_uuid(data: &[u8]) -> Result<Uuid, ArrowError> {
-    Uuid::from_slice(&data[0..16])
-        .map_err(|_| ArrowError::CastError(format!("Cant decode uuid from {:?}", &data[0..16])))
+    let bytes: [u8; 16] = array_from_slice(data, 0)?;
+    Ok(Uuid::from_bytes(bytes))
 }
 
 /// Decodes a Binary from the value section of a variant.
@@ -467,6 +473,15 @@ mod tests {
             NaiveDate::from_ymd_opt(2025, 4, 16).unwrap()
         );
 
+        #[test]
+        fn test_date_out_of_range() {
+            // A day count that overflows chrono's supported date range must error, not panic
+            for days in [i32::MAX, i32::MIN] {
+                let result = decode_date(&days.to_le_bytes());
+                assert!(matches!(result, Err(ArrowError::CastError(_))));
+            }
+        }
+
         test_decoder_bounds!(
             test_timestamp_micros,
             [0xe0, 0x52, 0x97, 0xdd, 0xe7, 0x32, 0x06, 0x00],
@@ -531,18 +546,15 @@ mod tests {
         );
     }
 
-    #[test]
-    fn test_uuid() {
-        let data = [
+    test_decoder_bounds!(
+        test_uuid,
+        [
             0xf2, 0x4f, 0x9b, 0x64, 0x81, 0xfa, 0x49, 0xd1, 0xb7, 0x4e, 0x8c, 0x09, 0xa6, 0xe3,
             0x1c, 0x56,
-        ];
-        let result = decode_uuid(&data).unwrap();
-        assert_eq!(
-            Uuid::parse_str("f24f9b64-81fa-49d1-b74e-8c09a6e31c56").unwrap(),
-            result
-        );
-    }
+        ],
+        decode_uuid,
+        Uuid::parse_str("f24f9b64-81fa-49d1-b74e-8c09a6e31c56").unwrap()
+    );
 
     mod time {
         use super::*;

parquet-variant/src/variant.rs

@@ -49,6 +49,19 @@ mod object;
 
 const MAX_SHORT_STRING_BYTES: usize = 0x3F;
 
+/// The maximum number of nested objects and arrays a [`Variant`] may contain.
+///
+/// Full [validation] recurses into nested values, so unbounded nesting would overflow the stack --
+/// an abort, not a catchable panic. Validation therefore rejects more deeply nested values, which
+/// also bounds the recursion of infallible accesses such as [`Debug`] and [`PartialEq`].
+///
+/// The variant [spec] does not specify a limit. This value matches the default `serde_json`
+/// recursion limit, so any variant parsed from JSON already satisfies it.
+///
+/// [validation]: Variant#Validation
+/// [spec]: https://github.com/apache/parquet-format/blob/master/VariantEncoding.md
+pub const MAX_NESTING_DEPTH: usize = 128;
+
 /// A Variant [`ShortString`]
 ///
 /// This implementation is a zero cost wrapper over `&str` that ensures
@@ -403,7 +416,17 @@ impl<'m, 'v> Variant<'m, 'v> {
         metadata: VariantMetadata<'m>,
         value: &'v [u8],
     ) -> Result<Self, ArrowError> {
-        Self::try_new_with_metadata_and_shallow_validation(metadata, value)?.with_full_validation()
+        Self::try_new_with_metadata_at_depth(metadata, value, 0)
+    }
+
+    // Same as [`Self::try_new_with_metadata`], tracking how deeply validation has recursed.
+    pub(crate) fn try_new_with_metadata_at_depth(
+        metadata: VariantMetadata<'m>,
+        value: &'v [u8],
+        depth: usize,
+    ) -> Result<Self, ArrowError> {
+        Self::try_new_with_metadata_and_shallow_validation(metadata, value)?
+            .with_full_validation_at_depth(depth)
     }
 
     /// Similar to [`Self::try_new_with_metadata`], but [unvalidated].
@@ -501,13 +524,20 @@ impl<'m, 'v> Variant<'m, 'v> {
     /// If [`Self::is_fully_validated`] is true, validation is a no-op. Otherwise, the cost is `O(m + v)`
     /// where `m` and `v` are the sizes of metadata and value buffers, respectively.
     ///
+    /// Values nested more than [`MAX_NESTING_DEPTH`] deep are rejected.
+    ///
     /// [objects]: VariantObject#Validation
     /// [arrays]: VariantList#Validation
     pub fn with_full_validation(self) -> Result<Self, ArrowError> {
+        self.with_full_validation_at_depth(0)
+    }
+
+    // Same as [`Self::with_full_validation`], tracking how deeply validation has recursed.
+    pub(crate) fn with_full_validation_at_depth(self, depth: usize) -> Result<Self, ArrowError> {
         use Variant::*;
         match self {
-            List(list) => list.with_full_validation().map(List),
-            Object(obj) => obj.with_full_validation().map(Object),
+            List(list) => list.with_full_validation_at_depth(depth).map(List),
+            Object(obj) => obj.with_full_validation_at_depth(depth).map(Object),
             _ => Ok(self),
         }
     }

parquet-variant/src/variant/list.rs

@@ -18,7 +18,7 @@ use crate::decoder::{OffsetSizeBytes, map_bytes_to_offsets};
 use crate::utils::{
     first_byte_from_slice, overflow_error, slice_from_slice, slice_from_slice_at_offset,
 };
-use crate::variant::{Variant, VariantMetadata};
+use crate::variant::{MAX_NESTING_DEPTH, Variant, VariantMetadata};
 
 use arrow_schema::ArrowError;
 
@@ -208,9 +208,25 @@ impl<'m, 'v> VariantList<'m, 'v> {
 
     /// Performs a full [validation] of this variant array and returns the result.
     ///
+    /// Values nested more than [`MAX_NESTING_DEPTH`] deep are rejected.
+    ///
     /// [validation]: Self#Validation
-    pub fn with_full_validation(mut self) -> Result<Self, ArrowError> {
+    pub fn with_full_validation(self) -> Result<Self, ArrowError> {
+        self.with_full_validation_at_depth(0)
+    }
+
+    // Same as [`Self::with_full_validation`], tracking how deeply validation has recursed.
+    pub(crate) fn with_full_validation_at_depth(
+        mut self,
+        depth: usize,
+    ) -> Result<Self, ArrowError> {
         if !self.validated {
+            if depth >= MAX_NESTING_DEPTH {
+                return Err(ArrowError::InvalidArgumentError(format!(
+                    "Variant nesting depth exceeds the maximum of {MAX_NESTING_DEPTH}"
+                )));
+            }
+
             // Validate the metadata dictionary first, if not already validated, because we pass it
             // by value to all the children (who would otherwise re-validate it repeatedly).
             self.metadata = self.metadata.with_full_validation()?;
@@ -231,7 +247,11 @@ impl<'m, 'v> VariantList<'m, 'v> {
 
             for next_offset in offset_iter {
                 let value_bytes = slice_from_slice(value_buffer, current_offset..next_offset)?;
-                Variant::try_new_with_metadata(self.metadata.clone(), value_bytes)?;
+                Variant::try_new_with_metadata_at_depth(
+                    self.metadata.clone(),
+                    value_bytes,
+                    depth + 1,
+                )?;
                 current_offset = next_offset;
             }
 
@@ -727,6 +747,36 @@ mod tests {
         assert_ne!(object.get("list1").unwrap(), object.get("list3").unwrap());
     }
 
+    /// return the value bytes for `depth` single-element lists nested around a null
+    fn make_nested_lists(depth: usize) -> Vec<u8> {
+        let mut value = vec![0u8]; // null primitive
+        for _ in 0..depth {
+            // header: array basic type, 4-byte offsets, then num_elements=1 and two offsets
+            let mut outer = vec![0x0F, 1];
+            outer.extend_from_slice(&0u32.to_le_bytes());
+            outer.extend_from_slice(&(value.len() as u32).to_le_bytes());
+            outer.append(&mut value);
+            value = outer;
+        }
+        value
+    }
+
+    #[test]
+    fn test_variant_list_nesting_depth_limit() {
+        let metadata = [0x01, 0, 0]; // empty dictionary
+
+        let value = make_nested_lists(MAX_NESTING_DEPTH);
+        Variant::try_new(&metadata, &value).unwrap();
+
+        // One level deeper would recurse past the limit -- previously a stack overflow
+        let value = make_nested_lists(MAX_NESTING_DEPTH + 1);
+        let err = Variant::try_new(&metadata, &value).unwrap_err();
+        assert!(
+            err.to_string().contains("nesting depth exceeds"),
+            "unexpected error: {err}"
+        );
+    }
+
     /// return metadata/value for a simple variant list with values in a range
     fn make_listi32(range: Range<i32>) -> (Vec<u8>, Vec<u8>) {
         let mut variant_builder = VariantBuilder::new();

parquet-variant/src/variant/metadata.rs

@@ -115,7 +115,7 @@ impl VariantMetadataHeader {
 /// - first offset is zero
 /// - last offset is in-bounds
 /// - all other offsets are in-bounds (*)
-/// - all offsets are monotonically increasing (*)
+/// - all offsets are non-decreasing (*)
 /// - all values are valid utf-8 (*)
 ///
 /// NOTE: [`Self::new`] only skips expensive (non-constant cost) validation checks (marked by `(*)`
@@ -309,15 +309,19 @@ impl<'m> VariantMetadata<'m> {
                     current_offset = next_offset;
                 }
             } else {
-                // Validate offsets are in-bounds and monotonically increasing
-                //
-                // Since shallow validation ensures the first and last offsets are in bounds,
-                // we can also verify all offsets are in-bounds by checking if
-                // offsets are monotonically increasing
-                if !offsets.is_sorted_by(|a, b| a < b) {
-                    return Err(ArrowError::InvalidArgumentError(
-                        "offsets not monotonically increasing".to_string(),
-                    ));
+                // Slicing each dictionary value validates that offsets are in-bounds, non-decreasing,
+                // and land on UTF-8 character boundaries. Equal offsets are legal: they encode an
+                // empty dictionary entry.
+                let mut current_offset = offsets.next().unwrap_or(0);
+                for next_offset in offsets {
+                    value_buffer
+                        .get(current_offset..next_offset)
+                        .ok_or_else(|| {
+                            ArrowError::InvalidArgumentError(format!(
+                                "range {current_offset}..{next_offset} is invalid or out of bounds"
+                            ))
+                        })?;
+                    current_offset = next_offset;
                 }
             }
 
@@ -608,6 +612,19 @@ mod tests {
             matches!(err, ArrowError::InvalidArgumentError(_)),
             "unexpected error: {err:?}"
         );
+
+        let bytes = &[
+            0b0000_0001, // header: offset_size_minus_one=0, ordered=0, version=1
+            2,
+            0x00,
+            0x02,
+            0x02, // an unsorted dict may hold an empty string anywhere
+            b'h',
+            b'i',
+        ];
+        let metadata = VariantMetadata::try_new(bytes).unwrap();
+        assert_eq!(&metadata[0], "hi");
+        assert_eq!(&metadata[1], "");
     }
 
     #[test]
@@ -663,4 +680,24 @@ mod tests {
 
         assert_eq!(m1, m2);
     }
+
+    #[test]
+    fn test_empty_string_field_names() {
+        // Field names are added to the dictionary in insertion order, so this dictionary is
+        // unsorted, and the empty field name makes its last two offsets equal.
+        let mut b = VariantBuilder::new().with_field_names(["b", "a", ""]);
+        let mut o = b.new_object();
+
+        o.insert("b", false);
+        o.insert("a", false);
+        o.insert("", false);
+
+        o.finish();
+
+        let (m, _) = b.finish();
+
+        let metadata = VariantMetadata::try_new(&m).unwrap();
+        assert!(!metadata.is_sorted());
+        assert_eq!(metadata.iter().collect::<Vec<_>>(), vec!["b", "a", ""]);
+    }
 }