dongjoon-hyun commented on code in PR #59117:
URL: https://github.com/apache/spark/pull/59117#discussion_r4147786622
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedRleValuesReader.java:
##########
@@ -100,12 +100,21 @@ public void initFromPage(int valueCount,
ByteBufferInputStream in) throws IOExce
// Initialize for repetition and definition levels
if (readLength) {
int length = readIntLittleEndian();
Review Comment:
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:
```java
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");
}
```
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedRleValuesReader.java:
##########
@@ -976,14 +984,26 @@ private int readIntLittleEndianPaddedOnBitWidth() throws
IOException {
}
/**
- * Reads the next group. Returns false if no more group available.
+ * Reads the next non-empty group, so that `currentCount` is positive on
return.
+ * <p>
+ * Callers never read more values than the page declares, so running out of
encoded data here
+ * means the page is corrupted. With a bit width of 0, `initFromPage`
already provides all the
+ * values of the page, so reaching here also means reading past the end.
*/
- private boolean readNextGroup() {
- if (in.available() <= 0) {
- currentCount = 0;
- return false;
- }
+ private void readNextGroup() {
+ do {
+ if (bitWidth == 0 || in.available() <= 0) {
+ throw new ParquetDecodingException(
+ "Corrupted RLE data: reading past the end of the encoded values");
+ }
+ readGroup();
+ } while (this.currentCount == 0);
+ }
+ /**
+ * Reads the next group, which may be empty.
+ */
+ private void readGroup() {
try {
int header = readUnsignedVarInt();
Review Comment:
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
