viirya commented on code in PR #59117:
URL: https://github.com/apache/spark/pull/59117#discussion_r4235887019
##########
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:
Thanks, confirmed: `PlainBinaryDictionary` reads the entry lengths with the
absolute `ByteBuffer.get(int)`, so a short BINARY dictionary fails with an
`IndexOutOfBoundsException`. I'll leave it for SPARK-59854 as you suggest,
together with the decompression errors.
##########
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:
Agreed. In 3398c18 the message is "the dictionary id section is empty, not
even the bit width", and the field is renamed to `emptyIdSection` with a
matching comment.
##########
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:
Done, the inner catch in `initDataReader` is gone, so an `IOException` from
a values reader's `initFromPage` also gets a single "could not read page ... in
col ..." from `readPageV1`/`readPageV2`.
##########
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:
Fixed in the description, 9 of 14.
##########
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:
Done, it now checks that the message includes the cause's message.
##########
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:
Done.
--
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]
