dongjoon-hyun commented on code in PR #59117:
URL: https://github.com/apache/spark/pull/59117#discussion_r4189443898
##########
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:
This catch now also wraps the init-time `ParquetDecodingException` of the
other value readers, including the one that SPARK-59831 (f868de6914d, 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](https://github.com/viirya/spark-1/runs/111388033615), 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()`.
##########
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:
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:
```java
} 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.
##########
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:
Neither wrapper message includes the cause's message, and with the widened
catch the init-time `ParquetDecodingException`s 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.
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java:
##########
@@ -130,7 +131,7 @@ public VectorizedColumnReader(
try {
Review Comment:
(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`.
##########
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:
About the release-note part of the description for the pages that could be
read correctly before and now fail here: it says that with
`spark.sql.files.ignoreCorruptFiles=true`, "the rows of these pages that were
read correctly before are now skipped with a warning instead of failing the
query". But `FileScanRDD` then skips the rest of the file (`finished = true`),
and `VectorizedParquetRecordReader.nextBatch` drops the batch it was decoding,
for all columns. For example, if row group 2 of a file with 10 row groups has a
dictionary-encoded page with an empty data section and a single-entry
dictionary, the whole file used to read correctly. With this PR and
`ignoreCorruptFiles=true`, the rows of the current batch and everything after
them, including row groups 3 to 10, are skipped with only a warning in the log.
Could it say that the rest of the file is skipped, as the paragraph above it
does? "instead of failing the query" also reads oddly, since these pages didn't
fail before.
##########
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:
Coming back to the bound from [the earlier
thread](https://github.com/apache/spark/pull/59117#discussion_r4147784267),
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 4e61adcec27 ("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?
##########
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:
No test goes through this widened catch: changing it back to `catch
(IOException e)` keeps every test green. `ParquetIOSuite`'s helper disables
dictionaries, and no test builds a corrupt dictionary page. A small case in
`ParquetVectorizedSuite` with the `MemPageStore` harness would pin it, for
example an `optional int32` column with
```scala
pageWriter.writeDictionaryPage(new DictionaryPage(
BytesInput.from(Array[Byte](1, 0, 0, 0, 2, 0, 0, 0)), 3,
Encoding.PLAIN_DICTIONARY))
```
plus one `RLE_DICTIONARY` data page. `recordReader.nextKeyValue()` then
fails with "could not decode the dictionary for [a] ...", caused by
`ParquetDecodingException: could not read int`. It needs a fixed-width type: a
short BINARY dictionary fails with an `IndexOutOfBoundsException`, which this
catch doesn't cover.
##########
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:
If a read times out here, nothing stops it. The decode loops don't check for
interruption, so neither `future.cancel(true)` nor `shutdownNow()` in
`afterAll` can end the thread, and each remaining hang-prone test waits another
30 seconds and leaves another thread spinning. With a regression to a silent
return in `readNextGroup` that keeps the new `if (currentCount == 0)
readNextGroup();` guards, 7 tests here hang: 7 x 30 seconds, plus 7 threads at
100% CPU for the rest of the sql/core test JVM (the default
`SparkParallelTestGrouping` group), which can slow down or flake the later
suites and hide the real failure. `ParquetIOSuite.readCorruptedRleSection`
likewise leaves the Spark task spinning after its 1 minute timeout. Recording
the first timeout in a flag and failing later `interceptCorrupted` calls
without submitting them, maybe with a shorter timeout since these reads take
microseconds, would bound it.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/VectorizedRleValuesReaderSuite.scala:
##########
@@ -185,10 +226,266 @@ class VectorizedRleValuesReaderSuite extends
SparkFunSuite {
produced += toRead
}
}
+
+ test("SPARK-59832: truncated dictionary ids fail instead of being partially
read") {
+ // A dictionary id page with 10 ids, read as if it had 20.
+ streams(dictIdPage(Array.fill(10)(3), bitWidth = 4)).foreach { in =>
Review Comment:
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.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetIOSuite.scala:
##########
@@ -2480,6 +2484,93 @@ class ParquetIOSuite extends ParquetTest with
SharedSparkSession {
}
}
}
+
+ /**
+ * Writes 100 rows of `column` to a single uncompressed data page, lets
`corrupt` modify the
+ * RLE section that starts with `03 00 00 00 C8 01 01` (the 4-byte length 3,
then one RLE run
+ * with header 100 << 1 and value 1), and returns the error of reading the
file with the
+ * vectorized reader. For a nullable INT32 column that is the definition
levels; for a
+ * BOOLEAN column written as a V2 page it is the RLE-encoded values.
+ */
+ private def readCorruptedRleSection(
+ column: String,
+ writerVersion: String = "PARQUET_1_0")(
+ corrupt: (Array[Byte], Int) => Unit): Throwable = {
+ var error: Throwable = null
+ withTempPath { dir =>
+ val path = dir.getCanonicalPath
+ spark.range(100).selectExpr(column).coalesce(1)
+ .write
+ .option(ParquetOutputFormat.ENABLE_DICTIONARY, "false")
+ .option(ParquetOutputFormat.WRITER_VERSION, writerVersion)
+ .option("compression", "uncompressed")
+ .parquet(path)
+ val file = dir.listFiles().filter(_.getName.endsWith(".parquet")).head
+ val bytes = Files.readAllBytes(file.toPath)
+ val section = Array[Byte](3, 0, 0, 0, 0xC8.toByte, 0x01, 1)
+ val offsets = bytes.indices.filter { i =>
+ bytes.slice(i, i + section.length).sameElements(section)
+ }
+ assert(offsets.size == 1)
+ corrupt(bytes, offsets.head)
+ Files.write(file.toPath, bytes)
+ dir.listFiles().filter(_.getName.endsWith(".crc")).foreach(_.delete())
+
+ withSQLConf(SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key -> "true") {
+ // Before SPARK-59832 a truncated read never returned, so run it with
a timeout.
+ // scalastyle:off awaitresult
+ error = intercept[SparkException] {
+ Await.result(
+
Future(spark.read.parquet(path).collect())(ExecutionContext.global), 1.minute)
+ }
+ // scalastyle:on awaitresult
+ }
+ }
+ error
+ }
+
+ private def causes(e: Throwable): Seq[Throwable] =
+ Iterator.iterate(e)(_.getCause).takeWhile(_ != null).toSeq
+
+ /** Sets the 4-byte length at `offset` to Int.MaxValue, far more than the
rest of the page. */
+ private def setMaxLength(bytes: Array[Byte], offset: Int): Unit =
+ Array(0xFF, 0xFF, 0xFF, 0x7F).zipWithIndex.foreach { case (b, i) =>
+ bytes(offset + i) = b.toByte
+ }
+
+ /**
+ * Checks that the page error is reported once with the column, directly on
top of the
+ * `ParquetDecodingException` about the invalid length.
+ */
+ private def assertInvalidLengthReportsColumn(e: Throwable, column: String):
Unit = {
+ val pageErrors = causes(e).filter(_.getMessage.startsWith("could not read
page"))
Review Comment:
`getMessage` can be null for a cause, e.g. the `new EOFException()` from
parquet's `sliceBuffers`. Against the merge-base reader, or after a regression
that lets `sliceStream(Int.MaxValue)` run again, both length tests then fail at
this line with `NullPointerException: Cannot invoke "String.startsWith(String)"
because the return value of "java.lang.Throwable.getMessage()" is null`,
instead of the assertion that prints `e`.
`Option(_.getMessage).exists(_.startsWith("could not read page"))` would keep
the diagnostic.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetIOSuite.scala:
##########
@@ -2480,6 +2484,93 @@ class ParquetIOSuite extends ParquetTest with
SharedSparkSession {
}
}
}
+
+ /**
+ * Writes 100 rows of `column` to a single uncompressed data page, lets
`corrupt` modify the
+ * RLE section that starts with `03 00 00 00 C8 01 01` (the 4-byte length 3,
then one RLE run
+ * with header 100 << 1 and value 1), and returns the error of reading the
file with the
+ * vectorized reader. For a nullable INT32 column that is the definition
levels; for a
+ * BOOLEAN column written as a V2 page it is the RLE-encoded values.
+ */
+ private def readCorruptedRleSection(
+ column: String,
+ writerVersion: String = "PARQUET_1_0")(
+ corrupt: (Array[Byte], Int) => Unit): Throwable = {
+ var error: Throwable = null
+ withTempPath { dir =>
+ val path = dir.getCanonicalPath
+ spark.range(100).selectExpr(column).coalesce(1)
+ .write
+ .option(ParquetOutputFormat.ENABLE_DICTIONARY, "false")
+ .option(ParquetOutputFormat.WRITER_VERSION, writerVersion)
+ .option("compression", "uncompressed")
+ .parquet(path)
+ val file = dir.listFiles().filter(_.getName.endsWith(".parquet")).head
+ val bytes = Files.readAllBytes(file.toPath)
+ val section = Array[Byte](3, 0, 0, 0, 0xC8.toByte, 0x01, 1)
+ val offsets = bytes.indices.filter { i =>
+ bytes.slice(i, i + section.length).sameElements(section)
+ }
+ assert(offsets.size == 1)
+ corrupt(bytes, offsets.head)
+ Files.write(file.toPath, bytes)
+ dir.listFiles().filter(_.getName.endsWith(".crc")).foreach(_.delete())
+
+ withSQLConf(SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key -> "true") {
+ // Before SPARK-59832 a truncated read never returned, so run it with
a timeout.
+ // scalastyle:off awaitresult
+ error = intercept[SparkException] {
+ Await.result(
+
Future(spark.read.parquet(path).collect())(ExecutionContext.global), 1.minute)
+ }
+ // scalastyle:on awaitresult
+ }
+ }
+ error
+ }
+
+ private def causes(e: Throwable): Seq[Throwable] =
+ Iterator.iterate(e)(_.getCause).takeWhile(_ != null).toSeq
+
+ /** Sets the 4-byte length at `offset` to Int.MaxValue, far more than the
rest of the page. */
+ private def setMaxLength(bytes: Array[Byte], offset: Int): Unit =
Review Comment:
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.
##########
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
Review Comment:
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.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetIOSuite.scala:
##########
@@ -2480,6 +2484,93 @@ class ParquetIOSuite extends ParquetTest with
SharedSparkSession {
}
}
}
+
+ /**
+ * Writes 100 rows of `column` to a single uncompressed data page, lets
`corrupt` modify the
+ * RLE section that starts with `03 00 00 00 C8 01 01` (the 4-byte length 3,
then one RLE run
+ * with header 100 << 1 and value 1), and returns the error of reading the
file with the
+ * vectorized reader. For a nullable INT32 column that is the definition
levels; for a
+ * BOOLEAN column written as a V2 page it is the RLE-encoded values.
+ */
+ private def readCorruptedRleSection(
+ column: String,
+ writerVersion: String = "PARQUET_1_0")(
+ corrupt: (Array[Byte], Int) => Unit): Throwable = {
+ var error: Throwable = null
+ withTempPath { dir =>
+ val path = dir.getCanonicalPath
+ spark.range(100).selectExpr(column).coalesce(1)
+ .write
+ .option(ParquetOutputFormat.ENABLE_DICTIONARY, "false")
+ .option(ParquetOutputFormat.WRITER_VERSION, writerVersion)
+ .option("compression", "uncompressed")
+ .parquet(path)
+ val file = dir.listFiles().filter(_.getName.endsWith(".parquet")).head
+ val bytes = Files.readAllBytes(file.toPath)
+ val section = Array[Byte](3, 0, 0, 0, 0xC8.toByte, 0x01, 1)
+ val offsets = bytes.indices.filter { i =>
+ bytes.slice(i, i + section.length).sameElements(section)
+ }
+ assert(offsets.size == 1)
+ corrupt(bytes, offsets.head)
+ Files.write(file.toPath, bytes)
+ dir.listFiles().filter(_.getName.endsWith(".crc")).foreach(_.delete())
+
+ withSQLConf(SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key -> "true") {
+ // Before SPARK-59832 a truncated read never returned, so run it with
a timeout.
+ // scalastyle:off awaitresult
Review Comment:
This is now the only `scalastyle:off awaitresult` in sql/core, and it's the
pattern this PR replaced in `VectorizedRleValuesReaderSuite` after the last
round; the `scala-execution-context-global-N` worker is also left for the
thread audit. Since the test thread waits interruptibly in
`DAGScheduler.runJob`, ScalaTest's `failAfter` with a `ThreadSignaler`, as in
`core/src/test/scala/org/apache/spark/DistributedSuite.scala`, works here
without another thread:
```scala
implicit val signaler: Signaler = ThreadSignaler
error = intercept[SparkException] {
failAfter(1.minute)(spark.read.parquet(path).collect())
}
```
`TimeLimits` is already mixed in through `SparkTestSuite`, so this only
needs `org.scalatest.concurrent.{Signaler, ThreadSignaler}`, and the
`scala.concurrent.{Await, ExecutionContext, Future}` import can go.
##########
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()
Review Comment:
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.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/VectorizedRleValuesReaderSuite.scala:
##########
@@ -185,10 +226,266 @@ class VectorizedRleValuesReaderSuite extends
SparkFunSuite {
produced += toRead
}
}
+
+ test("SPARK-59832: truncated dictionary ids fail instead of being partially
read") {
+ // A dictionary id page with 10 ids, read as if it had 20.
+ streams(dictIdPage(Array.fill(10)(3), bitWidth = 4)).foreach { in =>
+ val reader = new VectorizedRleValuesReader()
+ reader.initFromPage(20, in())
+ val c = new OnHeapColumnVector(20, IntegerType)
+ interceptCorrupted(PastEnd)(reader.readIntegers(20, c, 0))
+ }
+ streams(dictIdPage(Array.fill(10)(3), bitWidth = 4)).foreach { in =>
+ val reader = new VectorizedRleValuesReader()
+ reader.initFromPage(20, in())
+ interceptCorrupted(PastEnd)(reader.skipIntegers(20))
+ }
+ streams(dictIdPage(Array.fill(10)(3), bitWidth = 4)).foreach { in =>
+ val reader = new VectorizedRleValuesReader()
+ reader.initFromPage(20, in())
+ (0 until 10).foreach(_ => assert(reader.readInteger() == 3))
+ interceptCorrupted(PastEnd)(reader.readInteger())
+ }
+ }
+
+ test("SPARK-59832: truncated booleans fail instead of looping forever") {
+ val encoded = encodeRle(Array.fill(10)(1), 1)
+ streams(BytesUtils.intToBytes(encoded.length) ++ encoded).foreach { in =>
+ val reader = new VectorizedRleValuesReader(1)
+ reader.initFromPage(20, in())
+ val c = new OnHeapColumnVector(20, BooleanType)
+ interceptCorrupted(PastEnd)(reader.readBooleans(20, c, 0))
+ }
+ }
+
+ test("SPARK-59832: truncated definition levels fail in readBatch") {
+ // The page declares 100 values but its definition levels only encode 64.
Before the fix
+ // readBatch returned without progress, so
VectorizedColumnReader.readBatch spun forever.
+ val n = 100
+ val encoded = encodeRle(Array.fill(64)(1), 1)
+ val plainBytes = plainIntBytes(n)(valueAt)
+ def run(withDefLevels: Boolean, rowIndexes: PrimitiveIterator.OfLong):
Unit = {
+ val reader = new VectorizedRleValuesReader(1, false)
+ reader.initFromPage(n,
ByteBufferInputStream.wrap(ByteBuffer.wrap(encoded)))
+ val valueReader = new VectorizedPlainValuesReader
+ valueReader.initFromPage(n,
ByteBufferInputStream.wrap(ByteBuffer.wrap(plainBytes)))
+ val state = ParquetTestAccess.newState(intColumnDescriptor(1), false,
rowIndexes)
+ ParquetTestAccess.resetForNewPage(state, n, 0L)
+ ParquetTestAccess.resetForNewBatch(state, n)
+ val values = new OnHeapColumnVector(n, IntegerType)
+ val defLevels = if (withDefLevels) new OnHeapColumnVector(n,
IntegerType) else null
+ interceptCorrupted(PastEnd) {
+ ParquetTestAccess.readBatch(reader, state, values, defLevels,
valueReader, integerUpdater)
+ }
+ }
+ run(withDefLevels = false, rowIndexes = null)
+ run(withDefLevels = true, rowIndexes = null)
+ // Skips the truncated levels to reach rows 80 to 90.
+ run(withDefLevels = false, rowIndexes = longIterator((80 to 90).toArray))
+ }
+
+ test("SPARK-59832: truncated repetition levels fail in readBatchRepeated") {
+ // Every row is a top-level row (repetition level 0), but only 64 of 100
are encoded.
+ readRepeated(repLevels = Array.fill(64)(0), defLevels =
Array.fill(100)(1), n = 100)
+ }
+
+ test("SPARK-59832: truncated definition levels fail in readBatchRepeated") {
+ // The repetition levels are complete, but only 64 of 100 definition
levels are encoded.
+ // Before the fix the missing definition levels were silently left as 0.
+ readRepeated(repLevels = Array.fill(100)(0), defLevels =
Array.fill(64)(1), n = 100)
+ // Skips the truncated levels to reach rows 80 to 90.
+ readRepeated(repLevels = Array.fill(100)(0), defLevels =
Array.fill(64)(1), n = 100,
+ rowIndexes = longIterator((80 to 90).toArray))
+ }
+
+ private def readRepeated(
+ repLevels: Array[Int],
+ defLevels: Array[Int],
+ n: Int,
+ rowIndexes: PrimitiveIterator.OfLong = null): Unit = {
+ val prim = Types.primitive(PrimitiveTypeName.INT32,
Repetition.REPEATED).named("col")
+ val descriptor = new ColumnDescriptor(Array("col"), prim, 1, 1)
+ val repReader = new VectorizedRleValuesReader(1, false)
+ repReader.initFromPage(
+ n, ByteBufferInputStream.wrap(ByteBuffer.wrap(encodeRle(repLevels, 1))))
+ val defReader = new VectorizedRleValuesReader(1, false)
+ defReader.initFromPage(
+ n, ByteBufferInputStream.wrap(ByteBuffer.wrap(encodeRle(defLevels, 1))))
+ val valueReader = new VectorizedPlainValuesReader
+ valueReader.initFromPage(
+ n,
ByteBufferInputStream.wrap(ByteBuffer.wrap(plainIntBytes(n)(valueAt))))
+ val state = ParquetTestAccess.newState(descriptor, false, rowIndexes)
+ ParquetTestAccess.resetForNewPage(state, n, 0L)
+ ParquetTestAccess.resetForNewBatch(state, n)
+ val repLevelsVec = new OnHeapColumnVector(n, IntegerType)
+ val defLevelsVec = new OnHeapColumnVector(n, IntegerType)
+ val values = new OnHeapColumnVector(n, IntegerType)
+ interceptCorrupted(PastEnd) {
+ ParquetTestAccess.readBatchRepeated(repReader, state, repLevelsVec,
defReader,
+ defLevelsVec, values, valueReader, integerUpdater)
+ }
+ }
+
+ test("SPARK-59832: reading past a page with bit width 0 fails") {
+ // With bit width 0 the page is one implicit run of `valueCount` zeros;
the bytes that
+ // follow belong to the next section of the page and must not be decoded
as levels.
+ val trailing = Array[Byte](3, 1, 2, 3)
+ val reader = new VectorizedRleValuesReader(0)
+ reader.initFromPage(5,
ByteBufferInputStream.wrap(ByteBuffer.wrap(trailing)))
+ val c = new OnHeapColumnVector(6, IntegerType)
+ c.putInts(0, 6, -1)
+ reader.readIntegers(5, c, 0)
+ assert((0 until 5).forall(c.getInt(_) == 0))
+ interceptCorrupted(PastEnd)(reader.readIntegers(1, c, 5))
+ }
+
+ test("SPARK-59832: invalid level length is rejected") {
+ val payload = Array[Byte](10, 20, 30, 40)
+ Seq(-1, -4, -5, payload.length + 1, Int.MaxValue).foreach { length =>
+ streams(BytesUtils.intToBytes(length) ++ payload).foreach { in =>
+ val reader = new VectorizedRleValuesReader(1)
+ val e = intercept[ParquetDecodingException](reader.initFromPage(8,
in()))
+ assert(e.getMessage.contains(s"Corrupted RLE data: invalid length
$length"),
+ e.getMessage)
+ }
+ }
+ }
+
+ test("SPARK-59832: invalid bit-packed run is rejected before allocating its
buffer") {
+ def check(bitWidth: Int, numGroups: Long, data: Array[Byte])(expected:
String): Unit = {
+ val page = Array(bitWidth.toByte) ++ varint((numGroups << 1) | 1) ++ data
+ streams(page).foreach { in =>
+ val reader = new VectorizedRleValuesReader()
+ reader.initFromPage(10, in())
+ val c = new OnHeapColumnVector(10, IntegerType)
+ interceptCorrupted(expected)(reader.readIntegers(10, c, 0))
+ }
+ }
+ // Truncated last group. This was already rejected before (by `in.slice`),
even when the
+ // bytes cover every value read. parquet-java and Arrow accept it, for
compatibility with
+ // writers that do not pad the last group.
+ check(bitWidth = 4, numGroups = 2, data = Array.fill[Byte](5)(0x11))(
+ "bit-packed run of 2 groups needs 8 bytes, but only 5 bytes are left")
+ // numGroups * 8 and numGroups * bitWidth overflow to 0.
+ check(bitWidth = 8, numGroups = 1L << 29, data =
encodeRle(Array.fill(10)(3), 8))(
+ s"bit-packed run of ${1 << 29} groups is too long")
+ // numGroups * 8 and numGroups * bitWidth overflow to negative values.
+ check(bitWidth = 2, numGroups = Int.MaxValue, data =
Array.fill[Byte](16)(0))(
+ s"bit-packed run of ${Int.MaxValue} groups is too long")
+ // Would allocate an int[2^30] buffer before noticing that only 3 bytes
are left.
+ check(bitWidth = 1, numGroups = 1L << 27, data = Array[Byte](1, 2, 3))(
+ s"bit-packed run of ${1 << 27} groups needs ${1 << 27} bytes, but only 3
bytes are left")
+ }
+
+ test("SPARK-59832: data ending inside a run header or an RLE value fails") {
+ // The continuation byte of the run header is missing, and an RLE run of
10 without its
+ // value.
+ Seq(Array[Byte](4, 0x80.toByte), Array[Byte](4, 0x14)).foreach { page =>
+ streams(page).foreach { in =>
+ val reader = new VectorizedRleValuesReader()
+ reader.initFromPage(10, in())
+ val c = new OnHeapColumnVector(10, IntegerType)
+ interceptCorrupted("failed to read from input
stream")(reader.readIntegers(10, c, 0))
+ }
+ }
+ }
+
+ test("SPARK-59832: cut-off level length is rejected") {
+ streams(Array[Byte](3, 0)).foreach { in =>
+ val reader = new VectorizedRleValuesReader(1)
+ val e = intercept[ParquetDecodingException](reader.initFromPage(8, in()))
+ assert(e.getMessage.contains(
+ "Corrupted RLE data: the 4-byte length is cut off, only 2 bytes are
left in the page"),
+ e.getMessage)
+ }
+ }
+
+ test("SPARK-59832: invalid dictionary id bit width is rejected") {
+ Seq(33, 255).foreach { bitWidth =>
+ streams(Array[Byte](bitWidth.toByte, 0x14, 0)).foreach { in =>
+ val reader = new VectorizedRleValuesReader()
+ val e = intercept[ParquetDecodingException](reader.initFromPage(10,
in()))
+ assert(e.getMessage.contains(s"Corrupted RLE data: invalid bit width
$bitWidth"),
+ e.getMessage)
+ }
+ }
+ }
+
+ test("SPARK-59832: dictionary ids of an empty page are not decoded as 0") {
+ val reader = new VectorizedRleValuesReader()
+ reader.initFromPage(5,
ByteBufferInputStream.wrap(ByteBuffer.wrap(Array.emptyByteArray)))
+ // An all-null page reads no ids.
+ val c = new OnHeapColumnVector(5, IntegerType)
+ reader.readIntegers(0, c, 0)
+ reader.skipIntegers(0)
+ interceptCorrupted(PastEnd)(reader.readIntegers(5, c, 0))
+ }
+
+ test("SPARK-59832: runs of length 0 are skipped") {
+ // [bit width 4][RLE run of 0 x 7][bit-packed run of 0 groups][RLE run of
5 x 3]
+ val page = Array[Byte](4, 0, 7, 1, 10, 3)
+ val r1 = new VectorizedRleValuesReader()
+ r1.initFromPage(5, ByteBufferInputStream.wrap(ByteBuffer.wrap(page)))
+ assert((0 until 5).map(_ => r1.readInteger()) == Seq.fill(5)(3))
+ val r2 = new VectorizedRleValuesReader()
+ r2.initFromPage(5, ByteBufferInputStream.wrap(ByteBuffer.wrap(page)))
+ val c = new OnHeapColumnVector(5, IntegerType)
+ r2.readIntegers(5, c, 0)
+ assert((0 until 5).map(c.getInt) == Seq.fill(5)(3))
+ }
+
+ test("SPARK-59832: reads that end exactly at the end of the encoded values
succeed") {
+ // A run of 20 identical ids (RLE) followed by 17 mixed ids (bit-packed,
padded to 24).
+ val ids = Array.tabulate(37)(i => if (i < 20) 5 else i % 7)
+ streams(dictIdPage(ids, bitWidth = 3)).foreach { in =>
+ val reader = new VectorizedRleValuesReader()
+ reader.initFromPage(ids.length, in())
+ val c = new OnHeapColumnVector(ids.length, IntegerType)
+ reader.readIntegers(ids.length, c, 0)
+ assert((0 until ids.length).map(c.getInt) == ids.toSeq)
+ }
+ streams(dictIdPage(ids, bitWidth = 3)).foreach { in =>
+ val reader = new VectorizedRleValuesReader()
+ reader.initFromPage(ids.length, in())
+ reader.skipIntegers(30)
+ assert((30 until ids.length).map(_ => reader.readInteger()) ==
ids.drop(30).toSeq)
+ }
+
+ val booleans = Array.tabulate(37)(i => if (i < 20) 1 else i % 2)
+ val encoded = encodeRle(booleans, 1)
+ // The level length covers exactly the rest of the page.
+ streams(BytesUtils.intToBytes(encoded.length) ++ encoded).foreach { in =>
+ val reader = new VectorizedRleValuesReader(1)
+ reader.initFromPage(booleans.length, in())
+ val c = new OnHeapColumnVector(booleans.length, BooleanType)
+ reader.readBooleans(booleans.length, c, 0)
+ assert((0 until booleans.length).map(c.getBoolean) == booleans.map(_ ==
1).toSeq)
+ }
+ }
}
private object VectorizedRleValuesReaderSuite {
+ /**
+ * The page as a single buffer and split into 1-byte buffers. The split
keeps any part of the
+ * page longer than 1 byte, including a slice of it, on a
`MultiBufferInputStream`.
+ */
+ private def streams(bytes: Array[Byte]): Seq[() => ByteBufferInputStream] = {
Review Comment:
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".
--
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]
