dongjoon-hyun commented on code in PR #59117:
URL: https://github.com/apache/spark/pull/59117#discussion_r4147262164
##########
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) {
Review Comment:
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.
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedRleValuesReader.java:
##########
@@ -1014,10 +1045,8 @@ private boolean readNextGroup() {
}
}
} catch (IOException e) {
- throw new ParquetDecodingException("Failed to read from input stream",
e);
+ throw new ParquetDecodingException("Corrupted RLE data: failed to read
from input stream", e);
Review Comment:
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.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/VectorizedRleValuesReaderSuite.scala:
##########
@@ -38,6 +45,17 @@ class VectorizedRleValuesReaderSuite extends SparkFunSuite {
import VectorizedRleValuesReaderSuite._
+ /** Runs `f` in another thread, so that a reader that never returns fails
the test. */
+ private def interceptCorrupted(f: => Any): Unit = {
+ // scalastyle:off awaitresult
+ // Await.result, unlike ThreadUtils.awaitResult, rethrows the exception
unwrapped.
+ val e = intercept[ParquetDecodingException] {
+ Await.result(Future(f)(ExecutionContext.global), 30.seconds)
+ }
+ // scalastyle:on awaitresult
+ assert(e.getMessage.contains("Corrupted RLE data"), e.getMessage)
Review Comment:
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.
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java:
##########
@@ -362,7 +363,7 @@ private void initDataReader(
try {
dataColumn.initFromPage(pageValueCount, in);
- } catch (IOException e) {
+ } catch (IOException | ParquetDecodingException e) {
Review Comment:
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.
##########
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();
+ if (length < 0 || length > in.available()) {
+ throw new ParquetDecodingException("Corrupted RLE data: invalid
length " + length +
+ ", " + in.available() + " bytes are left in the page");
+ }
this.in = in.sliceStream(length);
}
} else {
// Initialize for values
if (in.available() > 0) {
init(in.read());
Review Comment:
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
```java
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.
##########
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);
Review Comment:
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.
--
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]
