viirya commented on code in PR #59117:
URL: https://github.com/apache/spark/pull/59117#discussion_r4176265390
##########
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:
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.
##########
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:
Added "data ending inside a run header or an RLE value fails" with both
inputs, on single and 1-byte buffers.
##########
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:
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.
##########
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:
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.
##########
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:
Done. It now throws "Corrupted RLE data: invalid bit width N", which
`readPageV1`/`readPageV2` wrap with the page and column.
##########
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:
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.
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedRleValuesReader.java:
##########
@@ -994,6 +1014,17 @@ private boolean readNextGroup() {
}
case PACKED -> {
int numGroups = header >>> 1;
+ // Validate before allocating the buffer, so a corrupted header can
neither overflow
+ // `currentCount` nor make us allocate more than the remaining bytes
can fill.
+ if (numGroups > Integer.MAX_VALUE / 8) {
+ throw new ParquetDecodingException(
+ "Corrupted RLE data: bit-packed run of " + numGroups + " groups
is too long");
+ }
+ if ((long) numGroups * bitWidth > in.available()) {
Review Comment:
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.
--
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]
