parquet/column/reader/
legacy_fixed_len_byte_array.rs

1// SPDX-License-Identifier: Apache-2.0
2// SPDX-FileCopyrightText: See the NOTICE file distributed with this work for additional information regarding copyright ownership
3// SPDX-FileContributor: Licensed to the Apache Software Foundation (ASF) under one or more contributor license agreements
4
5//! Temporary reader mitigation for [arrow-rs #11261], fixed by #11262.
6//! These historical layouts are not additional Parquet encodings.
7//!
8//! [arrow-rs #11261]: https://github.com/apache/arrow-rs/issues/11261
9//!
10//! The ordinary decoder checks canonical PLAIN sizes before selecting a repair.
11//! A bounded prefix probe can rule out the historical layout, but a match never
12//! authorizes repair without an independent physical count. Repairs happen once
13//! per page; normal value decoding does not need to know the historical layout.
14//! Producer strings and embedded Arrow metadata never select a repair.
15//!
16//! Canonical PLAIN has `N * W` bytes; the historical representation has exactly
17//! `N * (W + 4)` bytes with every little-endian length prefix equal to `W`. These
18//! sizes are disjoint for nonempty pages with the same independently established
19//! `N`. Without that count, the representations can overlap.
20//!
21//! To retire the mitigation, remove this module, the compatibility hook in
22//! `decoder::prepare_v1_fixed_len_byte_array`, and the repair calls in the two
23//! normalization helpers there. Replace noncanonical PLAIN repair with a length
24//! error and reject FLBA/DELTA_LENGTH_BYTE_ARRAY instead of normalizing it.
25//! No changes to the column reader's page setup, read/skip loops, byte-budget
26//! validation or canonical tests are required. Keep physical-value validation:
27//! it detects corruption independently of legacy support. Remove legacy fixture
28//! tests separately.
29
30use bytes::Bytes;
31
32use super::decoder::{PreparedFixedLenByteArrayPage, normalize_fixed_len_byte_array_data};
33use crate::basic::Encoding;
34use crate::data_type::Int32Type;
35use crate::encodings::decoding::{Decoder, DeltaBitPackDecoder};
36use crate::errors::{ParquetError, Result};
37
38/// Handle nullable V1 pages that may need compatibility processing. The caller
39/// supplies a lazy, independent count; only ambiguous PLAIN or degenerate pages
40/// invoke it. Returns `true` if preparation is complete; `false` leaves `page`
41/// unchanged for the caller's ordinary canonical path. Work in place so that
42/// ambiguous but canonical pages do not need additional buffer-reference clones.
43pub(super) fn prepare_v1_page(
44    page: &mut PreparedFixedLenByteArrayPage,
45    num_levels: usize,
46    type_length: usize,
47    count_values: &impl Fn() -> Result<usize>,
48) -> Result<bool> {
49    let (num_values, deferred) = match page.encoding {
50        Encoding::PLAIN
51            if num_levels != 0
52                && type_length != 0
53                && may_be_length_prefixed_plain(&page.data, type_length) =>
54        {
55            // Even all prefixes matching is not permission to repair. The
56            // normalizer must first compare the independently counted canonical size.
57            (count_values()?, false)
58        }
59        Encoding::DELTA_LENGTH_BYTE_ARRAY => {
60            if num_levels == 0 || type_length == 0 {
61                (count_values()?, false)
62            } else {
63                let declared = delta_length_value_count(page.data.clone())?;
64                if declared > num_levels {
65                    return Err(general_err!(
66                        "DELTA_LENGTH_BYTE_ARRAY has more values than levels"
67                    ));
68                }
69                (declared, true)
70            }
71        }
72        _ => return Ok(false),
73    };
74    let (encoding, data) = normalize_fixed_len_byte_array_data(
75        std::mem::take(&mut page.data),
76        num_values,
77        type_length,
78        page.encoding,
79    )?;
80    page.encoding = encoding;
81    page.validate_count = deferred;
82    page.data = data;
83    Ok(true)
84}
85
86/// A negative result rules out the supported historical PLAIN layout. A positive
87/// result only requests an independent count: even every prefix matching could
88/// be an ordinary canonical payload. Bound the probe independently of page size.
89fn may_be_length_prefixed_plain(data: &[u8], type_length: usize) -> bool {
90    let Some(stride) = type_length.checked_add(4) else {
91        return false;
92    };
93    !data.is_empty()
94        && data.len().is_multiple_of(stride)
95        && data
96            .chunks_exact(stride)
97            .take(5)
98            .all(|value| u32::from_le_bytes(value[..4].try_into().unwrap()) as usize == type_length)
99}
100
101/// Read the declared number of lengths, not an independent physical count.
102/// The caller must still check lengths, payload bytes and actual definition levels.
103fn delta_length_value_count(data: Bytes) -> Result<usize> {
104    let mut decoder = DeltaBitPackDecoder::<Int32Type>::new();
105    decoder.set_data(data, 0)?;
106    Ok(decoder.values_left())
107}
108
109/// Remove BYTE_ARRAY length prefixes from a noncanonical PLAIN FLBA section.
110/// The caller has checked `expected_len = num_values * type_length` for overflow
111/// and established that the input does not have that canonical length.
112#[cold]
113pub(super) fn decode_length_prefixed_plain(
114    data: Bytes,
115    num_values: usize,
116    type_length: usize,
117    expected_len: usize,
118    page: &str,
119) -> Result<Bytes> {
120    let stride = type_length.checked_add(4);
121    if stride.and_then(|s| num_values.checked_mul(s)) != Some(data.len()) {
122        return Err(general_err!(
123            "Invalid FIXED_LEN_BYTE_ARRAY {} payload length: expected {} bytes \
124             ({} values of {} bytes), got {}",
125            page,
126            expected_len,
127            num_values,
128            type_length,
129            data.len()
130        ));
131    }
132    // The checked size above guarantees a nonzero, representable stride.
133    let stride = stride.unwrap();
134    for value in data.chunks_exact(stride) {
135        let length = u32::from_le_bytes(value[..4].try_into().unwrap()) as usize;
136        if length != type_length {
137            return Err(general_err!(
138                "Invalid FIXED_LEN_BYTE_ARRAY {} payload length prefix: expected {}, got {}",
139                page,
140                type_length,
141                length
142            ));
143        }
144    }
145
146    // Validate the entire page before allocating or exposing any legacy values.
147    let mut values = Vec::with_capacity(expected_len);
148    for value in data.chunks_exact(stride) {
149        values.extend_from_slice(&value[4..]);
150    }
151    Ok(values.into())
152}
153
154/// Validate the BYTE_ARRAY-only delta-length representation and return its
155/// contiguous fixed-width values. The caller supplies the overflow-checked
156/// canonical payload size and passes the returned value section to PLAIN.
157#[cold]
158pub(super) fn decode_delta_length(
159    data: Bytes,
160    num_values: usize,
161    type_length: usize,
162    expected_len: usize,
163) -> Result<Bytes> {
164    let mut decoder = DeltaBitPackDecoder::<Int32Type>::new();
165    decoder.set_data(data.clone(), num_values)?;
166    if decoder.values_left() != num_values {
167        return Err(general_err!(
168            "Invalid FIXED_LEN_BYTE_ARRAY DELTA_LENGTH_BYTE_ARRAY value count: expected {}, got {}",
169            num_values,
170            decoder.values_left()
171        ));
172    }
173    // Bound scratch space independently of untrusted page counts.
174    let mut lengths = [0_i32; 128];
175    let mut remaining = num_values;
176    while remaining != 0 {
177        let count = remaining.min(lengths.len());
178        let read = decoder.get(&mut lengths[..count])?;
179        if read != count {
180            return Err(general_err!(
181                "Truncated FLBA DELTA_LENGTH_BYTE_ARRAY lengths"
182            ));
183        }
184        for &length in &lengths[..count] {
185            if usize::try_from(length).ok() != Some(type_length) {
186                return Err(general_err!(
187                    "Invalid FIXED_LEN_BYTE_ARRAY DELTA_LENGTH_BYTE_ARRAY value length: expected {}, got {}",
188                    type_length,
189                    length
190                ));
191            }
192        }
193        remaining -= count;
194    }
195    let offset = decoder.get_offset();
196    if data.len().checked_sub(offset) != Some(expected_len) {
197        return Err(general_err!(
198            "Invalid FIXED_LEN_BYTE_ARRAY DELTA_LENGTH_BYTE_ARRAY payload length: expected {} value bytes",
199            expected_len
200        ));
201    }
202    Ok(data.slice(offset..))
203}
204
205#[cfg(test)]
206mod tests {
207    use super::*;
208    use crate::basic::Encoding;
209    use crate::column::reader::decoder::{
210        normalize_fixed_len_byte_array_data, normalize_fixed_len_byte_array_payload,
211    };
212    use crate::column::reader::tests::{
213        check_incremental_fixed_len_byte_array_pages, nullable_fixed_page, nullable_fixed_reader,
214    };
215    use crate::encodings::encoding::{DeltaBitPackEncoder, Encoder};
216    use rand::{prelude::*, rngs::StdRng};
217
218    #[test]
219    fn ordinary_pages_do_not_request_compatibility_counting() {
220        for encoding in [
221            Encoding::PLAIN,
222            Encoding::RLE_DICTIONARY,
223            Encoding::DELTA_BYTE_ARRAY,
224        ] {
225            let mut page = PreparedFixedLenByteArrayPage {
226                encoding,
227                data: Bytes::from_static(b"abcdefghijkl"),
228                validate_count: false,
229            };
230            assert!(
231                !prepare_v1_page(&mut page, 3, 4, &|| {
232                    panic!("ordinary page must not be pre-scanned")
233                })
234                .unwrap()
235            );
236            assert_eq!(page.encoding, encoding);
237            assert_eq!(page.data.as_ref(), b"abcdefghijkl");
238            assert!(!page.validate_count);
239        }
240    }
241
242    #[test]
243    fn matching_prefixes_require_an_independent_count() {
244        let calls = std::cell::Cell::new(0);
245        let raw = Bytes::from(4_u32.to_le_bytes().repeat(10));
246        let mut page = PreparedFixedLenByteArrayPage {
247            encoding: Encoding::PLAIN,
248            data: raw.clone(),
249            validate_count: false,
250        };
251        assert!(
252            prepare_v1_page(&mut page, 20, 4, &|| {
253                calls.set(calls.get() + 1);
254                Ok(10) // ten present values, not twenty logical levels
255            })
256            .unwrap()
257        );
258        assert_eq!(calls.get(), 1);
259        assert_eq!(page.data, raw);
260        assert_eq!(page.encoding, Encoding::PLAIN);
261        assert!(!page.validate_count);
262
263        // Five matching prefixes do not permit accepting a bad sixth prefix.
264        let mut malformed = 4_u32.to_le_bytes().repeat(12);
265        malformed[40..44].fill(0xff);
266        let mut reader = nullable_fixed_reader(vec![nullable_fixed_page(
267            &[1; 6],
268            Encoding::PLAIN,
269            &malformed,
270        )]);
271        let mut values = vec![];
272        assert!(
273            reader
274                .read_records(1, Some(&mut vec![]), None, &mut values)
275                .is_err()
276        );
277        assert!(values.is_empty());
278    }
279
280    #[test]
281    fn incremental_delta_length_count_validation() {
282        check_incremental_fixed_len_byte_array_pages(Encoding::DELTA_LENGTH_BYTE_ARRAY, |count| {
283            let mut lengths = DeltaBitPackEncoder::<Int32Type>::new();
284            lengths.put(&vec![4; count]).unwrap();
285            let mut payload = lengths.flush_buffer().unwrap().to_vec();
286            payload.extend_from_slice(&b"abcd".repeat(count));
287            payload
288        });
289
290        for lengths in [vec![], vec![4; 5], vec![3, 5]] {
291            let mut encoder = DeltaBitPackEncoder::<Int32Type>::new();
292            encoder.put(&lengths).unwrap();
293            let mut payload = encoder.flush_buffer().unwrap().to_vec();
294            payload.extend(std::iter::repeat_n(
295                b'x',
296                lengths.iter().sum::<i32>() as usize,
297            ));
298            let def = if lengths.is_empty() {
299                [0; 4]
300            } else {
301                [1, 0, 1, 0]
302            };
303            let mut reader = nullable_fixed_reader(vec![nullable_fixed_page(
304                &def,
305                Encoding::DELTA_LENGTH_BYTE_ARRAY,
306                &payload,
307            )]);
308            let result = reader.read_records(1, Some(&mut vec![]), None, &mut vec![]);
309            if lengths.is_empty() {
310                assert_eq!(result.unwrap(), (1, 0, 1));
311                assert_eq!(reader.skip_records(3).unwrap(), 3);
312            } else {
313                // Invalid declared counts or individual lengths remain eager errors.
314                assert!(result.is_err());
315            }
316        }
317    }
318
319    #[test]
320    fn prefix_probe_only_rules_out_legacy() {
321        for width in [0, 1, 2, 3, 4, 16, 32] {
322            for count in [0, 1, 5, 6] {
323                let mut data = vec![];
324                for _ in 0..count {
325                    data.extend_from_slice(&(width as u32).to_le_bytes());
326                    data.extend(std::iter::repeat_n(0x55, width));
327                }
328                assert_eq!(may_be_length_prefixed_plain(&data, width), count != 0);
329                if !data.is_empty() {
330                    let mut truncated = data.clone();
331                    truncated.pop();
332                    assert!(!may_be_length_prefixed_plain(&truncated, width));
333                    data[0] ^= 1;
334                    assert!(!may_be_length_prefixed_plain(&data, width));
335                }
336            }
337        }
338        assert!(!may_be_length_prefixed_plain(&[], usize::MAX));
339        let mut data = 4_u32.to_le_bytes().repeat(12);
340        data[40..44].fill(0xff);
341        assert!(may_be_length_prefixed_plain(&data, 4)); // probe is bounded
342    }
343
344    #[test]
345    fn length_prefixed_plain() {
346        let mut rng = StdRng::seed_from_u64(11261);
347        for page in ["dictionary page", "PLAIN data page"] {
348            for width in [0, 1, 4, 8, 16, 32] {
349                for count in [0, 1, 2, 33] {
350                    let raw: Bytes = (0..width * count).map(|_| rng.random::<u8>()).collect();
351                    let mut legacy = vec![];
352                    for i in 0..count {
353                        legacy.extend_from_slice(&(width as u32).to_le_bytes());
354                        legacy.extend_from_slice(&raw[i * width..(i + 1) * width]);
355                    }
356                    assert_eq!(
357                        normalize_fixed_len_byte_array_payload(legacy.into(), count, width, page)
358                            .unwrap(),
359                        raw
360                    );
361                }
362            }
363            // An exact aggregate size is insufficient: every prefix must match.
364            for data in [
365                b"\x03\0\0\0abc\x05\0\0\0defgh".as_slice(),
366                b"\x04\0\0\0abcd\xff\xff\xff\xffefgh",
367                b"\x04\0\0\0abcd\x04\0\0\0efg",
368                b"\x04\0\0\0abcd\x04\0\0\0efghi",
369            ] {
370                assert!(
371                    normalize_fixed_len_byte_array_payload(
372                        Bytes::copy_from_slice(data),
373                        2,
374                        4,
375                        page,
376                    )
377                    .is_err()
378                );
379            }
380        }
381    }
382
383    #[test]
384    fn data_pages() {
385        let length_prefixed = |raw: &[u8]| {
386            let mut legacy = vec![];
387            for value in raw.as_chunks::<4>().0 {
388                legacy.extend_from_slice(&4_u32.to_le_bytes());
389                legacy.extend_from_slice(value);
390            }
391            legacy
392        };
393        super::super::tests::check_fixed_len_byte_array_pages(
394            |raw| {
395                let mut lengths = DeltaBitPackEncoder::<Int32Type>::new();
396                lengths.put(&vec![4; raw.len() / 4]).unwrap();
397                let mut delta = lengths.flush_buffer().unwrap().to_vec();
398                delta.extend_from_slice(raw);
399                vec![
400                    (Encoding::PLAIN, length_prefixed(raw)),
401                    (Encoding::DELTA_LENGTH_BYTE_ARRAY, delta),
402                ]
403            },
404            |raw| {
405                let mut bad_prefix = length_prefixed(raw);
406                if raw.is_empty() {
407                    // All-null pages must have an empty PLAIN value section.
408                    bad_prefix.extend_from_slice(&4_u32.to_le_bytes());
409                } else {
410                    bad_prefix[0] = 3;
411                }
412                vec![bad_prefix]
413            },
414        );
415    }
416
417    #[test]
418    fn delta_length() {
419        let encode = |lengths: &[i32], values: &[u8]| {
420            let mut encoder = DeltaBitPackEncoder::<Int32Type>::new();
421            encoder.put(lengths).unwrap();
422            let mut data = encoder.flush_buffer().unwrap().to_vec();
423            data.extend_from_slice(values);
424            Bytes::from(data)
425        };
426        let normalize = |data, count| {
427            normalize_fixed_len_byte_array_data(data, count, 4, Encoding::DELTA_LENGTH_BYTE_ARRAY)
428        };
429        for count in [0, 1, 2, 129, 2051] {
430            let values = b"abcd".repeat(count);
431            let data = encode(&vec![4; count], &values);
432            let (encoding, result) = normalize(data.clone(), count).unwrap();
433            assert_eq!(encoding, Encoding::PLAIN);
434            assert_eq!(result.as_ref(), values);
435            assert_eq!(
436                result.as_ptr(),
437                data.as_ptr().wrapping_add(data.len() - result.len())
438            );
439        }
440        for (lengths, values, count) in [
441            (vec![3, 5], b"abcdefgh".as_slice(), 2),
442            (vec![4, -1], b"abcdefgh", 2),
443            (vec![4, 4], b"abcdefgh", 1),
444            (vec![4, 4], b"abcdefg", 2),
445            (vec![4, 4], b"abcdefghi", 2),
446            (vec![], b"x", 0),
447        ] {
448            assert!(normalize(encode(&lengths, values), count).is_err());
449        }
450        assert!(normalize(Bytes::new(), 1).is_err());
451        assert!(normalize(Bytes::new(), usize::MAX).is_err());
452    }
453}
parquet/column/reader/legacy_fixed_len_byte_array.rs

parquet/column/reader/
legacy_fixed_len_byte_array.rs
