[SPARK-59832][SQL] Fail on truncated input in the vectorized RLE reader - #59117

Nit: missing blank line before this test.

Fixed, thanks.

This check also rejects a final bit-packed run whose trailing zero padding is missing, even when every value the page needs is present. parquet-java's RunLengthBitPackingHybridDecoder tolerates this (bytesToRead = Math.min(bytesToRead, in.available())), and Arrow C++ accepts it since apache/arrow#47992. Such files exist in the wild: Polars before pola-rs/polars#13883 (see pola-rs/polars#13818) and Impala 1.1.1.

It's not a regression, because the previous in.slice(totalBytes) threw an EOFException too. But the description says "The non-vectorized reader already fails on these pages", which isn't true here, and the "Truncated last group" test now pins the difference: [4][0x05][0x11 x 5] read as 10 values decodes fine with spark.sql.parquet.enableVectorizedReader=false. Could we fix the description, and maybe file a follow-up to accept a short last group the way Arrow does (only values fully covered by the remaining bytes, only within the last group)?

Good catch, the description was wrong there. I updated it: this PR keeps rejecting a short last group as before (in.slice already threw), and notes that parquet-java and Arrow C++ accept it. The test case now has a comment saying the same. I filed SPARK-59853 to accept it the way Arrow does.

None of the new tests reach the new throw sites in readValues (L649) or skipValues(int, ParquetReadState, ...) (L743). This test truncates only the repetition levels, and in the flat row-range case readBatchInternal never calls skipValues with more than currentCount, so it throws at L220 instead. On master, truncated definition levels in the repeated path don't hang; they silently produce wrong nested values because the trailing def levels stay 0. Could you add a readBatchRepeated case that truncates only the def-level stream, with and without row indexes?

Added "truncated definition levels fail in readBatchRepeated", with and without row indexes. They reach the throws in readValues and in skipValues(int, ParquetReadState, ...) respectively. On master this case returns without an error, so I also added it to the description.

Before this change, a positive length larger than the rest of the page made sliceStream throw an EOFException, and readPageV1 (or initDataReader for RLE booleans) wrapped it as "could not read page ... in col ...". The new ParquetDecodingException is unchecked, so it skips those wrappers and the error no longer says which column is corrupt. It matches SPARK-59812, but for wide tables it may be worth also catching ParquetDecodingException there (or in VectorizedColumnReader.readBatch) and adding the column descriptor.

Agreed. readPageV1 and initDataReader now also catch ParquetDecodingException and wrap it with the page and column like an IOException, and a new end-to-end test checks that an oversized definition level length reports the column. Decoding errors thrown later from readBatch never had the column, including the ones from SPARK-59812, so I filed SPARK-59854 for that.

When a page is 3 bytes or less, bytes.grouped(3) yields one chunk, and ByteBufferInputStream.wrap(List) then returns a SingleBufferInputStream. So the "multiple buffers" variant is single-buffer for the truncated dictionary id page ([4, 20, 3]). In the boolean test, the RLE bytes also end up in a single buffer after sliceStream(2). As a result, no test hits the end-of-data check on a MultiBufferInputStream. Could we reuse VectorizedPlainValuesReaderSuite.toStream(bytes, split = true), which asserts that the page spans more than one buffer, or split into 1-byte buffers?

Right, thanks. streams now splits into 1-byte buffers, so any part of the page longer than 1 byte, including a slice of it, stays on a MultiBufferInputStream. I didn't reuse VectorizedPlainValuesReaderSuite.toStream because its 3-byte split has the same problem after a slice.

For backports: wrapping the whole body in do { ... } while (currentCount == 0) re-indents about 20 unchanged lines. branch-4.2/4.1/4.0 still slice per group (they don't have SPARK-56895), so the cherry-pick conflicts in the PACKED loop. If the current body becomes a readGroup() helper called from do { readGroup(); } while (currentCount == 0);, and the new checks are added without touching the existing int totalBytes = numGroups * bitWidth; in.slice(totalBytes) lines (use the long multiplication only inside the check), it applies cleanly to every maintenance branch in my simulation. Either way, the backports need to keep the numGroups > Integer.MAX_VALUE / 8 || totalBytes > in.available() pre-check before currentBuffer is allocated.

Done. The existing body is now readGroup() and readNextGroup() loops over it. The int totalBytes = numGroups * bitWidth; in.slice(totalBytes) lines are unchanged, and the long multiplication is only in the new check, before currentBuffer is allocated.

The "fail on truncated input" goal has one remaining gap: a run header whose varint is truncated at end-of-page still hangs rather than throwing.

This guard rejects an empty stream, but a stream ending in a lone continuation byte (e.g. a trailing 0x80) passes it, and the readGroup() below then loops forever in readUnsignedVarInt(): at EOF in.read() returns -1, and (-1 & 0x80) != 0 is always true, so the do/while never terminates. It's reachable on a corrupt V1 level section (e.g. a declared length of 1 whose only byte is 0x80).

This is pre-existing — the old readNextGroup had the same guard and called the same readUnsignedVarInt — so it's a "fold in here or follow-up?" like the short-last-group case. The minimal fix is to treat EOF as corruption in the varint loop:

b = in.read();
if (b < 0) {
  throw new ParquetDecodingException(
    "Corrupted RLE data: reading past the end of the encoded values");
}

I checked this one, and it doesn't hang. In parquet-java 1.18.1 (and 1.17.0), SingleBufferInputStream.read() and MultiBufferInputStream.read() throw EOFException at the end of the stream instead of returning -1, and these are the only stream types that BytesInput.toInputStream() and sliceStream() produce. So readUnsignedVarInt() stops there and readGroup() wraps it.

On this PR's head, a dictionary id page [4, 0x80] and a V1 level section with a declared length of 1 whose only byte is 0x80 both fail with ParquetDecodingException: Corrupted RLE data: failed to read from input stream (cause EOFException), on single-buffer and multi-buffer streams. The reader before this PR fails the same way, with the old message.

There is no test for this path though, so I asked for one in the comment on L1048. Its 30s timeout would catch a hang if the stream behavior ever changed.

Thanks for checking, you're right. In parquet-common 1.18.1 both SingleBufferInputStream.read() and MultiBufferInputStream.read() throw EOFException at the end of the stream instead of returning -1, so a truncated header fails with "Corrupted RLE data: failed to read from input stream" rather than hanging. I withdraw this one; a test like the [4, 0x80] case you suggested on L1048 would pin it.

Thanks for confirming. I added "data ending inside a run header or an RLE value fails" with [4, 0x80] and [4, 0x14], on single and 1-byte buffers; it runs with a timeout, so a hang would fail it too.

No test reaches this path. Every new negative case ends in one of the explicit checks, and the old truncated-last-group input is now caught by the byte-count check above, so restoring the old message here still passes all 26 tests in VectorizedRleValuesReaderSuite.

Could you add a case where the data ends inside a run header or an RLE value? For example, through streams + interceptCorrupted:

• [4, 0x80]: the run header's continuation byte is missing
• [4, 0x14]: an RLE run of 10 without its value byte

Both give "Corrupted RLE data: failed to read from input stream" (cause EOFException) on the single-buffer and the 1-byte-buffer streams, and the 30s timeout in interceptCorrupted would also cover the concern in the L995 thread.

Added "data ending inside a run header or an RLE value fails" with both inputs, on single and 1-byte buffers.

Since the generic IOException fallback now has the same "Corrupted RLE data" prefix, this assertion can't tell which check rejected the input, so the two new pre-allocation checks aren't pinned:

• Without the (long) numGroups * bitWidth > in.available() check, the suite still passes with a heap of 5g or more: the 1L << 27 case really allocates int[1 << 30] and then fails in in.slice with the same prefix. With the default -Xmx4g it fails only after 30s with a TimeoutException, because the OutOfMemoryError never completes the Future.
• Without the numGroups > Integer.MAX_VALUE / 8 check, all tests pass: both overflow inputs are also rejected by the byte-count check.

Could interceptCorrupted take the expected message fragment, e.g. "needs 134217728 bytes, but only 3 bytes are left", "is too long", "reading past the end"? For the same reason, the bit-width-0 test could use a small header for trailing (e.g. Array[Byte](3, 1, 2, 3)), so that a regression of that guard fails with an assertion instead of a 4 GiB allocation.

Done. interceptCorrupted now takes the expected message, so each case pins the check that rejects it, e.g. "is too long" and "needs 134217728 bytes, but only 3 bytes are left". The bit width 0 test uses [3, 1, 2, 3] now.

No test goes through this catch. Both end-to-end tests corrupt the definition levels of a V1 page: the truncated run throws at read time, and the invalid length is wrapped by the readPageV1 catch. Changing this line back to catch (IOException e) doesn't fail anything.

This catch is what gives V2 pages the column, because readPageV2 still catches only IOException. For a boolean column written with parquet.writer.version=PARQUET_2_0 (RLE values) whose 4-byte length is overwritten with FF FF FF 7F, I get

could not read page Page V2 [...] in col [b] required boolean b
  <- could not read page in col [b] required boolean b
  <- Corrupted RLE data: invalid length 2147483647, 3 bytes are left in the page

and without this line only FAILED_READ_FILE <- ParquetDecodingException. Could readCorruptedDefinitionLevels take the column expression and the writer version, and cover this case too? With id >= 0 as b the same byte pattern 03 00 00 00 C8 01 01 matches the boolean data section. Note that the ParquetDecodingException is two levels below the first "could not read page" IOException there.

Thanks. Following your double-wrapping nit, I moved the catch from initDataReader to readPageV2, so V1 and V2 both get one "could not read page ... in col ..." directly on top of the ParquetDecodingException. readCorruptedRleSection now takes the column and the writer version, a new test covers the RLE boolean length in a V2 page, and both length tests check that the page error appears once.

Nit (pre-existing): a corrupted bit-width byte (33..255) still fails with IllegalArgumentException: bitWidth must be >= 0 and <= 32 from init(). That isn't caught by the widened catches, so the error has neither the "Corrupted RLE data" prefix nor the page/column, unlike the level length just above. Since in.read() returns 0..255, something like

int bitWidth = in.read();
if (bitWidth > 32) {
  throw new ParquetDecodingException("Corrupted RLE data: invalid bit width " + bitWidth);
}
init(bitWidth);

would let the existing catch in initDataReader add the column. Fine to leave it for SPARK-59854 if you prefer.

Done. It now throws "Corrupted RLE data: invalid bit width N", which readPageV1/readPageV2 wrap with the page and column.

A question on skipping runs of length 0 instead of rejecting them. parquet-java's encoder never writes an empty run and Arrow C++ treats a zero count as invalid, so in practice an empty run means a corrupted header. With skipping, the same bytes decode differently depending on the reader: for a dictionary id section [2][00 03][C8 01 01] (an empty RLE run of id 3, then 100 x id 1), the vectorized reader returns dictionary[1] for all 100 rows, and spark.sql.parquet.enableVectorizedReader=false returns dictionary[3], both without an error.

Would it be simpler to throw "Corrupted RLE data" for a run of length 0? Then the loop isn't needed. The batch paths already skipped such runs before this PR, so keeping it is defensible too; I just want to make sure it is a deliberate choice.

It's deliberate. The spec doesn't allow runs of length 0 (both run lengths must be in [1, 2^31 - 1]), and Arrow C++ rejects them, but the batch paths already skipped them before this PR, so rejecting them could make a file that reads today fail. Skipping them in readInteger too makes all paths agree. The non-vectorized result in your example (dictionary[3] for all rows) comes from RunLengthBitPackingHybridDecoder.readInt decrementing currentCount to -1 and never reading the next header, so it isn't a useful reference either. I added a line on this to the description.

This bounds the run by the bytes that are left, but not by the number of values in the page, so a bit-packed run whose bytes are all present can still allocate far more than the page needs. For example, a dictionary id section [01][PACKED header for 2^26 groups][2^26 zero bytes] in a page with valueCount = 1 passes both checks, allocates int[2^29] (2 GiB) and unpacks about 537M ids to return one. 64 MiB of zeros compresses to about 65 KB with GZIP, so a small file can do this in every task.

It isn't a regression, since parquet-java's RunLengthBitPackingHybridDecoder also allocates new int[numGroups * 8] without a bound. But initFromPage already receives valueCount, so keeping it and adding something like

if (numGroups > (valueCount + 7L) / 8) {
  throw new ParquetDecodingException("Corrupted RLE data: bit-packed run of " + numGroups +
    " groups is longer than the " + valueCount + " values in the page");
}

would cap the buffer at the page's value count with one more comparison per run. A run of a valid stream never has more than ceil(valueCount / 8) groups.

I tried this, but it also rejects a non-compliant stream that decodes correctly today: for example, 9 values written as a bit-packed run of 3 groups currently return the right 9 values, because the extra values are never read. I'd like to keep this PR to inputs that can't be read correctly today, so I left this cap for a follow-up, together with the run value and header length checks.

Nit (pre-existing): if the page ends inside this 4-byte length, e.g. a V1 page of an optional column whose data is only 2 bytes, readIntLittleEndian() fails with a bare EOFException whose message is null. readPageV1 still adds the page and column, but unlike the other truncation shapes there is no "Corrupted RLE data" text, and no test covers it. The length of RLE boolean values goes through the same path. Checking the remaining bytes first would make it consistent:

if (in.available() < 4) {
  throw new ParquetDecodingException("Corrupted RLE data: the 4-byte length is cut off, only " +
    in.available() + " bytes are left in the page");
}

Done, with a test.

Two more pre-existing gaps around this decoding, as possible follow-ups next to the run-value check from my review:

• The varint header has no length limit. A run length must fit in 31 bits, so a valid header is at most 5 bytes with a 5th byte of at most 0x0F, but readUnsignedVarInt() keeps going and Java masks the shift: [80 80 80 80 80 01] decodes to 8, i.e. an RLE run of 4, and the high bits of a 5th byte are dropped. Rejecting a 6th byte, or bits above 0x0F in the 5th, would turn this into "Corrupted RLE data".
• Dictionary ids are not checked against the dictionary size. For a lazily decoded column, e.g. required int32 with a 3-entry dictionary and the id section [02 0A 03] (an RLE run of five id 3), readBatch returns normally, and the ArrayIndexOutOfBoundsException is thrown later from ParquetDictionary.decodeToInt when a downstream operator reads the vector. That is outside FileScanRDD.hasNext, so the error has no file path and spark.sql.files.ignoreCorruptFiles doesn't apply. An RLE run could be checked once per run, but bit-packed ids would need a check per value, so this probably belongs in a follow-up.

Thanks. The header length limit has the same issue as the run count cap: a 6-byte header with zero high bytes (e.g. 94 80 80 80 80 00) decodes correctly today because Java masks the shift amount, so I left it for the follow-up. The dictionary id bound needs a per-value check for bit-packed ids, so I'd keep it separate too.

This catch now also wraps the init-time ParquetDecodingException of the other value readers, including the one that SPARK-59831 (f868de6, merged to master after this PR's base) added to VectorizedByteStreamSplitValuesReader.initFromPage. readPage()'s visitor then rethrows the IOException as a RuntimeException, so master's ParquetVectorizedSuite test "SPARK-59831: BYTE_STREAM_SPLIT page of a required column with a wrong value count" fails:

Expected exception org.apache.parquet.io.ParquetDecodingException to be thrown, but java.lang.RuntimeException was thrown

This is the only test failure in the CI run on 1a2868d, which merges master. The same test is on branch-4.x and branch-4.3, so the backports would fail the same way; branch-4.2 and older don't have the vectorized BYTE_STREAM_SPLIT reader.

Could you rebase, and either keep the exception type here (see the comment on L455) or update that test in this PR? The test also checks the message (3 encoded values do not match the page value count 4), so a wrapper would need to include e.getMessage().

Fixed in bf06bd8 with the type-preserving wrap from your L455 comment, and merged master in 517133c. The SPARK-59831 test passes now.

About where the page and column are added: turning the unchecked ParquetDecodingException into a checked IOException at three places (L134, L435 and here) leaves three shapes for the same kind of corruption:

• Errors thrown while reading, such as the shortened definition level run in the new ParquetIOSuite test, come from readBatch and reach FAILED_READ_FILE as a bare ParquetDecodingException without the page or column.
• Errors thrown while initializing a page become RuntimeException <- IOException("could not read page ... in col ...") <- ParquetDecodingException.
• Errors in the dictionary page become IOException("could not decode the dictionary for ...") <- ParquetDecodingException.

Keeping the type would avoid this, for example:

} catch (IOException e) {
  throw new IOException("could not read page " + page + " in col " + descriptor, e);
} catch (ParquetDecodingException e) {
  throw new ParquetDecodingException(
    "could not read page " + page + " in col " + descriptor + ": " + e.getMessage(), e);
}

That fixes the SPARK-59831 test from the L435 comment, keeps this PR's ParquetIOSuite assertions, and drops the extra RuntimeException layer. SPARK-59854 could then add the column to the errors thrown from readBatch with the same kind of catch, instead of catching RuntimeException, which would also wrap the SchemaColumnConvertNotSupportedException and SparkUpgradeException that FileDataSourceV2.attachFilePath dispatches on.

Done, at all three places including the dictionary page. Errors thrown while initializing a page or decoding the dictionary now come as a ParquetDecodingException with the page or dictionary, the column and the original message, without the RuntimeException layer. The errors thrown from readBatch are left for SPARK-59854, along the lines you describe.

Neither wrapper message includes the cause's message, and with the widened catch the init-time ParquetDecodingExceptions of the other readers (for example the length decoding of DELTA_LENGTH_BYTE_ARRAY and DELTA_BYTE_ARRAY pages, and the checks from SPARK-59812 and SPARK-59831) end up two levels deeper. Outside a subquery that's fine (4 or 5 throwables), but ThreadUtils.awaitResult in a scalar or IN subquery adds two more. For a DELTA_LENGTH_BYTE_ARRAY page with truncated length mini-blocks, the chain is now

SparkException("Exception thrown in awaitResult") <- ExecutionException <- FAILED_READ_FILE
  <- RuntimeException <- IOException("could not read page ... in col ...")
  <- ParquetDecodingException("Error reading mini block.") <- EOFException

and Spark Connect forwards only the first 5 (ErrorUtils.MAX_ERROR_CHAIN_LENGTH), so the client sees the page and column but not the reason. Before this PR it was SparkException <- ExecutionException <- FAILED_READ_FILE <- ParquetDecodingException <- EOFException, all delivered. Appending ": " + e.getMessage() here, or the type-preserving wrap from the L455 comment, would keep the reason visible.

The wrapper now includes the original message, so the reason is in the same exception that adds the column, and the chain is one layer shorter than before, since the RuntimeException from readPage() is gone too.

(Pre-existing) Related to reporting the column: pageReader.readDictionaryPage() (L129, just above this try) and pageReader.readPage() (L312) are outside every catch that adds it. In parquet-java 1.18.1, CodecFactory.HeapBytesDecompressor decompresses ZSTD and LZ4_RAW pages eagerly inside those calls, but SNAPPY and GZIP pages lazily, inside bytes.toInputStream() or initDictionary. So the same corrupted page reports could not read page ... in col [a] ... with SNAPPY or GZIP, but a bare ParquetDecodingException("could not decompress page") (or "Could not decompress dictionary page") without the column with ZSTD, and a bare MalformedInputException with LZ4_RAW. Fine to leave this for SPARK-59854; note that L129 needs its own try, because its result decides the if.

Agreed, I'll leave this for SPARK-59854, including the separate try that L129 needs.

Coming back to the bound from the earlier thread, with two new data points. This is pre-existing, not a regression.

• The description says these pages now fail "instead of ... running out of memory", but a bit-packed run whose bytes are all present is still not bounded by the page's value count. A dictionary id section of a 1-value page, [01][81 80 80 40][2^26 zero bytes] (2^26 groups; the zeros gzip to about 65 KB), passes both checks, allocates int[2^29] (2 GiB) and unpacks about 537M ids to return one. With a 1g heap that is an OutOfMemoryError, which ignoreCorruptFiles doesn't cover.
• branch-3.5 has 4e61adc ("Bounds-check RLE/bit-packed group counts in the vectorized Parquet reader", not released yet). It keeps pageValueCount and rejects runs longer than the page (count > pageValueCount + 7 for bit-packed runs), including the "9 values in 3 groups" stream. So the same file already behaves differently on branch-3.5 and master, and a 3.5 backport of this PR would conflict in this block.

Capping instead of rejecting would address the concern about streams that read correctly today: keep valueCount, decode only min(run length, values left) values (ceil(n / 8) groups for a bit-packed run), and throw in readNextGroup when no values are left. The "9 values in 3 groups" stream still returns its 9 values, the buffer is bounded by the page, and the bitWidth == 0 special case in readNextGroup becomes the same check. If you'd rather leave this for SPARK-59853, could the description be limited to headers that overflow or whose bytes are missing?

Thanks for the data points. I'd rather leave the bound for a follow-up: capping changes what we decode, and it should be decided together with the branch-3.5 behavior from 4e61adc, which rejects such runs. I narrowed the description to headers that overflow or whose bytes are missing, and added a note that a run whose bytes are all present isn't bounded yet.

The doc says the message must contain expected, but L79 checks "Corrupted RLE data: " + expected. The three initFromPage tests (L347, L396 and L407) also repeat the same intercept and prefixed contains by hand, so the prefix is written in four places. Since initFromPage has no loop, a thread-less assertCorrupted(expected)(f) that interceptCorrupted uses around future.get would keep the check in one place, and the doc could mention the prefix.

Added assertCorrupted(expected)(f), which interceptCorrupted uses around future.get. The three initFromPage tests use it, so the prefix is only there and in its doc.

Nit on the description: 8 of the 14 SPARK-59832 tests go through streams, not 7 ("data ending inside a run header or an RLE value" also uses it). Also, the three initFromPage negative tests (cut-off length, invalid length and invalid bit width) use a plain intercept on the test thread rather than "a separate thread with a timeout".

Fixed in the description: 8 of 14, and only the tests that read values run on a separate thread.

Nit: this block (a new reader, initFromPage(n, in()), a new vector, then interceptCorrupted(expected)(reader.readIntegers(n, c, 0)) for each of streams(page)) is repeated at L357-362 and L384-389 with only the page, the count and the message changed, and this test builds the same dictIdPage(Array.fill(10)(3), bitWidth = 4) three times. A class-level assertReadIntegersFails(page, n, expected) would cover the three blocks and turn check into one line.

Added assertReadIntegersFails(page, n, expected). The three blocks and check use it, and the test builds the page once.

Nit: BytesUtils.intToBytes(Int.MaxValue).copyToArray(bytes, offset) writes the same FF FF FF 7F, like VectorizedRleValuesReaderSuite does after the intLE comment in the last round.

Done.

Nit: SparkTestSuite.afterAll runs doThreadPostAudit() right after this, and shutdownNow() only interrupts the idle rle-reader-suite-N worker, so the worker can still be alive when the audit lists the threads. The CI run on 1a2868d didn't print the warning for this suite, so it's timing-dependent, but executor.awaitTermination(10, TimeUnit.SECONDS) after shutdownNow() would make it much less likely.

Added awaitTermination(10, SECONDS) after shutdownNow().