viirya commented on code in PR #59117:
URL: https://github.com/apache/spark/pull/59117#discussion_r4232800945
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java:
##########
@@ -431,7 +432,7 @@ private int readPageV1(DataPageV1 page) throws IOException {
defColumn.initFromPage(pageValueCount, in);
initDataReader(pageValueCount, page.getValueEncoding(), in);
return pageValueCount;
- } catch (IOException e) {
+ } catch (IOException | ParquetDecodingException e) {
Review Comment:
Fixed in bf06bd8 with the type-preserving wrap from your L455 comment, and
merged master in 517133c. The SPARK-59831 test passes now.
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java:
##########
@@ -451,7 +452,7 @@ private int readPageV2(DataPageV2 page) throws IOException {
try {
initDataReader(pageValueCount, page.getDataEncoding(),
page.getData().toInputStream());
return pageValueCount;
- } catch (IOException e) {
+ } catch (IOException | ParquetDecodingException e) {
Review Comment:
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.
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java:
##########
@@ -431,7 +432,7 @@ private int readPageV1(DataPageV1 page) throws IOException {
defColumn.initFromPage(pageValueCount, in);
initDataReader(pageValueCount, page.getValueEncoding(), in);
return pageValueCount;
- } catch (IOException e) {
+ } catch (IOException | ParquetDecodingException e) {
throw new IOException("could not read page " + page + " in col " +
descriptor, e);
Review Comment:
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.
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java:
##########
@@ -130,7 +131,7 @@ public VectorizedColumnReader(
try {
Review Comment:
Agreed, I'll leave this for SPARK-59854, including the separate `try` that
L129 needs.
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedRleValuesReader.java:
##########
@@ -994,6 +1022,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:
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 4e61adcec27, 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.
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedRleValuesReader.java:
##########
@@ -976,14 +992,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:
You're right. The description now says that from the batch that reaches such
a page, the rest of the file, including later row groups, is skipped with only
a warning, and no longer says "instead of failing the query".
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java:
##########
@@ -130,7 +131,7 @@ public VectorizedColumnReader(
try {
this.dictionary =
dictionaryPage.getEncoding().initDictionary(descriptor, dictionaryPage);
this.isCurrentPageDictionaryEncoded = true;
- } catch (IOException e) {
+ } catch (IOException | ParquetDecodingException e) {
Review Comment:
Added "SPARK-59832: a corrupted dictionary page reports the column" to
`ParquetVectorizedSuite` with your dictionary page. It fails on the previous
commit (an `IOException` is thrown) and on master (no column in the message).
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/VectorizedRleValuesReaderSuite.scala:
##########
@@ -19,25 +19,66 @@ package org.apache.spark.sql.execution.datasources.parquet
import java.nio.ByteBuffer
import java.util.PrimitiveIterator
+import java.util.concurrent.{Callable, ExecutionException, TimeUnit}
import scala.jdk.CollectionConverters._
-import org.apache.parquet.bytes.ByteBufferInputStream
+import org.apache.parquet.bytes.{ByteBufferInputStream, BytesInput, BytesUtils}
+import org.apache.parquet.column.ColumnDescriptor
+import org.apache.parquet.io.ParquetDecodingException
+import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName
+import org.apache.parquet.schema.Type.Repetition
+import org.apache.parquet.schema.Types
import org.apache.spark.SparkFunSuite
import
org.apache.spark.sql.execution.datasources.parquet.VectorizedRleValuesReaderTestUtils._
import org.apache.spark.sql.execution.vectorized.{OnHeapColumnVector,
WritableColumnVector}
-import org.apache.spark.sql.types.IntegerType
+import org.apache.spark.sql.types.{BooleanType, IntegerType}
+import org.apache.spark.util.ThreadUtils
/**
* Focused correctness tests for `VectorizedRleValuesReader.readBatch`
PACKED-mode decoding,
* covering patterns the P0 optimization cares about: run boundaries, batch
boundaries, and
* nested def-level grouping. Uses the same reflection bridge as the benchmark.
+ *
+ * The SPARK-59832 tests check that corrupted or truncated input fails with a
+ * `ParquetDecodingException` instead of returning wrong values, looping
forever or allocating
+ * a huge buffer, and that reads ending exactly at the end of the encoded
values still work.
*/
class VectorizedRleValuesReaderSuite extends SparkFunSuite {
import VectorizedRleValuesReaderSuite._
+ // Runs the reads that may never return before SPARK-59832. A thread per
read, so that a read
+ // stuck in a loop does not block the following ones.
+ private lazy val executor =
ThreadUtils.newDaemonCachedThreadPool("rle-reader-suite")
+
+ override def afterAll(): Unit = {
+ try {
+ executor.shutdownNow()
+ } finally {
+ super.afterAll()
+ }
+ }
+
+ /**
+ * Runs `f` in another thread, so that a reader that never returns fails the
test, and checks
+ * that it fails with a `ParquetDecodingException` whose message contains
`expected`.
+ */
+ private def interceptCorrupted(expected: String)(f: => Any): Unit = {
+ val future = executor.submit(new Callable[Any] {
+ override def call(): Any = f
+ })
+ val e = intercept[ParquetDecodingException] {
+ try {
+ future.get(30, TimeUnit.SECONDS)
Review Comment:
Done. After the first timeout, `interceptCorrupted` fails without submitting
the read, and the timeout is 10 seconds now. In `ParquetIOSuite` only one read
could hang before this change, so I left it with the one-minute `failAfter`.
--
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]
