Backport manifest-list bound width inference to DF55 fork

iceberg-rust-spec/src/spec/manifest.rs

@@ -684,6 +684,14 @@ impl AvroMap<ByteBuf> {
     }
 }
 
+/// Decodes a manifest bound from its single-value-serialized bytes against the current field type.
+///
+/// Type promotion does not rewrite existing bounds, so a bound may be encoded at a narrower
+/// write-time type than the current schema reports. Per the Iceberg spec, the write-time type is
+/// inferred from the byte length: a 4-byte bound for a current `long`/`double`/`timestamp` was
+/// written as `int`/`float`/`date` and is promoted here. All other cases (including decimals, which
+/// decode at any width from the scale in `data_type`) fall through to the plain single-value
+/// decoding.
 pub(crate) fn decode_bound(bytes: &[u8], data_type: &Type) -> Result<Value, Error> {
     match (data_type, bytes.len()) {
         (Type::Primitive(PrimitiveType::Long), 4) => Ok(Value::LongInt(i64::from(
@@ -2203,6 +2211,61 @@ mod tests {
         );
     }
 
+    const LONG: Type = Type::Primitive(PrimitiveType::Long);
+    const DOUBLE: Type = Type::Primitive(PrimitiveType::Double);
+    const TIMESTAMP: Type = Type::Primitive(PrimitiveType::Timestamp);
+
+    #[test]
+    fn decode_bound_widens_int_written_before_long_promotion() {
+        // A negative value also exercises sign extension, which a plain zero-pad would get wrong.
+        let decoded = decode_bound(&(-42_i32).to_le_bytes(), &LONG).unwrap();
+        assert_eq!(decoded, Value::LongInt(-42));
+    }
+
+    #[test]
+    fn decode_bound_reads_full_width_long() {
+        let decoded = decode_bound(&1_234_567_890_123_i64.to_le_bytes(), &LONG).unwrap();
+        assert_eq!(decoded, Value::LongInt(1_234_567_890_123));
+    }
+
+    #[test]
+    fn decode_bound_widens_float_written_before_double_promotion() {
+        let decoded = decode_bound(&2.5_f32.to_le_bytes(), &DOUBLE).unwrap();
+        assert_eq!(decoded, Value::Double(ordered_float::OrderedFloat(2.5_f64)));
+    }
+
+    #[test]
+    fn decode_bound_reads_full_width_double() {
+        let decoded = decode_bound(&2.5_f64.to_le_bytes(), &DOUBLE).unwrap();
+        assert_eq!(decoded, Value::Double(ordered_float::OrderedFloat(2.5_f64)));
+    }
+
+    #[test]
+    fn decode_bound_promotes_date_written_before_timestamp_promotion() {
+        // 4-byte date is upscaled to microseconds: 3 days * 86_400_000_000 us/day.
+        let decoded = decode_bound(&3_i32.to_le_bytes(), &TIMESTAMP).unwrap();
+        assert_eq!(decoded, Value::Timestamp(259_200_000_000));
+    }
+
+    #[test]
+    fn decode_bound_decodes_decimal_at_minimal_width() {
+        let data_type = Type::Primitive(PrimitiveType::Decimal {
+            precision: 9,
+            scale: 2,
+        });
+        // 1500 unscaled fits in two big-endian bytes.
+        let decoded = decode_bound(&[0x05, 0xDC], &data_type).unwrap();
+        assert_eq!(
+            decoded,
+            Value::Decimal(decimal_from_i128_with_scale(1500, 2).unwrap())
+        );
+    }
+
+    #[test]
+    fn decode_bound_rejects_malformed_length() {
+        assert!(decode_bound(&[0, 1, 2], &LONG).is_err());
+    }
+
     #[test]
     fn numeric_bounds_decode_against_manifest_schema() {
         let old_schema = StructType::new(vec![