perf(extension): reserve packed scalar buffers

.changes/unreleased/changed-20261019-packed-extension-reserve.yaml

@@ -0,0 +1,4 @@
+kind: Changed
+body: |-
+    **Reduce allocations in packed extension encoding and decoding.** Packed extension encoding starts with a conservative one-byte-per-element output capacity, and large fixed-width packed decoders reserve space for complete elements before appending. Incomplete trailing bytes retain their existing behavior.
+time: 2026-10-19T18:46:00.000000000+00:00

benchmarks/buffa/Cargo.toml

@@ -43,6 +43,8 @@ iso = []
 # `encode_sink` gates the sink-comparison bench (`task bench-encode-sink`) out of
 # the default run for the same reason: its IDs must not enter saved baselines.
 encode_sink = []
+# Keep the packed-extension microbenchmark out of the default suite.
+packed_extension = []
 
 [[bench]]
 name = "protobuf"
@@ -124,3 +126,8 @@ required-features = ["iso", "column_batch"]
 name = "encode_sink"
 harness = false
 required-features = ["encode_sink", "api_response", "log_record", "google_message1", "media_frame"]
+
+[[bench]]
+name = "packed_extension"
+harness = false
+required-features = ["packed_extension"]

benchmarks/buffa/benches/packed_extension.rs

@@ -0,0 +1,153 @@
+// Measure packed fixed-width extension decoding and packed scalar encoding.
+//
+// Run separately from the default benchmark suite to avoid changing saved
+// benchmark IDs:
+//
+//   cd benchmarks/buffa
+//   cargo bench --features packed_extension --bench packed_extension
+//
+// For a fair base-vs-branch comparison, run this same benchmark file against
+// both implementations on the same machine and Rust toolchain.
+use buffa::extension::codecs::{Fixed32, Fixed64, Int32, PackedRepeated, Repeated};
+use buffa::unknown_fields::{UnknownField, UnknownFieldData, UnknownFields};
+use buffa::ExtensionCodec;
+use criterion::{
+    black_box, criterion_group, criterion_main, BatchSize, BenchmarkId, Criterion, Throughput,
+};
+
+const FIELD_NUMBER: u32 = 7;
+
+fn packed_fixed32(count: usize) -> UnknownFields {
+    let mut payload = Vec::with_capacity(count * 4);
+    for i in 0..count {
+        payload.extend_from_slice(&(i as u32).to_le_bytes());
+    }
+    let mut fields = UnknownFields::new();
+    fields.push(UnknownField {
+        number: FIELD_NUMBER,
+        data: UnknownFieldData::LengthDelimited(payload),
+    });
+    fields
+}
+
+fn packed_fixed64(count: usize) -> UnknownFields {
+    let mut payload = Vec::with_capacity(count * 8);
+    for i in 0..count {
+        payload.extend_from_slice(&(i as u64).to_le_bytes());
+    }
+    let mut fields = UnknownFields::new();
+    fields.push(UnknownField {
+        number: FIELD_NUMBER,
+        data: UnknownFieldData::LengthDelimited(payload),
+    });
+    fields
+}
+
+fn packed_extension(c: &mut Criterion) {
+    let mut group = c.benchmark_group("extension/decode_packed_fixed");
+    for count in [1usize, 4, 8, 128, 4096, 16_384, 32_768, 65_536] {
+        let fields32 = packed_fixed32(count);
+        let fields64 = packed_fixed64(count);
+        group.throughput(Throughput::Elements(count as u64));
+
+        group.bench_with_input(
+            BenchmarkId::new("fixed32", count),
+            &fields32,
+            |b, fields| {
+                b.iter(|| black_box(Repeated::<Fixed32>::decode(FIELD_NUMBER, black_box(fields))));
+            },
+        );
+        group.bench_with_input(
+            BenchmarkId::new("fixed64", count),
+            &fields64,
+            |b, fields| {
+                b.iter(|| black_box(Repeated::<Fixed64>::decode(FIELD_NUMBER, black_box(fields))));
+            },
+        );
+    }
+    group.finish();
+}
+
+fn packed_extension_encode(c: &mut Criterion) {
+    let mut group = c.benchmark_group("extension/encode_packed_scalar");
+    for count in [8usize, 128, 4_096] {
+        let fixed32: Vec<u32> = (0..count as u32).collect();
+        let fixed64: Vec<u64> = (0..count as u64).collect();
+        let varint: Vec<i32> = (0..count as i32).collect();
+        let negative_varint: Vec<i32> = (0..count as i32).map(|value| -value - 1).collect();
+        group.throughput(Throughput::Elements(count as u64));
+
+        group.bench_with_input(BenchmarkId::new("fixed32", count), &fixed32, |b, values| {
+            b.iter_batched(
+                || (*values).clone(),
+                |input| {
+                    let mut fields = UnknownFields::new();
+                    PackedRepeated::<Fixed32>::try_encode(
+                        FIELD_NUMBER,
+                        black_box(input),
+                        &mut fields,
+                    )
+                    .unwrap();
+                    black_box(fields);
+                },
+                BatchSize::SmallInput,
+            );
+        });
+        group.bench_with_input(BenchmarkId::new("fixed64", count), &fixed64, |b, values| {
+            b.iter_batched(
+                || (*values).clone(),
+                |input| {
+                    let mut fields = UnknownFields::new();
+                    PackedRepeated::<Fixed64>::try_encode(
+                        FIELD_NUMBER,
+                        black_box(input),
+                        &mut fields,
+                    )
+                    .unwrap();
+                    black_box(fields);
+                },
+                BatchSize::SmallInput,
+            );
+        });
+        group.bench_with_input(BenchmarkId::new("varint", count), &varint, |b, values| {
+            b.iter_batched(
+                || (*values).clone(),
+                |input| {
+                    let mut fields = UnknownFields::new();
+                    PackedRepeated::<Int32>::try_encode(
+                        FIELD_NUMBER,
+                        black_box(input),
+                        &mut fields,
+                    )
+                    .unwrap();
+                    black_box(fields);
+                },
+                BatchSize::SmallInput,
+            );
+        });
+        group.bench_with_input(
+            BenchmarkId::new("int32_negative", count),
+            &negative_varint,
+            |b, values| {
+                b.iter_batched(
+                    || (*values).clone(),
+                    |input| {
+                        let mut fields = UnknownFields::new();
+                        PackedRepeated::<Int32>::try_encode(
+                            FIELD_NUMBER,
+                            black_box(input),
+                            &mut fields,
+                        )
+                        .unwrap();
+                        black_box(fields);
+                    },
+                    BatchSize::SmallInput,
+                );
+            },
+        );
+    }
+    group.finish();
+}
+
+criterion_group!(benches, packed_extension, packed_extension_encode);
+criterion_main!(benches);

buffa/src/extension.rs

@@ -758,6 +758,11 @@ pub mod codecs {
                     }
                 }
                 fn decode_packed(bytes: &[u8], out: &mut Vec<$ty>) {
+                    // Full elements are known from the wire length. Avoid
+                    // repeated Vec growth for long packed extension fields.
+                    if bytes.len() >= 64 {
+                        out.reserve(bytes.len() / 4);
+                    }
                     let mut buf = bytes;
                     while buf.remaining() >= 4 {
                         let $d = buf.get_u32_le();
@@ -817,6 +822,9 @@ pub mod codecs {
                     }
                 }
                 fn decode_packed(bytes: &[u8], out: &mut Vec<$ty>) {
+                    if bytes.len() >= 128 {
+                        out.reserve(bytes.len() / 8);
+                    }
                     let mut buf = bytes;
                     while buf.remaining() >= 8 {
                         let $d = buf.get_u64_le();
@@ -1156,7 +1164,10 @@ pub mod codecs {
             if value.is_empty() {
                 return Ok(());
             }
-            let mut buf = Vec::new();
+            // Each packed scalar needs at least one wire byte. Keep a minimum
+            // capacity of eight bytes for tiny vectors, then use that lower
+            // bound for larger vectors.
+            let mut buf = Vec::with_capacity(value.len().max(8));
             for v in &value {
                 C::encode_packed(v, &mut buf);
             }
@@ -1837,6 +1848,28 @@ mod tests {
         assert_eq!(c.extension(&E), vec![1_u32, 2_u32]);
     }
 
+    #[test]
+    fn repeated_fixed64_packed_preserves_mixed_records_and_partial_tails() {
+        const E: Extension<Repeated<Fixed64>> = Extension::new(1, CARRIER);
+        for tail_len in 0..8 {
+            let mut c = Carrier::default();
+            c.unknown.push(UnknownField {
+                number: 1,
+                data: UnknownFieldData::Fixed64(u64::MAX),
+            });
+            let mut bytes = Vec::new();
+            bytes.extend_from_slice(&0x0102_0304_0506_0708_u64.to_le_bytes());
+            bytes.extend_from_slice(&0_u64.to_le_bytes());
+            bytes.extend_from_slice(&[0xAB; 7][..tail_len]);
+            c.unknown.push(ld(1, bytes));
+            c.unknown.push(UnknownField {
+                number: 1,
+                data: UnknownFieldData::Fixed64(7),
+            });
+            assert_eq!(c.extension(&E), vec![u64::MAX, 0x0102_0304_0506_0708, 0, 7]);
+        }
+    }
+
     #[test]
     fn repeated_scalar_wrong_wire_type_skipped() {
         const E: Extension<Repeated<Int32>> = Extension::new(1, CARRIER);
@@ -1943,6 +1976,91 @@ mod tests {
         assert_eq!(c.extension(&E), vec![-1_i64, 0, i64::MAX]);
     }
 
+    #[test]
+    fn packed_fixed_width_extenders_append_only_complete_elements() {
+        for count in [0usize, 1, 15, 16, 17, 1024] {
+            let words32: Vec<u32> = (0..count).map(|i| 0xbf80_0000u32 + i as u32).collect();
+            let mut bytes32 = Vec::with_capacity(count * 4 + 3);
+            for &word in &words32 {
+                bytes32.extend_from_slice(&word.to_le_bytes());
+            }
+            bytes32.extend_from_slice(&[0xaa, 0xbb, 0xcc]);
+
+            let mut fixed32 = vec![7u32];
+            Fixed32::decode_packed(&bytes32, &mut fixed32);
+            assert_eq!(fixed32[0], 7);
+            assert_eq!(&fixed32[1..], words32.as_slice());
+
+            let mut sfixed32 = Vec::new();
+            Sfixed32::decode_packed(&bytes32, &mut sfixed32);
+            assert_eq!(sfixed32.len(), count);
+            for (actual, &expected) in sfixed32.iter().zip(&words32) {
+                assert_eq!(*actual as u32, expected);
+            }
+
+            let mut floats = Vec::new();
+            Float::decode_packed(&bytes32, &mut floats);
+            assert_eq!(floats.len(), count);
+            for (actual, &expected) in floats.iter().zip(&words32) {
+                assert_eq!(actual.to_bits(), expected);
+            }
+
+            let words64: Vec<u64> = (0..count)
+                .map(|i| 0xbff0_0000_0000_0000u64 + i as u64)
+                .collect();
+            let mut bytes64 = Vec::with_capacity(count * 8 + 7);
+            for &word in &words64 {
+                bytes64.extend_from_slice(&word.to_le_bytes());
+            }
+            bytes64.extend_from_slice(&[1, 2, 3, 4, 5, 6, 7]);
+
+            let mut fixed64 = vec![9u64];
+            Fixed64::decode_packed(&bytes64, &mut fixed64);
+            assert_eq!(fixed64[0], 9);
+            assert_eq!(&fixed64[1..], words64.as_slice());
+
+            let mut sfixed64 = Vec::new();
+            Sfixed64::decode_packed(&bytes64, &mut sfixed64);
+            assert_eq!(sfixed64.len(), count);
+            for (actual, &expected) in sfixed64.iter().zip(&words64) {
+                assert_eq!(*actual as u64, expected);
+            }
+
+            let mut doubles = Vec::new();
+            Double::decode_packed(&bytes64, &mut doubles);
+            assert_eq!(doubles.len(), count);
+            for (actual, &expected) in doubles.iter().zip(&words64) {
+                assert_eq!(actual.to_bits(), expected);
+            }
+        }
+    }
+
+    #[test]
+    fn packed_fixed_width_extension_accepts_mixed_records() {
+        const E: Extension<PackedRepeated<Fixed32>> = Extension::new(1, CARRIER);
+        let mut c = Carrier::default();
+        c.unknown.push(fixed32(1, 9));
+        c.unknown.push(ld(1, vec![10, 0, 0, 0, 11, 0, 0, 0, 0xee]));
+        c.unknown.push(varint(1, 99)); // wrong wire type
+        c.unknown.push(fixed32(1, 12));
+        assert_eq!(c.extension(&E), vec![9, 10, 11, 12]);
+    }
+
+    #[test]
+    fn packed_extension_encodes_large_scalar_vectors() {
+        const FIXED: Extension<PackedRepeated<Fixed32>> = Extension::new(1, CARRIER);
+        const VARINT: Extension<PackedRepeated<Int32>> = Extension::new(2, CARRIER);
+        let mut c = Carrier::default();
+        let fixed: Vec<u32> = (0..1024).collect();
+        let varint: Vec<i32> = (0..1024).collect();
+
+        c.set_extension(&FIXED, fixed.clone());
+        c.set_extension(&VARINT, varint.clone());
+        assert_eq!(c.extension(&FIXED), fixed);
+        assert_eq!(c.extension(&VARINT), varint);
+        assert_eq!(c.unknown.len(), 2);
+    }
+
     // ── MessageCodec ────────────────────────────────────────────────────────
 
     // Minimal test message: two i32 fields (numbers 1 and 2). Unknown fields