Skip to content

Commit 65f1ea4

Browse files
viiryadongjoon-hyun
viirya
authored and
dongjoon-hyun
committed
[SPARK-59832][4.2][SQL] Fail on truncated input in the vectorized RLE reader
### What changes were proposed in this pull request? Backport of #59117 to branch-4.2. This branch does not have SPARK-56895, so `readGroup` still slices a bit-packed run one group at a time. The new checks run before that loop, as on master. Tests that came into the conflicts as context but are not on this branch are left out: the SPARK-57828 test in `ParquetIOSuite`, the SPARK-59831 test in `ParquetVectorizedSuite`, and the multi-buffer test from SPARK-56895 in `VectorizedRleValuesReaderSuite`. `ParquetVectorizedSuite` now imports `BytesInput`, `Encoding` and `Statistics` for the new dictionary page test, and the master-only SPARK-59620 import and doc comment are left out as well. ### Why are the changes needed? Same as #59117. On a corrupt page, the vectorized RLE reader, which decodes definition/repetition levels, dictionary ids and RLE booleans, silently returned wrong values or looped forever when the encoded data ended early, and a corrupt bit-packed header could overflow `currentCount` or allocate a huge buffer. The reader now fails with a `ParquetDecodingException`. ### Does this PR introduce _any_ user-facing change? Only for corrupt or non-compliant Parquet files, as described in #59117: such pages now fail with a `ParquetDecodingException` instead of returning wrong values or hanging, and with `spark.sql.files.ignoreCorruptFiles=true` the rest of such a file is skipped. A few of these pages could be read before and gave the right result by chance; #59117 lists them. ### How was this patch tested? The 18 tests added in #59117 pass on this branch, as do the existing tests in `ParquetEncodingSuite`, `ParquetVectorizedSuite`, `ParquetColumnIndexSuite`, `ParquetIOSuite`, `ParquetV1QuerySuite`, `ParquetV2QuerySuite`, `ParquetDeltaByteArrayEncodingSuite`, `ParquetDeltaLengthByteArrayEncodingSuite` and `VectorizedPlainValuesReaderSuite`. ``` build/sbt 'sql/testOnly *VectorizedRleValuesReaderSuite *ParquetEncodingSuite *ParquetVectorizedSuite *ParquetIOSuite' ``` ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude Code (Claude Opus 5.5) Closes #59482 from viirya/SPARK-59832-branch-4.2. Authored-by: Liang-Chi Hsieh <[email protected]> Signed-off-by: Dongjoon Hyun <[email protected]>
1 parent 6d48a38 commit 65f1ea4

6 files changed

Lines changed: 534 additions & 29 deletions

File tree

‎sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java‎

Lines changed: 14 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -32,6 +32,7 @@
3232
import org.apache.parquet.column.page.*;
3333
import org.apache.parquet.column.values.RequiresPreviousReader;
3434
import org.apache.parquet.column.values.ValuesReader;
35+
import org.apache.parquet.io.ParquetDecodingException;
3536
import org.apache.parquet.schema.LogicalTypeAnnotation;
3637
import org.apache.parquet.schema.LogicalTypeAnnotation.DateLogicalTypeAnnotation;
3738
import org.apache.parquet.schema.LogicalTypeAnnotation.DecimalLogicalTypeAnnotation;
@@ -131,6 +132,9 @@ public VectorizedColumnReader(
131132
this.isCurrentPageDictionaryEncoded = true;
132133
} catch (IOException e) {
133134
throw new IOException("could not decode the dictionary for " + descriptor, e);
135+
} catch (ParquetDecodingException e) {
136+
throw new ParquetDecodingException(
137+
"could not decode the dictionary for " + descriptor + ": " + e.getMessage(), e);
134138
}
135139
} else {
136140
this.dictionary = null;
@@ -335,9 +339,7 @@ private void initDataReader(
335339
if (dataEncoding.usesDictionary()) {
336340
this.dataColumn = null;
337341
if (dictionary == null) {
338-
throw new IOException(
339-
"could not read page in col " + descriptor +
340-
" as the dictionary was missing for encoding " + dataEncoding);
342+
throw new IOException("the dictionary was missing for encoding " + dataEncoding);
341343
}
342344
@SuppressWarnings("deprecation")
343345
Encoding plainDict = Encoding.PLAIN_DICTIONARY; // var to allow warning suppression
@@ -352,11 +354,9 @@ private void initDataReader(
352354
this.isCurrentPageDictionaryEncoded = false;
353355
}
354356

355-
try {
356-
dataColumn.initFromPage(pageValueCount, in);
357-
} catch (IOException e) {
358-
throw new IOException("could not read page in col " + descriptor, e);
359-
}
357+
// readPageV1 and readPageV2 add the page and the column to an IOException or a
358+
// ParquetDecodingException thrown here.
359+
dataColumn.initFromPage(pageValueCount, in);
360360
// for PARQUET-246 (See VectorizedDeltaByteArrayReader.setPreviousValues)
361361
if (CorruptDeltaByteArrays.requiresSequentialReads(writerVersion, dataEncoding) &&
362362
previousReader instanceof RequiresPreviousReader) {
@@ -413,6 +413,9 @@ private int readPageV1(DataPageV1 page) throws IOException {
413413
return pageValueCount;
414414
} catch (IOException e) {
415415
throw new IOException("could not read page " + page + " in col " + descriptor, e);
416+
} catch (ParquetDecodingException e) {
417+
throw new ParquetDecodingException(
418+
"could not read page " + page + " in col " + descriptor + ": " + e.getMessage(), e);
416419
}
417420
}
418421

@@ -433,6 +436,9 @@ private int readPageV2(DataPageV2 page) throws IOException {
433436
return pageValueCount;
434437
} catch (IOException e) {
435438
throw new IOException("could not read page " + page + " in col " + descriptor, e);
439+
} catch (ParquetDecodingException e) {
440+
throw new ParquetDecodingException(
441+
"could not read page " + page + " in col " + descriptor + ": " + e.getMessage(), e);
436442
}
437443
}
438444
}

‎sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedRleValuesReader.java‎

Lines changed: 63 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -62,6 +62,9 @@ private enum MODE {
6262
private int bytesWidth;
6363
private BytePacker packer;
6464

65+
// True if the dictionary id section of the page is empty, not even holding the bit width.
66+
private boolean emptyIdSection;
67+
6568
// Current decoding mode and values
6669
private MODE mode;
6770
private int currentCount;
@@ -96,16 +99,35 @@ public VectorizedRleValuesReader(int bitWidth, boolean readLength) {
9699
@Override
97100
public void initFromPage(int valueCount, ByteBufferInputStream in) throws IOException {
98101
this.in = in;
102+
this.emptyIdSection = false;
99103
if (fixedWidth) {
100104
// Initialize for repetition and definition levels
101105
if (readLength) {
106+
if (in.available() < 4) {
107+
throw new ParquetDecodingException("Corrupted RLE data: the 4-byte length is cut off, " +
108+
"only " + in.available() + " bytes are left in the page");
109+
}
102110
int length = readIntLittleEndian();
111+
if (length < 0 || length > in.available()) {
112+
throw new ParquetDecodingException("Corrupted RLE data: invalid length " + length +
113+
", " + in.available() + " bytes are left in the page");
114+
}
103115
this.in = in.sliceStream(length);
104116
}
105117
} else {
106118
// Initialize for values
107119
if (in.available() > 0) {
108-
init(in.read());
120+
int width = in.read();
121+
if (width > 32) {
122+
throw new ParquetDecodingException("Corrupted RLE data: invalid bit width " + width);
123+
}
124+
init(width);
125+
} else {
126+
// No encoded values (e.g. an all-null page). Do not treat this as a bit width of 0,
127+
// which would decode every value as 0; any read fails in `readNextGroup` instead.
128+
this.emptyIdSection = true;
129+
this.currentCount = 0;
130+
return;
109131
}
110132
}
111133
if (bitWidth == 0) {
@@ -208,7 +230,7 @@ private void readBatchInternal(
208230
int leftInPage = state.valuesToReadInPage;
209231

210232
while (leftInBatch > 0 && leftInPage > 0) {
211-
if (currentCount == 0 && !readNextGroup()) break;
233+
if (currentCount == 0) readNextGroup();
212234
int n = Math.min(leftInBatch, Math.min(leftInPage, this.currentCount));
213235

214236
long rangeStart = state.currentRangeStart();
@@ -320,7 +342,7 @@ private void readBatchInternalWithDefLevels(
320342
int leftInPage = state.valuesToReadInPage;
321343

322344
while (leftInBatch > 0 && leftInPage > 0) {
323-
if (currentCount == 0 && !readNextGroup()) break;
345+
if (currentCount == 0) readNextGroup();
324346
int n = Math.min(leftInBatch, Math.min(leftInPage, this.currentCount));
325347

326348
long rangeStart = state.currentRangeStart();
@@ -433,7 +455,7 @@ public void readBatchRepeatedInternal(
433455
values, nulls, valuesReused, valueReader, updater);
434456

435457
while ((leftInBatch > 0 || !state.lastListCompleted) && leftInPage > 0) {
436-
if (currentCount == 0 && !readNextGroup()) break;
458+
if (currentCount == 0) readNextGroup();
437459

438460
// Values to read in the current RLE/PACKED block, must be <= what's left in the page
439461
int valuesLeftInBlock = Math.min(leftInPage, currentCount);
@@ -639,7 +661,7 @@ private void readValues(
639661
int n = total;
640662
int initialValueOffset = state.valueOffset;
641663
while (n > 0) {
642-
if (currentCount == 0 && !readNextGroup()) break;
664+
if (currentCount == 0) readNextGroup();
643665
int num = Math.min(n, this.currentCount);
644666
readValuesN(num, state, defLevels, values, nulls, valueReader, updater);
645667
state.levelOffset += num;
@@ -737,7 +759,7 @@ private void skipValues(
737759
VectorizedValuesReader valuesReader,
738760
ParquetVectorUpdater updater) {
739761
while (n > 0) {
740-
if (currentCount == 0 && !readNextGroup()) break;
762+
if (currentCount == 0) readNextGroup();
741763
int num = Math.min(n, this.currentCount);
742764
switch (mode) {
743765
case RLE -> {
@@ -770,7 +792,7 @@ private void skipValues(
770792
public void readIntegers(int total, WritableColumnVector c, int rowId) {
771793
int left = total;
772794
while (left > 0) {
773-
if (currentCount == 0 && !readNextGroup()) break;
795+
if (currentCount == 0) readNextGroup();
774796
int n = Math.min(left, this.currentCount);
775797
switch (mode) {
776798
case RLE -> c.putInts(rowId, n, currentValue);
@@ -963,7 +985,6 @@ private int readIntLittleEndian() throws IOException {
963985
*/
964986
private int readIntLittleEndianPaddedOnBitWidth() throws IOException {
965987
return switch (bytesWidth) {
966-
case 0 -> 0;
967988
case 1 -> in.read();
968989
case 2 -> {
969990
int ch2 = in.read();
@@ -982,14 +1003,30 @@ private int readIntLittleEndianPaddedOnBitWidth() throws IOException {
9821003
}
9831004

9841005
/**
985-
* Reads the next group. Returns false if no more group available.
1006+
* Reads the next non-empty group, so that `currentCount` is positive on return.
1007+
* <p>
1008+
* Callers never read more values than the page declares, so running out of encoded data here
1009+
* means the page is corrupted. With a bit width of 0, `initFromPage` already provides all the
1010+
* values of the page, so reaching here also means reading past the end.
9861011
*/
987-
private boolean readNextGroup() {
988-
if (in.available() <= 0) {
989-
currentCount = 0;
990-
return false;
1012+
private void readNextGroup() {
1013+
if (emptyIdSection) {
1014+
throw new ParquetDecodingException(
1015+
"Corrupted RLE data: the dictionary id section is empty, not even the bit width");
9911016
}
1017+
do {
1018+
if (bitWidth == 0 || in.available() <= 0) {
1019+
throw new ParquetDecodingException(
1020+
"Corrupted RLE data: reading past the end of the encoded values");
1021+
}
1022+
readGroup();
1023+
} while (this.currentCount == 0);
1024+
}
9921025

1026+
/**
1027+
* Reads the next group, which may be empty.
1028+
*/
1029+
private void readGroup() {
9931030
try {
9941031
int header = readUnsignedVarInt();
9951032
this.mode = (header & 1) == 0 ? MODE.RLE : MODE.PACKED;
@@ -1000,6 +1037,17 @@ private boolean readNextGroup() {
10001037
}
10011038
case PACKED -> {
10021039
int numGroups = header >>> 1;
1040+
// Validate before allocating the buffer, so a corrupted header can neither overflow
1041+
// `currentCount` nor make us allocate more than the remaining bytes can fill.
1042+
if (numGroups > Integer.MAX_VALUE / 8) {
1043+
throw new ParquetDecodingException(
1044+
"Corrupted RLE data: bit-packed run of " + numGroups + " groups is too long");
1045+
}
1046+
if ((long) numGroups * bitWidth > in.available()) {
1047+
throw new ParquetDecodingException("Corrupted RLE data: bit-packed run of " +
1048+
numGroups + " groups needs " + (long) numGroups * bitWidth + " bytes, but only " +
1049+
in.available() + " bytes are left");
1050+
}
10031051
this.currentCount = numGroups * 8;
10041052

10051053
if (this.currentBuffer.length < this.currentCount) {
@@ -1016,10 +1064,8 @@ private boolean readNextGroup() {
10161064
}
10171065
}
10181066
} catch (IOException e) {
1019-
throw new ParquetDecodingException("Failed to read from input stream", e);
1067+
throw new ParquetDecodingException("Corrupted RLE data: failed to read from input stream", e);
10201068
}
1021-
1022-
return true;
10231069
}
10241070

10251071
/**
@@ -1028,7 +1074,7 @@ private boolean readNextGroup() {
10281074
private void skipValues(int n) {
10291075
int left = n;
10301076
while (left > 0) {
1031-
if (this.currentCount == 0 && !readNextGroup()) break;
1077+
if (this.currentCount == 0) readNextGroup();
10321078
int num = Math.min(left, this.currentCount);
10331079
switch (mode) {
10341080
case RLE -> {}

‎sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetIOSuite.scala‎

Lines changed: 89 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@
1818
package org.apache.spark.sql.execution.datasources.parquet
1919

2020
import java.math.{BigDecimal => JBigDecimal}
21+
import java.nio.file.Files
2122
import java.time.{LocalDateTime, LocalTime}
2223
import java.util.Locale
2324

@@ -27,16 +28,21 @@ import scala.reflect.ClassTag
2728
import scala.reflect.runtime.universe.TypeTag
2829

2930
import com.google.common.primitives.UnsignedLong
31+
import org.apache.commons.lang3.exception.ExceptionUtils
3032
import org.apache.hadoop.fs.{FileSystem, Path}
3133
import org.apache.hadoop.mapreduce.{JobContext, TaskAttemptContext}
34+
import org.apache.parquet.bytes.BytesUtils
3235
import org.apache.parquet.column.{Encoding, ParquetProperties}
3336
import org.apache.parquet.column.ParquetProperties.WriterVersion.PARQUET_1_0
3437
import org.apache.parquet.example.data.Group
3538
import org.apache.parquet.example.data.simple.{SimpleGroup, SimpleGroupFactory}
3639
import org.apache.parquet.hadoop._
3740
import org.apache.parquet.hadoop.example.ExampleParquetWriter
41+
import org.apache.parquet.io.ParquetDecodingException
3842
import org.apache.parquet.io.api.Binary
3943
import org.apache.parquet.schema.{MessageType, MessageTypeParser}
44+
import org.scalatest.concurrent.{Signaler, ThreadSignaler}
45+
import org.scalatest.time.{Seconds, Span}
4046

4147
import org.apache.spark.{SPARK_VERSION_SHORT, SparkException, TestUtils}
4248
import org.apache.spark.rdd.RDD
@@ -1927,6 +1933,89 @@ class ParquetIOSuite extends ParquetTest with SharedSparkSession {
19271933
}
19281934
}
19291935
}
1936+
1937+
/**
1938+
* Writes 100 rows of `column` to a single uncompressed data page, lets `corrupt` modify the
1939+
* RLE section that starts with `03 00 00 00 C8 01 01` (the 4-byte length 3, then one RLE run
1940+
* with header 100 << 1 and value 1), and returns the error of reading the file with the
1941+
* vectorized reader. For a nullable INT32 column that is the definition levels; for a
1942+
* BOOLEAN column written as a V2 page it is the RLE-encoded values.
1943+
*/
1944+
private def readCorruptedRleSection(
1945+
column: String,
1946+
writerVersion: String = "PARQUET_1_0")(
1947+
corrupt: (Array[Byte], Int) => Unit): Throwable = {
1948+
var error: Throwable = null
1949+
withTempPath { dir =>
1950+
val path = dir.getCanonicalPath
1951+
spark.range(100).selectExpr(column).coalesce(1)
1952+
.write
1953+
.option(ParquetOutputFormat.ENABLE_DICTIONARY, "false")
1954+
.option(ParquetOutputFormat.WRITER_VERSION, writerVersion)
1955+
.option("compression", "uncompressed")
1956+
.parquet(path)
1957+
val file = dir.listFiles().filter(_.getName.endsWith(".parquet")).head
1958+
val bytes = Files.readAllBytes(file.toPath)
1959+
val section = Array[Byte](3, 0, 0, 0, 0xC8.toByte, 0x01, 1)
1960+
val offset = bytes.indexOfSlice(section)
1961+
assert(offset >= 0 && bytes.indexOfSlice(section, offset + 1) < 0)
1962+
corrupt(bytes, offset)
1963+
Files.write(file.toPath, bytes)
1964+
dir.listFiles().filter(_.getName.endsWith(".crc")).foreach(_.delete())
1965+
1966+
withSQLConf(SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key -> "true") {
1967+
// Before SPARK-59832 a truncated read never returned, so run it with a timeout. The test
1968+
// thread waits interruptibly for the job, so it can be signaled directly.
1969+
implicit val signaler: Signaler = ThreadSignaler
1970+
error = intercept[SparkException] {
1971+
failAfter(Span(60, Seconds))(spark.read.parquet(path).collect())
1972+
}
1973+
}
1974+
}
1975+
error
1976+
}
1977+
1978+
private def causes(e: Throwable): Seq[Throwable] =
1979+
ExceptionUtils.getThrowableList(e).asScala.toSeq
1980+
1981+
/** Sets the 4-byte length at `offset` to Int.MaxValue, far more than the rest of the page. */
1982+
private def setMaxLength(bytes: Array[Byte], offset: Int): Unit =
1983+
BytesUtils.intToBytes(Int.MaxValue).copyToArray(bytes, offset)
1984+
1985+
/**
1986+
* Checks that the page error is reported once with the column, directly on top of the
1987+
* `ParquetDecodingException` about the invalid length.
1988+
*/
1989+
private def assertInvalidLengthReportsColumn(e: Throwable, column: String): Unit = {
1990+
val pageErrors =
1991+
causes(e).filter(c => Option(c.getMessage).exists(_.startsWith("could not read page")))
1992+
assert(pageErrors.size == 1, e)
1993+
assert(pageErrors.head.getMessage.contains(s"in col [$column]"), e)
1994+
assert(pageErrors.head.getCause.isInstanceOf[ParquetDecodingException], e)
1995+
assert(pageErrors.head.getCause.getMessage.contains(
1996+
s"Corrupted RLE data: invalid length ${Int.MaxValue}"), e)
1997+
}
1998+
1999+
test("SPARK-59832: vectorized reader fails on truncated definition levels") {
2000+
// Shorten the run to 64 values (header 64 << 1 = varint 0x80 0x01).
2001+
val e = readCorruptedRleSection("if(id < 0, null, cast(id as int)) as a") {
2002+
(bytes, offset) => bytes(offset + 4) = 0x80.toByte
2003+
}
2004+
assert(causes(e).exists {
2005+
case c: ParquetDecodingException => c.getMessage.contains("Corrupted RLE data")
2006+
case _ => false
2007+
}, e)
2008+
}
2009+
2010+
test("SPARK-59832: invalid definition level length reports the column") {
2011+
val e = readCorruptedRleSection("if(id < 0, null, cast(id as int)) as a")(setMaxLength)
2012+
assertInvalidLengthReportsColumn(e, "a")
2013+
}
2014+
2015+
test("SPARK-59832: invalid length of RLE booleans in a V2 page reports the column") {
2016+
val e = readCorruptedRleSection("id >= 0 as b", writerVersion = "PARQUET_2_0")(setMaxLength)
2017+
assertInvalidLengthReportsColumn(e, "b")
2018+
}
19302019
}
19312020

19322021
class JobCommitFailureParquetOutputCommitter(outputPath: Path, context: TaskAttemptContext)

0 commit comments

Comments
 (0)