parquet/column/reader/
legacy_fixed_len_byte_array.rs

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

parquet/column/reader/
legacy_fixed_len_byte_array.rs
