GH-47981: [C++][Parquet] Add compatibility with non-compliant RLE stream - #47992

…LE stream (apache#47992)

### Rationale for this change

RLE-bit-packed streams are required by the Parquet spec to have 8-padded bit-packed runs, but some non-compliant encoders (such as Polars versions before pola-rs/polars#13883) might generate a truncated last bit-packed run, which nevertheless contains enough *logical* values.

### What changes are included in this PR?

1. Compatibility code for non-compliant RLE streams as described above
2. Guard against zero-size dictionaries to avoid hitting an assertion in `DictionaryConverter`

### Are these changes tested?

Yes, by additional unit tests.

### Are there any user-facing changes?

No, except a bugfix.

* GitHub Issue: apache#47981

Authored-by: Antoine Pitrou <[email protected]>
Signed-off-by: Antoine Pitrou <[email protected]>

### What changes were proposed in this pull request?

This is a follow-up of SPARK-59812. It makes `VectorizedRleValuesReader`, which decodes definition/repetition levels, dictionary ids and RLE booleans, fail with a `ParquetDecodingException` when the encoded data ends early or a header is invalid.

- `readNextGroup()` throws when there is no encoded data left, instead of returning false. Callers never read more values than the page declares, so running out of data always means a corrupt page. All the `if (currentCount == 0 && !readNextGroup()) break;` loop guards become `if (currentCount == 0) readNextGroup();`. With a bit width of 0, `initFromPage` already provides every value of the page, so reaching `readNextGroup()` also means reading past the end.
- `readNextGroup()` skips runs of length 0, so `currentCount` is always positive when it returns. The existing group decoding moves unchanged into a `readGroup()` helper that `readNextGroup()` calls in a loop. The Parquet spec does not allow runs of length 0; skipping them keeps what the batch paths already did, so no file that could be read before fails because of them.
- For a bit-packed run, check that `numGroups * 8` does not overflow and that `numGroups * bitWidth` bytes are left before allocating the value buffer.
- `initFromPage` rejects a level length that is cut off, negative or larger than the bytes left in the page, and a dictionary id bit width above 32.
- `VectorizedColumnReader` now also adds the page or dictionary and the column to a `ParquetDecodingException` thrown while initializing a page (in `readPageV1` and `readPageV2`) or decoding the dictionary page, as it already did for an `IOException`, so a corrupt length still reports the column. It wraps it in a new `ParquetDecodingException` that includes the original message, so the exception type stays the same.
- A read error inside a run header or an RLE value (e.g. the data ends in the middle of a header) now also starts with "Corrupted RLE data".
- A dictionary-encoded page with no data at all (not even the bit width byte) no longer decodes every value as id 0. Reading from it fails, like parquet-java's `DictionaryValuesReader` ("Attempt to read from empty page"). An all-null page, which reads no ids, still works.

The added checks run once per run or page, not per value.

### Why are the changes needed?

SPARK-34863 made `readNextGroup()` return false at the end of the stream. Before that, it called `in.read()`, which throws an `EOFException` that was wrapped in a `ParquetDecodingException`. Since then, on a corrupt page:

- The batch read and skip loops stop early, so dictionary ids are only partially read or skipped, and the query returns wrong results without an error.
- For definition and repetition levels, `readBatch` and `readBatchRepeated` return without making progress, and `VectorizedColumnReader.readBatch` loops forever because `rowsToReadInBatch` and `valuesToReadInPage` never change.
- For repeated columns, when only the definition levels end early, the missing definition levels are silently left as 0, which produces wrong nested values.
- `readBooleans` ignores the return value, and loops forever at the end of the stream.
- `readInteger` ignores the return value and decrements `currentCount` below 0, returning the last run's value again. It also returns the value of a run of length 0 instead of skipping it.
- A bit-packed header with a large group count overflows `numGroups * 8` and `numGroups * bitWidth` (the latter was added by SPARK-56895). `in.slice` with a negative length moves a `SingleBufferInputStream` backwards, and a large count allocates up to an `int[2^31 - 8]` buffer before noticing that the bytes are missing.
- In Parquet 1.18.1, `SingleBufferInputStream.sliceBuffers` only rejects `length > remaining`, so a negative level length moves the stream backwards and the next section of the page is decoded from the wrong offset.

When the encoded data ends early, the non-vectorized reader already fails: parquet-java's `RunLengthBitPackingHybridDecoder` throws "Reading past RLE/BitPacking stream.".

This PR does not change how a last bit-packed run that is shorter than its padding is handled. Some non-compliant writers produce it (e.g. Polars before pola-rs/polars#13883). The vectorized reader already rejected it before this change, because `in.slice` threw an `EOFException`, while parquet-java and Arrow C++ (apache/arrow#47992) accept it when the bytes cover the values that are read. Accepting it is left for a follow-up, SPARK-59853.

A bit-packed run whose bytes are all present is still not bounded by the number of values in the page, so a corrupt header can still make the reader allocate a buffer much larger than the page needs. Rejecting such runs would also fail non-compliant pages that decode correctly today (e.g. 9 values written as a run of 3 groups), so bounding them is left for a follow-up.

### Does this PR introduce _any_ user-facing change?

Only for corrupt or non-compliant Parquet files: reading a page whose levels, dictionary ids or RLE booleans end early, or have an invalid length, bit width or bit-packed header, now fails with a `ParquetDecodingException` instead of returning wrong values, spinning forever, or allocating a buffer for a bit-packed run whose bytes are missing. Valid files are not affected. With `spark.sql.files.ignoreCorruptFiles=true`, the rest of such a file is now skipped.

A few of these pages could be read before and give the right result by chance, and now fail:
- Dictionary ids (or RLE boolean values) that end early, when every row after the cut is skipped by column index filtering, e.g. the rest of the page before a row range in a later page. The skip stopped silently at the end of the data. This was possible since 3.3.0 (SPARK-34863); 3.2 failed the same way as this PR.
- A dictionary-encoded page whose data is empty (no bit width byte, which the spec requires), when all its non-null rows are skipped, or when the dictionary has a single entry, so that decoding every id as 0 happened to be right. This was the behavior since the vectorized reader was added in 2.0.0.

Selecting rows after the cut, or a dictionary with more than one entry, returns wrong values for the same pages without this PR. With `spark.sql.files.ignoreCorruptFiles=true`, such a file is no longer read in full: from the batch that reaches such a page, the rest of the file, including later row groups, is skipped with only a warning in the log.

### How was this patch tested?

New unit tests in `VectorizedRleValuesReaderSuite`. The ones on hand-built dictionary id, boolean and length sections (9 of 14) run both on a single buffer and on the same data split into 1-byte buffers, so that every part of the data longer than 1 byte, including a slice of it, is read from a `MultiBufferInputStream`. Negative tests check the expected message, and the ones that read values run on a separate thread with a timeout, since some looped forever before this change.
- Truncated dictionary ids on `readIntegers`, `skipIntegers` and `readInteger`; truncated booleans on `readBooleans`; truncated definition levels on `readBatch` (with and without definition level output, and with row ranges that skip past the truncation); truncated repetition levels on `readBatchRepeated`; truncated definition levels on `readBatchRepeated` (with and without row ranges); reading past a page with bit width 0.
- Cut-off, negative and oversized (up to `Int.MaxValue`) level lengths, and dictionary id bit widths above 32.
- Bit-packed runs that are truncated, that overflow `currentCount` to 0 or a negative value, or that would allocate a huge buffer.
- Data ending inside a run header or an RLE value.
- A dictionary-encoded page with no data, and runs of length 0.
- Positive cases: reads and skips that end exactly at the end of the encoded values for mixed RLE and bit-packed dictionary ids and booleans, with a level length that covers exactly the rest of the page.

New end-to-end tests in `ParquetIOSuite` corrupt a written file: a shortened definition level run fails instead of hanging, and an oversized length of the definition levels (V1 page) or of RLE boolean values (V2 page) reports the page and column once. A new test in `ParquetVectorizedSuite` checks that a corrupted dictionary page reports the column.

All the new negative tests fail without this change. Existing suites also pass: `ParquetEncodingSuite`, `ParquetVectorizedSuite`, `ParquetColumnIndexSuite`, `ParquetIOSuite`, `ParquetV1QuerySuite`, `ParquetV2QuerySuite`, `ParquetDeltaByteArrayEncodingSuite`, `ParquetDeltaLengthByteArrayEncodingSuite`, `VectorizedPlainValuesReaderSuite` and `ParquetVectorUpdaterSuite`.

### Was this patch authored or co-authored using generative AI tooling?

Generated-by: Claude Code (Claude Opus 5.5)

Closes #59117 from viirya/SPARK-59832-rle-truncated.

Authored-by: Liang-Chi Hsieh <[email protected]>
Signed-off-by: Liang-Chi Hsieh <[email protected]>