dongjoon-hyun commented on code in PR #59117:
URL: https://github.com/apache/spark/pull/59117#discussion_r4235457380
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedRleValuesReader.java:
##########
@@ -976,14 +997,30 @@ 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 (emptyPage) {
+ throw new ParquetDecodingException(
+ "Corrupted RLE data: the page has no data, not even the bit width");
Review Comment:
Minor: when this fires, the page usually does have data. Its definition
levels say that there are non-null values, and only the dictionary id section
after them is empty. "the page has no data" then points at the wrong part when
someone inspects the file. Could the message, and the field comment at L65
("True if the page has no data at all"), say that the dictionary id section is
empty instead? For example, "Corrupted RLE data: the dictionary id section is
empty, not even the bit width".
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java:
##########
@@ -153,6 +154,9 @@ public VectorizedColumnReader(
this.isCurrentPageDictionaryEncoded = true;
} catch (IOException e) {
throw new IOException("could not decode the dictionary for " +
descriptor, e);
+ } catch (ParquetDecodingException e) {
Review Comment:
(Pre-existing) This adds the column to a `ParquetDecodingException`, but not
every corrupt dictionary page throws one. In parquet-java 1.18.1,
`PlainBinaryDictionary`'s constructor reads the length of each entry with
`BytesUtils.readIntLittleEndian(ByteBuffer, int)`, which uses the absolute
`ByteBuffer.get(int)`. So a BINARY (string) dictionary page that is cut short
fails with an `IndexOutOfBoundsException`, which passes through both catches
and reaches `FAILED_READ_FILE` without "could not decode the dictionary for
...".
The new `ParquetVectorizedSuite` test covers the INT32 case, where
`IntegerPlainValuesReader` happens to throw a `ParquetDecodingException`
("could not read int"). Not a blocker for this PR. It could go with
SPARK-59854, for example by also wrapping a `RuntimeException` from
`initDictionary` in a `ParquetDecodingException` that adds the column.
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java:
##########
@@ -455,6 +459,9 @@ private int readPageV1(DataPageV1 page) throws IOException {
return pageValueCount;
} catch (IOException e) {
throw new IOException("could not read page " + page + " in col " +
descriptor, e);
+ } catch (ParquetDecodingException e) {
Review Comment:
Minor: now that `readPageV1` and `readPageV2` both add the page and column
around the whole `initDataReader` call, for both exception types, the inner
`catch (IOException e)` around `dataColumn.initFromPage` in `initDataReader`
(L390-392) only adds a second layer. An `IOException` from a values reader's
`initFromPage` comes as "could not read page Page [...] in col X" on top of
"could not read page in col X", while a `ParquetDecodingException` from the
same call gets only the outer one.
`assertInvalidLengthReportsColumn` in `ParquetIOSuite` checks for exactly
one page error, which holds only on the `ParquetDecodingException` path.
Dropping the inner catch would give both exception types the same shape.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetVectorizedSuite.scala:
##########
@@ -524,6 +524,33 @@ class ParquetVectorizedSuite extends ParquetTest with
SharedSparkSession {
}
}
+ test("SPARK-59832: a corrupted dictionary page reports the column") {
+ // A PLAIN INT32 dictionary page that declares 3 entries but holds only 2.
+ val parquetSchema = MessageTypeParser.parseMessageType("message root {
optional int32 a; }")
+ val ty = parquetSchema.asGroupType().getType("a").asPrimitiveType()
+ val cd = new ColumnDescriptor(Array("a"), ty, 0, 1)
+ val memPageStore = new MemPageStore(1)
+ val pageWriter = memPageStore.getPageWriter(cd)
+ pageWriter.writeDictionaryPage(new DictionaryPage(
+ BytesInput.from(Array[Byte](1, 0, 0, 0, 2, 0, 0, 0)), 3,
Encoding.PLAIN_DICTIONARY))
+ // One non-null value: the definition levels (length 2, an RLE run of 1 x
1), then the
+ // dictionary ids (bit width 2, an RLE run of 1 x 0).
+ pageWriter.writePage(BytesInput.from(Array[Byte](2, 0, 0, 0, 2, 1, 2, 2,
0)), 1, 1,
+ Statistics.createStats(ty), Encoding.RLE, Encoding.RLE,
Encoding.RLE_DICTIONARY)
+ val recordReader = new VectorizedParquetRecordReader(
+ DateTimeUtils.getZoneId("EST"), "CORRECTED", "UTC", "CORRECTED", "UTC",
true, 1)
+ try {
+ recordReader.initialize(parquetSchema, parquetSchema,
+ TestParquetRowGroupReader(Seq(TestPageReadStore(memPageStore,
Seq(0L)))), 1)
+ val e = intercept[ParquetDecodingException](recordReader.nextKeyValue())
+ assert(e.getMessage.contains("could not decode the dictionary for [a]"),
e.getMessage)
+ assert(e.getMessage.contains("could not read int"), e.getMessage)
Review Comment:
Nit: "could not read int" is parquet-java's message (from
`IntegerPlainValuesReader`), so this line can break on a parquet upgrade that
rewords it, even though the column is still reported. Since the point is that
the wrapper includes the original message,
`assert(e.getMessage.contains(e.getCause.getMessage), e)` would check that
without the hard-coded text.
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedRleValuesReader.java:
##########
@@ -976,14 +997,30 @@ 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 (emptyPage) {
Review Comment:
Nit: `emptyPage` doesn't change inside the loop, and when it's set the first
iteration throws, so this check could move above the `do`. The behavior is the
same; it just makes clear that the flag only matters before the first group.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/VectorizedRleValuesReaderSuite.scala:
##########
@@ -19,26 +19,93 @@ package org.apache.spark.sql.execution.datasources.parquet
import java.nio.ByteBuffer
import java.util.PrimitiveIterator
+import java.util.concurrent.{Callable, ExecutionException, TimeoutException,
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.filter2.columnindex.RowRanges
+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.{SparkException, 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. The decode loops do
not check for
+ // interruption, so a stuck read keeps its thread busy until the JVM exits.
After the first
+ // timeout, later reads fail without running, so that a regression leaves
one busy thread.
+ private lazy val executor =
ThreadUtils.newDaemonCachedThreadPool("rle-reader-suite")
+ @volatile private var timedOut = false
+
+ override def afterAll(): Unit = {
Review Comment:
Nit: `executor` is a `lazy val`, but `afterAll` always forces it, so a run
that skips every thread-based test (e.g. `-z PACKED`) creates the pool just to
shut it down. That's harmless, since a cached pool starts no thread before the
first `submit`, so a plain `val` would behave the same and be simpler.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/VectorizedRleValuesReaderSuite.scala:
##########
@@ -196,10 +263,252 @@ 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.
+ val page = dictIdPage(Array.fill(10)(3), bitWidth = 4)
+ assertReadIntegersFails(page, 20, PastEnd)
+ streams(page).foreach { in =>
+ val reader = new VectorizedRleValuesReader()
+ reader.initFromPage(20, in())
+ interceptCorrupted(PastEnd)(reader.skipIntegers(20))
+ }
+ streams(page).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)
+ assertCorrupted(s"invalid length $length")(reader.initFromPage(8,
in()))
+ }
+ }
+ }
+
+ 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 =
+ assertReadIntegersFails(
+ Array(bitWidth.toByte) ++ varint((numGroups << 1) | 1) ++ data, 10,
expected)
+ // 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 =>
+ assertReadIntegersFails(page, 10, "failed to read from input stream")
+ }
+ }
+
+ test("SPARK-59832: cut-off level length is rejected") {
+ streams(Array[Byte](3, 0)).foreach { in =>
+ val reader = new VectorizedRleValuesReader(1)
+ assertCorrupted("the 4-byte length is cut off, only 2 bytes are left in
the page") {
+ reader.initFromPage(8, in())
+ }
+ }
+ }
+
+ 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()
+ assertCorrupted(s"invalid bit width
$bitWidth")(reader.initFromPage(10, in()))
+ }
+ }
+ }
+
+ 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("the page has no data, not even the bit width") {
+ reader.readIntegers(5, c, 0)
+ }
+ }
+
+ test("SPARK-59832: runs of length 0 are skipped") {
Review Comment:
Nit on the description: now that this test also goes through
`streams(page)`, 9 of the 14 SPARK-59832 tests run on both buffer layouts, not
8.
--
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]
