From 20ce4449db9295d39afd6752c8f1e6bd108b546b Mon Sep 17 00:00:00 2001 From: Liang-Chi Hsieh Date: Fri, 16 Oct 2026 00:13:12 -0700 Subject: [PATCH 1/4] [SPARK-59854][SQL] Report the column in decoding errors of the vectorized Parquet reader Add the column to decoding errors of the vectorized Parquet reader that did not report it: - ParquetDecodingExceptions thrown while decoding a batch in VectorizedColumnReader.readBatch. - ParquetDecodingExceptions thrown by PageReader.readPage and readDictionaryPage, which decompress ZSTD pages eagerly. - RuntimeExceptions thrown by Encoding.initDictionary, e.g. for a truncated BINARY dictionary. - Invalid DELTA_BINARY_PACKED headers, which now fail with a ParquetDecodingException instead of an IllegalArgumentException. --- .../parquet/VectorizedColumnReader.java | 162 +++++++++++------- .../VectorizedDeltaBinaryPackedReader.java | 13 +- .../datasources/parquet/ParquetIOSuite.scala | 43 +++++ .../parquet/ParquetVectorizedSuite.scala | 88 +++++++++- 4 files changed, 236 insertions(+), 70 deletions(-) diff --git a/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java b/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java index d7dca4caff725..12c17f5115123 100644 --- a/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java +++ b/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java @@ -98,6 +98,11 @@ public class VectorizedColumnReader { */ private long pageFirstRowIndex; + /** + * The data page the values are being decoded from, for error messages. + */ + private DataPage currentPage; + private final PageReader pageReader; private final ColumnDescriptor descriptor; private final LogicalTypeAnnotation logicalTypeAnnotation; @@ -147,7 +152,14 @@ public VectorizedColumnReader( int96RebaseMode, int96RebaseTz); - DictionaryPage dictionaryPage = pageReader.readDictionaryPage(); + DictionaryPage dictionaryPage; + try { + // This decompresses the page eagerly for some codecs, e.g. ZSTD. + dictionaryPage = pageReader.readDictionaryPage(); + } catch (ParquetDecodingException e) { + throw new ParquetDecodingException( + "could not read the dictionary page for " + descriptor + ": " + e.getMessage(), e); + } if (dictionaryPage != null) { try { this.dictionary = dictionaryPage.getEncoding().initDictionary(descriptor, dictionaryPage); @@ -157,6 +169,11 @@ public VectorizedColumnReader( } catch (ParquetDecodingException e) { throw new ParquetDecodingException( "could not decode the dictionary for " + descriptor + ": " + e.getMessage(), e); + } catch (RuntimeException e) { + // initDictionary only parses the bytes of the page, so a failure here comes from the page + // or its metadata, e.g. an IndexOutOfBoundsException for a truncated BINARY dictionary. + throw new ParquetDecodingException( + "could not decode the dictionary for " + descriptor + ": " + e, e); } } else { this.dictionary = null; @@ -264,79 +281,96 @@ void readBatch( } readState.resetForNewPage(pageValueCount, pageFirstRowIndex); } - PrimitiveType.PrimitiveTypeName typeName = - descriptor.getPrimitiveType().getPrimitiveTypeName(); - if (isCurrentPageDictionaryEncoded) { - // Save starting offset in case we need to decode dictionary IDs. - int startOffset = readState.valueOffset; - // Save starting row index so we can check if we need to eagerly decode dict ids later - long startRowId = readState.rowId; - - // Read and decode dictionary ids. - if (readState.maxRepetitionLevel == 0) { - defColumn.readIntegers(readState, dictionaryIds, column, definitionLevels, - (VectorizedValuesReader) dataColumn); - } else { - repColumn.readIntegersRepeated(readState, repetitionLevels, defColumn, definitionLevels, - dictionaryIds, column, (VectorizedValuesReader) dataColumn); - } - - // TIMESTAMP_MILLIS encoded as INT64 can't be lazily decoded as we need to post process - // the values to add microseconds precision. - if (column.hasDictionary() || (startRowId == pageFirstRowIndex && - isLazyDecodingSupported(typeName, column.dataType()))) { - // Column vector supports lazy decoding of dictionary values so just set the dictionary. - // We can't do this if startRowId is not the first row index in the page AND the column - // doesn't have a dictionary (i.e. some non-dictionary encoded values have already been - // added). - PrimitiveType primitiveType = descriptor.getPrimitiveType(); - - // We need to make sure that we initialize the right type for the dictionary otherwise - // WritableColumnVector will throw an exception when trying to decode to an Int when the - // dictionary is in fact initialized as Long - LogicalTypeAnnotation typeAnnotation = primitiveType.getLogicalTypeAnnotation(); - boolean castLongToInt = - typeAnnotation instanceof DecimalLogicalTypeAnnotation annotation && - annotation.getPrecision() <= Decimal.MAX_INT_DIGITS() && - primitiveType.getPrimitiveTypeName() == INT64; - - // We require a long value, but we need to use dictionary to decode the original - // signed int first - boolean isUnsignedInt32 = updaterFactory.isUnsignedIntTypeMatched(32); - - // We require a decimal value, but we need to use dictionary to decode the original - // signed long first - boolean isUnsignedInt64 = updaterFactory.isUnsignedIntTypeMatched(64); - - boolean needTransform = castLongToInt || isUnsignedInt32 || isUnsignedInt64; - column.setDictionary(new ParquetDictionary(dictionary, needTransform)); - } else { - updater.decodeDictionaryIds(readState.valueOffset - startOffset, startOffset, column, - dictionaryIds, dictionary); - } - } else { - if (column.hasDictionary() && readState.valueOffset != 0) { - // This batch already has dictionary encoded values but this new page is not. The batch - // does not support a mix of dictionary and not so we will decode the dictionary. - updater.decodeDictionaryIds(readState.valueOffset, 0, column, dictionaryIds, dictionary); - } - column.setDictionary(null); - VectorizedValuesReader valuesReader = (VectorizedValuesReader) dataColumn; - if (readState.maxRepetitionLevel == 0) { - defColumn.readBatch(readState, column, definitionLevels, valuesReader, updater); + // A corrupted page makes the readers below throw a ParquetDecodingException without the + // column. Only that is wrapped, so that other exceptions, e.g. SparkUpgradeException, + // keep their types. + try { + PrimitiveType.PrimitiveTypeName typeName = + descriptor.getPrimitiveType().getPrimitiveTypeName(); + if (isCurrentPageDictionaryEncoded) { + // Save starting offset in case we need to decode dictionary IDs. + int startOffset = readState.valueOffset; + // Save starting row index so we can check if we need to eagerly decode dict ids later + long startRowId = readState.rowId; + + // Read and decode dictionary ids. + if (readState.maxRepetitionLevel == 0) { + defColumn.readIntegers(readState, dictionaryIds, column, definitionLevels, + (VectorizedValuesReader) dataColumn); + } else { + repColumn.readIntegersRepeated(readState, repetitionLevels, defColumn, definitionLevels, + dictionaryIds, column, (VectorizedValuesReader) dataColumn); + } + + // TIMESTAMP_MILLIS encoded as INT64 can't be lazily decoded as we need to post process + // the values to add microseconds precision. + if (column.hasDictionary() || (startRowId == pageFirstRowIndex && + isLazyDecodingSupported(typeName, column.dataType()))) { + // Column vector supports lazy decoding of dictionary values so just set the dictionary. + // We can't do this if startRowId is not the first row index in the page AND the column + // doesn't have a dictionary (i.e. some non-dictionary encoded values have already been + // added). + PrimitiveType primitiveType = descriptor.getPrimitiveType(); + + // We need to make sure that we initialize the right type for the dictionary otherwise + // WritableColumnVector will throw an exception when trying to decode to an Int when the + // dictionary is in fact initialized as Long + LogicalTypeAnnotation typeAnnotation = primitiveType.getLogicalTypeAnnotation(); + boolean castLongToInt = + typeAnnotation instanceof DecimalLogicalTypeAnnotation annotation && + annotation.getPrecision() <= Decimal.MAX_INT_DIGITS() && + primitiveType.getPrimitiveTypeName() == INT64; + + // We require a long value, but we need to use dictionary to decode the original + // signed int first + boolean isUnsignedInt32 = updaterFactory.isUnsignedIntTypeMatched(32); + + // We require a decimal value, but we need to use dictionary to decode the original + // signed long first + boolean isUnsignedInt64 = updaterFactory.isUnsignedIntTypeMatched(64); + + boolean needTransform = castLongToInt || isUnsignedInt32 || isUnsignedInt64; + column.setDictionary(new ParquetDictionary(dictionary, needTransform)); + } else { + updater.decodeDictionaryIds(readState.valueOffset - startOffset, startOffset, column, + dictionaryIds, dictionary); + } } else { - repColumn.readBatchRepeated(readState, repetitionLevels, defColumn, definitionLevels, - column, valuesReader, updater); + if (column.hasDictionary() && readState.valueOffset != 0) { + // This batch already has dictionary encoded values but this new page is not. The batch + // does not support a mix of dictionary and not so we will decode the dictionary. + updater.decodeDictionaryIds(readState.valueOffset, 0, column, dictionaryIds, + dictionary); + } + column.setDictionary(null); + VectorizedValuesReader valuesReader = (VectorizedValuesReader) dataColumn; + if (readState.maxRepetitionLevel == 0) { + defColumn.readBatch(readState, column, definitionLevels, valuesReader, updater); + } else { + repColumn.readBatchRepeated(readState, repetitionLevels, defColumn, definitionLevels, + column, valuesReader, updater); + } } + } catch (ParquetDecodingException e) { + throw new ParquetDecodingException("could not decode page " + currentPage + " in col " + + descriptor + ": " + e.getMessage(), e); } } } private int readPage() { - DataPage page = pageReader.readPage(); + DataPage page; + try { + // This decompresses the page eagerly for some codecs, e.g. ZSTD. + page = pageReader.readPage(); + } catch (ParquetDecodingException e) { + throw new ParquetDecodingException( + "could not read the next page in col " + descriptor + ": " + e.getMessage(), e); + } if (page == null) { return -1; } + this.currentPage = page; this.pageFirstRowIndex = page.getFirstRowIndex().orElse(0L); return page.accept(new DataPage.Visitor() { diff --git a/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedDeltaBinaryPackedReader.java b/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedDeltaBinaryPackedReader.java index c04fb49f1addf..5133b3e99d941 100644 --- a/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedDeltaBinaryPackedReader.java +++ b/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedDeltaBinaryPackedReader.java @@ -26,7 +26,6 @@ import org.apache.parquet.column.values.bitpacking.Packer; import org.apache.parquet.io.ParquetDecodingException; -import org.apache.spark.network.util.JavaUtils; import org.apache.spark.sql.catalyst.util.RebaseDateTime; import org.apache.spark.sql.execution.datasources.DataSourceUtils; import org.apache.spark.sql.execution.vectorized.WritableColumnVector; @@ -89,8 +88,10 @@ public class VectorizedDeltaBinaryPackedReader extends VectorizedReaderBase { @Override public void initFromPage(int valueCount, ByteBufferInputStream in) throws IOException { - JavaUtils.checkArgument(valueCount >= 1, - "Page must have at least one value, but it has " + valueCount); + if (valueCount < 1) { + throw new ParquetDecodingException("Invalid DELTA_BINARY_PACKED page: it must have at " + + "least one value, but it has " + valueCount); + } this.in = in; // Read the header this.blockSizeInValues = BytesUtils.readUnsignedVarInt(in); @@ -105,8 +106,10 @@ public void initFromPage(int valueCount, ByteBufferInputStream in) throws IOExce + miniBlockNumInABlock + " (block size in values: " + blockSizeInValues + ")"); } double miniSize = (double) blockSizeInValues / miniBlockNumInABlock; - JavaUtils.checkArgument(miniSize % 8 == 0, - "miniBlockSize must be multiple of 8, but it's " + miniSize); + if (miniSize % 8 != 0) { + throw new ParquetDecodingException("Invalid DELTA_BINARY_PACKED mini block size: " + + miniSize + " (must be a multiple of 8)"); + } this.miniBlockSizeInValues = (int) miniSize; // True value count. May be less than valueCount because of nulls this.totalValueCount = BytesUtils.readUnsignedVarInt(in); diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetIOSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetIOSuite.scala index 7fc1bf186f6af..fd8281a55426e 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetIOSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetIOSuite.scala @@ -17,6 +17,7 @@ package org.apache.spark.sql.execution.datasources.parquet +import java.io.ByteArrayInputStream import java.math.{BigDecimal => JBigDecimal} import java.nio.file.Files import java.time.{LocalDateTime, LocalTime} @@ -36,6 +37,7 @@ import org.apache.parquet.column.{Encoding, ParquetProperties} import org.apache.parquet.column.ParquetProperties.WriterVersion.PARQUET_1_0 import org.apache.parquet.example.data.Group import org.apache.parquet.example.data.simple.{SimpleGroup, SimpleGroupFactory} +import org.apache.parquet.format.Util import org.apache.parquet.hadoop._ import org.apache.parquet.hadoop.example.ExampleParquetWriter import org.apache.parquet.io.ParquetDecodingException @@ -2569,6 +2571,47 @@ class ParquetIOSuite extends ParquetTest with SharedSparkSession { val e = readCorruptedRleSection("id >= 0 as b", writerVersion = "PARQUET_2_0")(setMaxLength) assertInvalidLengthReportsColumn(e, "b") } + + test("SPARK-59854: a ZSTD page that cannot be decompressed reports the column") { + // ZSTD pages are decompressed when they are read from the column chunk, not when decoded. + Seq( + (false, "could not read the next page in col", "could not decompress page"), + (true, "could not read the dictionary page for", "Could not decompress dictionary page") + ).foreach { case (dictionary, error, cause) => + withTempPath { dir => + val path = dir.getCanonicalPath + spark.range(1000).selectExpr("cast(id % 100 as int) as a").coalesce(1) + .write + .option(ParquetOutputFormat.ENABLE_DICTIONARY, dictionary.toString) + .option("compression", "zstd") + .parquet(path) + val file = dir.listFiles().filter(_.getName.endsWith(".parquet")).head + val footer = ParquetFileReader.readFooter( + spark.sessionState.newHadoopConf(), new Path(file.getCanonicalPath)) + val chunk = footer.getBlocks.get(0).getColumns.get(0) + assert(chunk.hasDictionaryPage == dictionary) + // Invert the first bytes of the first page, the ZSTD frame magic number. + val bytes = Files.readAllBytes(file.toPath) + val in = new ByteArrayInputStream(bytes, chunk.getStartingPos.toInt, bytes.length) + Util.readPageHeader(in) + val pageStart = bytes.length - in.available() + (pageStart until pageStart + 4).foreach(i => bytes(i) = (~bytes(i)).toByte) + Files.write(file.toPath, bytes) + dir.listFiles().filter(_.getName.endsWith(".crc")).foreach(_.delete()) + + withSQLConf(SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key -> "true") { + val e = intercept[SparkException](spark.read.parquet(path).collect()) + val columnErrors = + causes(e).filter(c => Option(c.getMessage).exists(_.contains("optional int32 a"))) + assert(columnErrors.size == 1, e) + assert(columnErrors.head.getMessage == + s"$error [a] optional int32 a: $cause", e) + assert(columnErrors.head.isInstanceOf[ParquetDecodingException], e) + assert(columnErrors.head.getCause.getMessage == cause, e) + } + } + } + } } class JobCommitFailureParquetOutputCommitter(outputPath: Path, context: TaskAttemptContext) diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetVectorizedSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetVectorizedSuite.scala index c1f45b54767ad..db4ca19f34496 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetVectorizedSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetVectorizedSuite.scala @@ -22,7 +22,7 @@ import java.util.{Optional, PrimitiveIterator} import scala.collection.mutable.ArrayBuffer import scala.language.implicitConversions -import org.apache.parquet.bytes.BytesInput +import org.apache.parquet.bytes.{ByteBufferInputStream, BytesInput} import org.apache.parquet.column.{ColumnDescriptor, Encoding, ParquetProperties} import org.apache.parquet.column.impl.ColumnWriteStoreV1 import org.apache.parquet.column.page._ @@ -551,6 +551,92 @@ class ParquetVectorizedSuite extends ParquetTest with SharedSparkSession { } } + /** + * Reads a row group of one page of `valueCount` values of the column `a` in `schema`, after + * an optional dictionary page, and returns the error of reading it. + */ + private def readCorruptedPage( + schema: String, + dictionaryPage: Option[DictionaryPage], + data: Array[Byte], + valueCount: Int, + valuesEncoding: Encoding): Throwable = { + val parquetSchema = MessageTypeParser.parseMessageType(schema) + val ty = parquetSchema.asGroupType().getType("a").asPrimitiveType() + val cd = new ColumnDescriptor(Array("a"), ty, 0, 1) + val memPageStore = new MemPageStore(valueCount) + val pageWriter = memPageStore.getPageWriter(cd) + dictionaryPage.foreach(pageWriter.writeDictionaryPage) + pageWriter.writePage(BytesInput.from(data), valueCount, valueCount, + Statistics.createStats(ty), Encoding.RLE, Encoding.RLE, valuesEncoding) + val recordReader = new VectorizedParquetRecordReader( + DateTimeUtils.getZoneId("EST"), "CORRECTED", "UTC", "CORRECTED", "UTC", true, valueCount) + try { + recordReader.initialize(parquetSchema, parquetSchema, + TestParquetRowGroupReader(Seq(TestPageReadStore(memPageStore, Seq(0L)))), valueCount) + intercept[ParquetDecodingException](recordReader.nextKeyValue()) + } finally { + recordReader.close() + } + } + + // The definition levels of 4 non-null values: the length 2, then an RLE run of 4 x 1. + private val fourNonNullValues = Array[Byte](2, 0, 0, 0, 4 << 1, 1) + + test("SPARK-59854: a decoding error in a batch reports the page and the column") { + Seq( + // Dictionary ids (bit width 1) with an RLE run of only 2 of the 4 values. + ("Corrupted RLE data: reading past the end of the encoded values", Some(new DictionaryPage( + BytesInput.from(Array[Byte](1, 0, 0, 0, 2, 0, 0, 0)), 2, Encoding.PLAIN_DICTIONARY)), + Array[Byte](1, 2 << 1, 0), Encoding.RLE_DICTIONARY), + // 2 of the 4 PLAIN values. + ("Failed to read 16 bytes", None, Array[Byte](1, 0, 0, 0, 2, 0, 0, 0), Encoding.PLAIN) + ).foreach { case (error, dictionaryPage, values, encoding) => + val e = readCorruptedPage("message root { optional int32 a; }", dictionaryPage, + fourNonNullValues ++ values, 4, encoding) + assert(e.getMessage.startsWith("could not decode page "), e.getMessage) + assert(e.getMessage.contains(s"in col [a] optional int32 a: $error"), e.getMessage) + assert(e.getCause.isInstanceOf[ParquetDecodingException], e) + assert(e.getCause.getMessage == error, e) + } + } + + test("SPARK-59854: a dictionary that cannot be decoded reports the column") { + Seq( + // A PLAIN BINARY dictionary page that declares 3 entries but holds only 1. + ("binary", new DictionaryPage(BytesInput.from(Array[Byte](1, 0, 0, 0, 'x')), 3, + Encoding.PLAIN_DICTIONARY), classOf[IndexOutOfBoundsException]), + // A dictionary page with an encoding that has no dictionary. + ("int32", new DictionaryPage(BytesInput.from(Array[Byte](1, 0, 0, 0)), 1, + Encoding.DELTA_BINARY_PACKED), classOf[UnsupportedOperationException]) + ).foreach { case (typeName, dictionaryPage, errorClass) => + // 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). + val e = readCorruptedPage(s"message root { optional $typeName a; }", Some(dictionaryPage), + Array[Byte](2, 0, 0, 0, 2, 1, 2, 2, 0), 1, Encoding.RLE_DICTIONARY) + assert(e.getMessage.startsWith( + s"could not decode the dictionary for [a] optional $typeName a: ${errorClass.getName}"), + e.getMessage) + assert(e.getCause.getClass == errorClass, e) + } + } + + test("SPARK-59854: an invalid DELTA_BINARY_PACKED header reports the column") { + // One non-null value. The header has a block of 128 values in 3 mini blocks. + val e = readCorruptedPage("message root { optional int32 a; }", None, + Array[Byte](2, 0, 0, 0, 2, 1, 0x80.toByte, 1, 3, 1, 0), 1, Encoding.DELTA_BINARY_PACKED) + assert(e.getMessage.contains("in col [a] optional int32 a: Invalid DELTA_BINARY_PACKED " + + "mini block size: 42.666666666666664 (must be a multiple of 8)"), e.getMessage) + assert(e.getCause.isInstanceOf[ParquetDecodingException], e) + + val reader = new VectorizedDeltaBinaryPackedReader() + val e2 = intercept[ParquetDecodingException] { + reader.initFromPage(0, ByteBufferInputStream.wrap(java.nio.ByteBuffer.allocate(0))) + } + assert(e2.getMessage == + "Invalid DELTA_BINARY_PACKED page: it must have at least one value, but it has 0") + } + truncateTypeTest("primitive type", IntegerType, LongType, IntegerType) truncateTypeTest("basic struct", From 68a4b4ef0f136db4bfaa52b31dd80387237b7185 Mon Sep 17 00:00:00 2001 From: Liang-Chi Hsieh Date: Fri, 16 Oct 2026 09:46:13 -0700 Subject: [PATCH 2/4] Address review comments - Wrap only the IndexOutOfBoundsException, NegativeArraySizeException and UnsupportedOperationException of initDictionary, and fix its comment. - Clear currentPage at the end of the column chunk. - Check the DELTA_BINARY_PACKED mini block size with integer arithmetic. - Share the corrupted file helper of ParquetIOSuite, with its timeout, in the ZSTD test, and check only that the column is reported once. Co-authored-by: Isaac --- .../parquet/VectorizedColumnReader.java | 13 ++- .../VectorizedDeltaBinaryPackedReader.java | 11 ++- .../datasources/parquet/ParquetIOSuite.scala | 91 +++++++++---------- .../parquet/ParquetVectorizedSuite.scala | 6 +- 4 files changed, 63 insertions(+), 58 deletions(-) diff --git a/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java b/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java index 12c17f5115123..b0f53b6fa72af 100644 --- a/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java +++ b/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java @@ -169,9 +169,14 @@ public VectorizedColumnReader( } catch (ParquetDecodingException e) { throw new ParquetDecodingException( "could not decode the dictionary for " + descriptor + ": " + e.getMessage(), e); - } catch (RuntimeException e) { - // initDictionary only parses the bytes of the page, so a failure here comes from the page - // or its metadata, e.g. an IndexOutOfBoundsException for a truncated BINARY dictionary. + } catch (IndexOutOfBoundsException | NegativeArraySizeException | + UnsupportedOperationException e) { + // initDictionary reads the bytes of the page, which also decompresses them for codecs + // that decompress lazily, e.g. SNAPPY, and parses them. A corrupted page or page header + // makes it throw these: + // an IndexOutOfBoundsException for a truncated BINARY dictionary, a + // NegativeArraySizeException for a negative dictionary size, and an + // UnsupportedOperationException for an encoding that has no dictionary. throw new ParquetDecodingException( "could not decode the dictionary for " + descriptor + ": " + e, e); } @@ -367,10 +372,10 @@ private int readPage() { throw new ParquetDecodingException( "could not read the next page in col " + descriptor + ": " + e.getMessage(), e); } + this.currentPage = page; if (page == null) { return -1; } - this.currentPage = page; this.pageFirstRowIndex = page.getFirstRowIndex().orElse(0L); return page.accept(new DataPage.Visitor() { diff --git a/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedDeltaBinaryPackedReader.java b/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedDeltaBinaryPackedReader.java index 5133b3e99d941..021a93bf3b2e1 100644 --- a/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedDeltaBinaryPackedReader.java +++ b/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedDeltaBinaryPackedReader.java @@ -105,12 +105,13 @@ public void initFromPage(int valueCount, ByteBufferInputStream in) throws IOExce throw new ParquetDecodingException("Invalid DELTA_BINARY_PACKED mini block count: " + miniBlockNumInABlock + " (block size in values: " + blockSizeInValues + ")"); } - double miniSize = (double) blockSizeInValues / miniBlockNumInABlock; - if (miniSize % 8 != 0) { - throw new ParquetDecodingException("Invalid DELTA_BINARY_PACKED mini block size: " - + miniSize + " (must be a multiple of 8)"); + if (blockSizeInValues % miniBlockNumInABlock != 0 || + (blockSizeInValues / miniBlockNumInABlock) % 8 != 0) { + throw new ParquetDecodingException("Invalid DELTA_BINARY_PACKED mini block count: " + + miniBlockNumInABlock + " (block size in values: " + blockSizeInValues + + ", the mini block size must be a multiple of 8)"); } - this.miniBlockSizeInValues = (int) miniSize; + this.miniBlockSizeInValues = blockSizeInValues / miniBlockNumInABlock; // True value count. May be less than valueCount because of nulls this.totalValueCount = BytesUtils.readUnsignedVarInt(in); if (totalValueCount < 0) { diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetIOSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetIOSuite.scala index fd8281a55426e..5db56845745b8 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetIOSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetIOSuite.scala @@ -38,8 +38,10 @@ import org.apache.parquet.column.ParquetProperties.WriterVersion.PARQUET_1_0 import org.apache.parquet.example.data.Group import org.apache.parquet.example.data.simple.{SimpleGroup, SimpleGroupFactory} import org.apache.parquet.format.Util +import org.apache.parquet.format.converter.ParquetMetadataConverter import org.apache.parquet.hadoop._ import org.apache.parquet.hadoop.example.ExampleParquetWriter +import org.apache.parquet.hadoop.util.HadoopInputFile import org.apache.parquet.io.ParquetDecodingException import org.apache.parquet.io.api.Binary import org.apache.parquet.schema.{MessageType, MessageTypeParser} @@ -2490,31 +2492,18 @@ 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. + * Writes 100 rows of `column` with the writer `options`, lets `corrupt` modify the bytes of + * the written file, and returns the error of reading the file with the vectorized reader. */ - private def readCorruptedRleSection( - column: String, - writerVersion: String = "PARQUET_1_0")( - corrupt: (Array[Byte], Int) => Unit): Throwable = { + private def readCorruptedFile(column: String, options: Map[String, String])( + corrupt: (Path, Array[Byte]) => 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) + spark.range(100).selectExpr(column).coalesce(1).write.options(options).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 offset = bytes.indexOfSlice(section) - assert(offset >= 0 && bytes.indexOfSlice(section, offset + 1) < 0) - corrupt(bytes, offset) + corrupt(new Path(file.getCanonicalPath), bytes) Files.write(file.toPath, bytes) dir.listFiles().filter(_.getName.endsWith(".crc")).foreach(_.delete()) @@ -2530,6 +2519,29 @@ class ParquetIOSuite extends ParquetTest with SharedSparkSession { error } + /** + * 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 = { + val options = Map( + ParquetOutputFormat.ENABLE_DICTIONARY -> "false", + ParquetOutputFormat.WRITER_VERSION -> writerVersion, + "compression" -> "uncompressed") + readCorruptedFile(column, options) { (_, bytes) => + val section = Array[Byte](3, 0, 0, 0, 0xC8.toByte, 0x01, 1) + val offset = bytes.indexOfSlice(section) + assert(offset >= 0 && bytes.indexOfSlice(section, offset + 1) < 0) + corrupt(bytes, offset) + } + } + private def causes(e: Throwable): Seq[Throwable] = ExceptionUtils.getThrowableList(e).asScala.toSeq @@ -2573,43 +2585,26 @@ class ParquetIOSuite extends ParquetTest with SharedSparkSession { } test("SPARK-59854: a ZSTD page that cannot be decompressed reports the column") { - // ZSTD pages are decompressed when they are read from the column chunk, not when decoded. - Seq( - (false, "could not read the next page in col", "could not decompress page"), - (true, "could not read the dictionary page for", "Could not decompress dictionary page") - ).foreach { case (dictionary, error, cause) => - withTempPath { dir => - val path = dir.getCanonicalPath - spark.range(1000).selectExpr("cast(id % 100 as int) as a").coalesce(1) - .write - .option(ParquetOutputFormat.ENABLE_DICTIONARY, dictionary.toString) - .option("compression", "zstd") - .parquet(path) - val file = dir.listFiles().filter(_.getName.endsWith(".parquet")).head - val footer = ParquetFileReader.readFooter( - spark.sessionState.newHadoopConf(), new Path(file.getCanonicalPath)) + Seq(false, true).foreach { dictionary => + val options = Map( + ParquetOutputFormat.ENABLE_DICTIONARY -> dictionary.toString, + "compression" -> "zstd") + val e = readCorruptedFile("cast(id % 10 as int) as a", options) { (file, bytes) => + val footer = ParquetFooterReader.readFooter( + HadoopInputFile.fromPath(file, spark.sessionState.newHadoopConf()), + ParquetMetadataConverter.NO_FILTER) val chunk = footer.getBlocks.get(0).getColumns.get(0) assert(chunk.hasDictionaryPage == dictionary) // Invert the first bytes of the first page, the ZSTD frame magic number. - val bytes = Files.readAllBytes(file.toPath) val in = new ByteArrayInputStream(bytes, chunk.getStartingPos.toInt, bytes.length) Util.readPageHeader(in) val pageStart = bytes.length - in.available() (pageStart until pageStart + 4).foreach(i => bytes(i) = (~bytes(i)).toByte) - Files.write(file.toPath, bytes) - dir.listFiles().filter(_.getName.endsWith(".crc")).foreach(_.delete()) - - withSQLConf(SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key -> "true") { - val e = intercept[SparkException](spark.read.parquet(path).collect()) - val columnErrors = - causes(e).filter(c => Option(c.getMessage).exists(_.contains("optional int32 a"))) - assert(columnErrors.size == 1, e) - assert(columnErrors.head.getMessage == - s"$error [a] optional int32 a: $cause", e) - assert(columnErrors.head.isInstanceOf[ParquetDecodingException], e) - assert(columnErrors.head.getCause.getMessage == cause, e) - } } + val columnErrors = + causes(e).filter(c => Option(c.getMessage).exists(_.contains("[a] optional int32 a"))) + assert(columnErrors.size == 1, e) + assert(columnErrors.head.isInstanceOf[ParquetDecodingException], e) } } } diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetVectorizedSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetVectorizedSuite.scala index db4ca19f34496..c1c796b7347b7 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetVectorizedSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetVectorizedSuite.scala @@ -606,6 +606,9 @@ class ParquetVectorizedSuite extends ParquetTest with SharedSparkSession { // A PLAIN BINARY dictionary page that declares 3 entries but holds only 1. ("binary", new DictionaryPage(BytesInput.from(Array[Byte](1, 0, 0, 0, 'x')), 3, Encoding.PLAIN_DICTIONARY), classOf[IndexOutOfBoundsException]), + // A dictionary page with a negative dictionary size. + ("int32", new DictionaryPage(BytesInput.from(Array[Byte](1, 0, 0, 0)), -1, + Encoding.PLAIN_DICTIONARY), classOf[NegativeArraySizeException]), // A dictionary page with an encoding that has no dictionary. ("int32", new DictionaryPage(BytesInput.from(Array[Byte](1, 0, 0, 0)), 1, Encoding.DELTA_BINARY_PACKED), classOf[UnsupportedOperationException]) @@ -626,7 +629,8 @@ class ParquetVectorizedSuite extends ParquetTest with SharedSparkSession { val e = readCorruptedPage("message root { optional int32 a; }", None, Array[Byte](2, 0, 0, 0, 2, 1, 0x80.toByte, 1, 3, 1, 0), 1, Encoding.DELTA_BINARY_PACKED) assert(e.getMessage.contains("in col [a] optional int32 a: Invalid DELTA_BINARY_PACKED " + - "mini block size: 42.666666666666664 (must be a multiple of 8)"), e.getMessage) + "mini block count: 3 (block size in values: 128, the mini block size must be a multiple " + + "of 8)"), e.getMessage) assert(e.getCause.isInstanceOf[ParquetDecodingException], e) val reader = new VectorizedDeltaBinaryPackedReader() From 3568ed4cb9bbf4e89a1b06bc0d48be77703837bc Mon Sep 17 00:00:00 2001 From: Liang-Chi Hsieh Date: Sat, 17 Oct 2026 12:27:26 -0700 Subject: [PATCH 3/4] Address the second round of review comments - Keep only the description of the current page, so that the page and its decompressor stream are not kept alive. - Read the dictionary page in the same try as initDictionary, and narrow its comment. - Reject a DELTA_BINARY_PACKED mini block bit width above 64, and drop a redundant check. - Test the page in the batch error message, the mini block size checks and the bit width check; share the test helpers and simplify the ZSTD test. Co-authored-by: Isaac --- .../parquet/VectorizedColumnReader.java | 54 +++++++++---------- .../VectorizedDeltaBinaryPackedReader.java | 12 +++-- .../parquet/column/page/TestDataPage.java | 5 ++ .../parquet/ParquetDeltaEncodingSuite.scala | 19 +++++++ .../datasources/parquet/ParquetIOSuite.scala | 46 +++++++--------- 5 files changed, 77 insertions(+), 59 deletions(-) diff --git a/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java b/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java index 97c2ebbe42dac..04e98037846da 100644 --- a/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java +++ b/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java @@ -99,9 +99,10 @@ public class VectorizedColumnReader { private long pageFirstRowIndex; /** - * The data page the values are being decoded from, for error messages. + * The data page the values are being decoded from, for error messages. Only its description is + * kept: for codecs that decompress lazily, the page holds the decompressor stream. */ - private DataPage currentPage; + private String currentPage; private final PageReader pageReader; private final ColumnDescriptor descriptor; @@ -151,38 +152,31 @@ public VectorizedColumnReader( int96RebaseMode, int96RebaseTz); - DictionaryPage dictionaryPage; + Dictionary dict = null; try { // This decompresses the page eagerly for some codecs, e.g. ZSTD. - dictionaryPage = pageReader.readDictionaryPage(); + DictionaryPage dictionaryPage = pageReader.readDictionaryPage(); + if (dictionaryPage != null) { + dict = dictionaryPage.getEncoding().initDictionary(descriptor, dictionaryPage); + } + } catch (IOException e) { + throw new IOException("could not decode the dictionary for " + descriptor, e); } catch (ParquetDecodingException e) { throw new ParquetDecodingException( - "could not read the dictionary page for " + descriptor + ": " + e.getMessage(), e); - } - if (dictionaryPage != null) { - try { - this.dictionary = dictionaryPage.getEncoding().initDictionary(descriptor, dictionaryPage); - this.isCurrentPageDictionaryEncoded = true; - } catch (IOException e) { - throw new IOException("could not decode the dictionary for " + descriptor, e); - } catch (ParquetDecodingException e) { - throw new ParquetDecodingException( - "could not decode the dictionary for " + descriptor + ": " + e.getMessage(), e); - } catch (IndexOutOfBoundsException | NegativeArraySizeException | - UnsupportedOperationException e) { - // initDictionary reads the bytes of the page, which also decompresses them for codecs - // that decompress lazily, e.g. SNAPPY, and parses them. A corrupted page or page header - // makes it throw these: - // an IndexOutOfBoundsException for a truncated BINARY dictionary, a - // NegativeArraySizeException for a negative dictionary size, and an - // UnsupportedOperationException for an encoding that has no dictionary. - throw new ParquetDecodingException( - "could not decode the dictionary for " + descriptor + ": " + e, e); - } - } else { - this.dictionary = null; - this.isCurrentPageDictionaryEncoded = false; + "could not decode the dictionary for " + descriptor + ": " + e.getMessage(), e); + } catch (IndexOutOfBoundsException | NegativeArraySizeException | + UnsupportedOperationException e) { + // Reading the dictionary page decompresses it, for some codecs only when initDictionary + // reads its bytes, and initDictionary parses it. A corrupted page or page header makes + // them throw these: an IndexOutOfBoundsException when a BINARY dictionary ends within the + // length of an entry, a NegativeArraySizeException for a negative size, and an + // UnsupportedOperationException for an encoding that has no dictionary. An entry that + // claims more bytes than the page has is not detected here (SPARK-60316). + throw new ParquetDecodingException( + "could not decode the dictionary for " + descriptor + ": " + e, e); } + this.dictionary = dict; + this.isCurrentPageDictionaryEncoded = dict != null; if (pageReader.getTotalValueCount() == 0) { throw new IOException("totalValueCount == 0"); } @@ -380,10 +374,10 @@ private int readPage() { throw new ParquetDecodingException( "could not read the next page in col " + descriptor + ": " + e.getMessage(), e); } - this.currentPage = page; if (page == null) { return -1; } + this.currentPage = page.toString(); this.pageFirstRowIndex = page.getFirstRowIndex().orElse(0L); return page.accept(new DataPage.Visitor() { diff --git a/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedDeltaBinaryPackedReader.java b/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedDeltaBinaryPackedReader.java index 021a93bf3b2e1..9e4f0339a2de9 100644 --- a/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedDeltaBinaryPackedReader.java +++ b/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedDeltaBinaryPackedReader.java @@ -101,7 +101,7 @@ public void initFromPage(int valueCount, ByteBufferInputStream in) throws IOExce throw new ParquetDecodingException("Invalid DELTA_BINARY_PACKED block size in values: " + blockSizeInValues + " (must be in [1, " + MAX_BLOCK_SIZE_IN_VALUES + "])"); } - if (miniBlockNumInABlock <= 0 || miniBlockNumInABlock > blockSizeInValues) { + if (miniBlockNumInABlock <= 0) { throw new ParquetDecodingException("Invalid DELTA_BINARY_PACKED mini block count: " + miniBlockNumInABlock + " (block size in values: " + blockSizeInValues + ")"); } @@ -476,9 +476,15 @@ private void readBlockHeader() { * see org.apache.parquet.column.values.delta.DeltaBinaryPackingValuesReader#unpackMiniBlock */ private void unpackMiniBlock() throws IOException { + int bitWidth = bitWidths[currentMiniBlock]; + // The bit widths of the unused mini blocks of the last block can be anything, so a width is + // checked only when its mini block is unpacked. + if (bitWidth > 64) { + throw new ParquetDecodingException("Invalid DELTA_BINARY_PACKED mini block bit width: " + + bitWidth + " (must be at most 64)"); + } Arrays.fill(this.unpackedValuesBuffer, 0); - BytePackerForLong packer = Packer.LITTLE_ENDIAN.newBytePackerForLong( - bitWidths[currentMiniBlock]); + BytePackerForLong packer = Packer.LITTLE_ENDIAN.newBytePackerForLong(bitWidth); for (int j = 0; j < miniBlockSizeInValues; j += 8) { ByteBuffer buffer = in.slice(packer.getBitWidth()); if (buffer.hasArray()) { diff --git a/sql/core/src/test/java/test/org/apache/parquet/column/page/TestDataPage.java b/sql/core/src/test/java/test/org/apache/parquet/column/page/TestDataPage.java index da30819195891..792735b85e519 100644 --- a/sql/core/src/test/java/test/org/apache/parquet/column/page/TestDataPage.java +++ b/sql/core/src/test/java/test/org/apache/parquet/column/page/TestDataPage.java @@ -41,4 +41,9 @@ public Optional getIndexRowCount() { public T accept(Visitor visitor) { return wrapped.accept(visitor); } + + @Override + public String toString() { + return wrapped.toString(); + } } diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetDeltaEncodingSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetDeltaEncodingSuite.scala index bf2bf5b010fad..eb0574a9bfdcf 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetDeltaEncodingSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetDeltaEncodingSuite.scala @@ -204,6 +204,25 @@ abstract class ParquetDeltaEncodingSuite[T] extends ParquetCompatibilityTest reader.initFromPage(100, craftHeader(128, 4, -1)) } assert(e3.getMessage.contains("total value count")) + + // The mini blocks must have a multiple of 8 values: 128 values do not split into 3 mini + // blocks, and split into 32 mini blocks of 4 values. + Seq(3, 32).foreach { miniBlockNum => + reader = new VectorizedDeltaBinaryPackedReader + val e4 = intercept[ParquetDecodingException] { + reader.initFromPage(100, craftHeader(128, miniBlockNum, 1)) + } + assert(e4.getMessage == s"Invalid DELTA_BINARY_PACKED mini block count: $miniBlockNum " + + "(block size in values: 128, the mini block size must be a multiple of 8)") + } + + // A page must have at least one value. + reader = new VectorizedDeltaBinaryPackedReader + val e5 = intercept[ParquetDecodingException] { + reader.initFromPage(0, craftHeader(128, 4, 1)) + } + assert(e5.getMessage == + "Invalid DELTA_BINARY_PACKED page: it must have at least one value, but it has 0") } test("skip()") { diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetIOSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetIOSuite.scala index 5db56845745b8..d18de3de5e4ca 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetIOSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetIOSuite.scala @@ -37,11 +37,9 @@ import org.apache.parquet.column.{Encoding, ParquetProperties} import org.apache.parquet.column.ParquetProperties.WriterVersion.PARQUET_1_0 import org.apache.parquet.example.data.Group import org.apache.parquet.example.data.simple.{SimpleGroup, SimpleGroupFactory} -import org.apache.parquet.format.Util -import org.apache.parquet.format.converter.ParquetMetadataConverter +import org.apache.parquet.format.{PageType, Util} import org.apache.parquet.hadoop._ import org.apache.parquet.hadoop.example.ExampleParquetWriter -import org.apache.parquet.hadoop.util.HadoopInputFile import org.apache.parquet.io.ParquetDecodingException import org.apache.parquet.io.api.Binary import org.apache.parquet.schema.{MessageType, MessageTypeParser} @@ -2496,14 +2494,14 @@ class ParquetIOSuite extends ParquetTest with SharedSparkSession { * the written file, and returns the error of reading the file with the vectorized reader. */ private def readCorruptedFile(column: String, options: Map[String, String])( - corrupt: (Path, Array[Byte]) => Unit): Throwable = { + corrupt: Array[Byte] => Unit): Throwable = { var error: Throwable = null withTempPath { dir => val path = dir.getCanonicalPath spark.range(100).selectExpr(column).coalesce(1).write.options(options).parquet(path) val file = dir.listFiles().filter(_.getName.endsWith(".parquet")).head val bytes = Files.readAllBytes(file.toPath) - corrupt(new Path(file.getCanonicalPath), bytes) + corrupt(bytes) Files.write(file.toPath, bytes) dir.listFiles().filter(_.getName.endsWith(".crc")).foreach(_.delete()) @@ -2534,7 +2532,7 @@ class ParquetIOSuite extends ParquetTest with SharedSparkSession { ParquetOutputFormat.ENABLE_DICTIONARY -> "false", ParquetOutputFormat.WRITER_VERSION -> writerVersion, "compression" -> "uncompressed") - readCorruptedFile(column, options) { (_, bytes) => + readCorruptedFile(column, options) { bytes => val section = Array[Byte](3, 0, 0, 0, 0xC8.toByte, 0x01, 1) val offset = bytes.indexOfSlice(section) assert(offset >= 0 && bytes.indexOfSlice(section, offset + 1) < 0) @@ -2585,28 +2583,24 @@ class ParquetIOSuite extends ParquetTest with SharedSparkSession { } test("SPARK-59854: a ZSTD page that cannot be decompressed reports the column") { - Seq(false, true).foreach { dictionary => - val options = Map( - ParquetOutputFormat.ENABLE_DICTIONARY -> dictionary.toString, - "compression" -> "zstd") - val e = readCorruptedFile("cast(id % 10 as int) as a", options) { (file, bytes) => - val footer = ParquetFooterReader.readFooter( - HadoopInputFile.fromPath(file, spark.sessionState.newHadoopConf()), - ParquetMetadataConverter.NO_FILTER) - val chunk = footer.getBlocks.get(0).getColumns.get(0) - assert(chunk.hasDictionaryPage == dictionary) - // Invert the first bytes of the first page, the ZSTD frame magic number. - val in = new ByteArrayInputStream(bytes, chunk.getStartingPos.toInt, bytes.length) - Util.readPageHeader(in) - val pageStart = bytes.length - in.available() - (pageStart until pageStart + 4).foreach(i => bytes(i) = (~bytes(i)).toByte) - } - val columnErrors = - causes(e).filter(c => Option(c.getMessage).exists(_.contains("[a] optional int32 a"))) - assert(columnErrors.size == 1, e) - assert(columnErrors.head.isInstanceOf[ParquetDecodingException], e) + Seq(PageType.DATA_PAGE -> false, PageType.DICTIONARY_PAGE -> true).foreach { + case (pageType, dictionary) => + val options = Map( + ParquetOutputFormat.ENABLE_DICTIONARY -> dictionary.toString, + "compression" -> "zstd") + val e = readCorruptedFile("cast(id % 10 as int) as a", options) { bytes => + // The first page of the only column chunk starts right after the magic number. + val in = new ByteArrayInputStream(bytes, ParquetFileWriter.MAGIC.length, bytes.length) + assert(Util.readPageHeader(in).getType == pageType) + // Invert the first bytes of the page, the ZSTD frame magic number. + val pageStart = bytes.length - in.available() + (pageStart until pageStart + 4).foreach(i => bytes(i) = (~bytes(i)).toByte) + } + assert(causes(e).exists(c => + Option(c.getMessage).exists(_.contains("[a] optional int32 a"))), e) } } + } class JobCommitFailureParquetOutputCommitter(outputPath: Path, context: TaskAttemptContext) From 20fe1cea204b5b20d708cc32acba25b5647cc67a Mon Sep 17 00:00:00 2001 From: Liang-Chi Hsieh Date: Mon, 19 Oct 2026 13:43:50 -0700 Subject: [PATCH 4/4] Address the third round of review comments - Name the column in the totalValueCount == 0 error of VectorizedColumnReader. - Move the DELTA_BINARY_PACKED bit width case into the batch test, and fix blank lines and a stream length in the tests. Co-authored-by: Isaac --- .../parquet/VectorizedColumnReader.java | 2 +- .../datasources/parquet/ParquetIOSuite.scala | 4 +- .../parquet/ParquetVectorizedSuite.scala | 39 ++++++++++++------- 3 files changed, 29 insertions(+), 16 deletions(-) diff --git a/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java b/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java index 04e98037846da..b3a261eefd11d 100644 --- a/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java +++ b/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java @@ -178,7 +178,7 @@ public VectorizedColumnReader( this.dictionary = dict; this.isCurrentPageDictionaryEncoded = dict != null; if (pageReader.getTotalValueCount() == 0) { - throw new IOException("totalValueCount == 0"); + throw new IOException("totalValueCount == 0 in col " + descriptor); } assert "LEGACY".equals(datetimeRebaseMode) || "EXCEPTION".equals(datetimeRebaseMode) || "CORRECTED".equals(datetimeRebaseMode); diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetIOSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetIOSuite.scala index d18de3de5e4ca..4f51c87a0f3f8 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetIOSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetIOSuite.scala @@ -2590,7 +2590,8 @@ class ParquetIOSuite extends ParquetTest with SharedSparkSession { "compression" -> "zstd") val e = readCorruptedFile("cast(id % 10 as int) as a", options) { bytes => // The first page of the only column chunk starts right after the magic number. - val in = new ByteArrayInputStream(bytes, ParquetFileWriter.MAGIC.length, bytes.length) + val start = ParquetFileWriter.MAGIC.length + val in = new ByteArrayInputStream(bytes, start, bytes.length - start) assert(Util.readPageHeader(in).getType == pageType) // Invert the first bytes of the page, the ZSTD frame magic number. val pageStart = bytes.length - in.available() @@ -2600,7 +2601,6 @@ class ParquetIOSuite extends ParquetTest with SharedSparkSession { Option(c.getMessage).exists(_.contains("[a] optional int32 a"))), e) } } - } class JobCommitFailureParquetOutputCommitter(outputPath: Path, context: TaskAttemptContext) diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetVectorizedSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetVectorizedSuite.scala index e9ff700fa6f7f..5749b602a635b 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetVectorizedSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetVectorizedSuite.scala @@ -17,6 +17,7 @@ package org.apache.spark.sql.execution.datasources.parquet +import java.io.IOException import java.util.{Optional, PrimitiveIterator} import scala.collection.mutable.ArrayBuffer @@ -580,7 +581,12 @@ class ParquetVectorizedSuite extends ParquetTest with SharedSparkSession { BytesInput.from(Array[Byte](1, 0, 0, 0, 2, 0, 0, 0)), 2, Encoding.PLAIN_DICTIONARY)), Array[Byte](1, 2 << 1, 0), Encoding.RLE_DICTIONARY), // 2 of the 4 PLAIN values. - ("Failed to read 16 bytes", None, Array[Byte](1, 0, 0, 0, 2, 0, 0, 0), Encoding.PLAIN) + ("Failed to read 16 bytes", None, Array[Byte](1, 0, 0, 0, 2, 0, 0, 0), Encoding.PLAIN), + // A DELTA_BINARY_PACKED block of 128 values in 4 mini blocks, with 4 values and the first + // value 0, then the min delta 0 and the mini block bit widths 65, 0, 0 and 0. The second + // value unpacks the first mini block. + ("Invalid DELTA_BINARY_PACKED mini block bit width: 65 (must be at most 64)", None, + Array[Byte](0x80.toByte, 1, 4, 4, 0, 0, 65, 0, 0, 0), Encoding.DELTA_BINARY_PACKED) ).foreach { case (error, dictionaryPage, values, encoding) => val e = readCorruptedPage("int32", dictionaryPage, fourNonNullValues ++ values, 4, encoding) assert(e.getMessage.startsWith("could not decode page Page [bytes.size="), e.getMessage) @@ -612,7 +618,7 @@ class ParquetVectorizedSuite extends ParquetTest with SharedSparkSession { } } - test("SPARK-59854: invalid DELTA_BINARY_PACKED data reports the column") { + test("SPARK-59854: an invalid DELTA_BINARY_PACKED header reports the column") { // One non-null value. The header has a block of 128 values in 3 mini blocks. val e = readCorruptedPage("int32", None, Array[Byte](2, 0, 0, 0, 2, 1, 0x80.toByte, 1, 3, 1, 0), 1, Encoding.DELTA_BINARY_PACKED) @@ -620,17 +626,24 @@ class ParquetVectorizedSuite extends ParquetTest with SharedSparkSession { "mini block count: 3 (block size in values: 128, the mini block size must be a multiple " + "of 8)"), e.getMessage) assert(e.getCause.isInstanceOf[ParquetDecodingException], e) + } - // Two non-null values, so the second one unpacks the first mini block. The header has a - // block of 128 values in 4 mini blocks, 2 values and the first value 0. The block has the - // min delta 0 and the mini block bit widths 65, 0, 0 and 0. - val e2 = readCorruptedPage("int32", None, - Array[Byte](2, 0, 0, 0, 2 << 1, 1, 0x80.toByte, 1, 4, 2, 0, 0, 65, 0, 0, 0), 2, - Encoding.DELTA_BINARY_PACKED) - assert(e2.getMessage.startsWith("could not decode page Page [bytes.size="), e2.getMessage) - assert(e2.getMessage.contains("in col [a] optional int32 a: Invalid DELTA_BINARY_PACKED " + - "mini block bit width: 65 (must be at most 64)"), e2.getMessage) - assert(e2.getCause.isInstanceOf[ParquetDecodingException], e2) + test("SPARK-59854: a column chunk without values reports the column") { + val parquetSchema = MessageTypeParser.parseMessageType("message root { optional int32 a; }") + val ty = parquetSchema.asGroupType().getType("a").asPrimitiveType() + val memPageStore = new MemPageStore(1) + // The column has a page writer, so the page store has a page reader for it, but no page. + memPageStore.getPageWriter(new ColumnDescriptor(Array("a"), ty, 0, 1)) + 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[IOException](recordReader.nextKeyValue()) + assert(e.getMessage == "totalValueCount == 0 in col [a] optional int32 a") + } finally { + recordReader.close() + } } truncateTypeTest("primitive type", IntegerType, LongType, IntegerType) @@ -935,7 +948,6 @@ class ParquetVectorizedSuite extends ParquetTest with SharedSparkSession { columnWriterStore.flush() } - /** * Reads a row group of the optional column `a` of type `columnType`, with an optional * dictionary page and one data page of `valueCount` values, and returns the error of reading @@ -966,6 +978,7 @@ class ParquetVectorizedSuite extends ParquetTest with SharedSparkSession { recordReader.close() } } + private def checkAnswer( totalRowCount: Int, fileSchema: MessageType,