You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by do...@apache.org on 2026/10/20 13:58:07 UTC
(spark) branch branch-4.x updated: [SPARK-59854][SQL] Report the column in decoding errors of the vectorized Parquet reader
This is an automated email from the ASF dual-hosted git repository.
dongjoon-hyun pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 50f367f5f77e [SPARK-59854][SQL] Report the column in decoding errors of the vectorized Parquet reader
50f367f5f77e is described below
commit 50f367f5f77e4bdf8e5f35cc72a38a31ea47d35a
Author: Liang-Chi Hsieh <vi...@gmail.com>
AuthorDate: Tue Oct 20 06:57:04 2026 -0700
[SPARK-59854][SQL] Report the column in decoding errors of the vectorized Parquet reader
### What changes were proposed in this pull request?
This is a follow-up of SPARK-59832 (#59117). That change added the page and the column to errors from initializing a data page in `VectorizedColumnReader`. This PR adds the column to the decoding errors that still do not report it:
- **Decoding a batch.** `readBatch` wraps a `ParquetDecodingException` thrown while decoding values or levels in a `ParquetDecodingException` that names the page and the column: `could not decode page <page> in col <column>: <original message>`. This covers the `ParquetDecodingException`s of the RLE, PLAIN, DELTA_BINARY_PACKED, DELTA_LENGTH_BYTE_ARRAY, DELTA_BYTE_ARRAY and BYTE_STREAM_SPLIT readers. Only `ParquetDecodingException` is caught, so other exceptions keep their types: `Sche [...]
- **Reading a data page.** parquet-java decompresses ZSTD pages eagerly, inside `PageReader.readPage()`, which was outside every catch. Its `ParquetDecodingException` (`could not decompress page`) now gets the column. Only `ParquetDecodingException` is caught here, because `readPage()` also decrypts pages, and decryption errors are not decoding errors.
- **Reading the dictionary.** `readDictionaryPage()` and `Encoding.initDictionary` now share one `try`, so a dictionary page that cannot be read fails with `could not decode the dictionary for <column>` whether it is decompressed eagerly (ZSTD) or when `initDictionary` reads its bytes (SNAPPY, GZIP). Besides `IOException` and `ParquetDecodingException`, it wraps the three runtime exceptions a corrupted dictionary page or page header causes there: an `IndexOutOfBoundsException` when a [...]
- **DELTA_BINARY_PACKED.** `VectorizedDeltaBinaryPackedReader.initFromPage` checked the page value count and the mini block size with `checkArgument`, which throws an `IllegalArgumentException`. These checks now throw a `ParquetDecodingException`, like the block size checks next to them, and the mini block size is checked with integer arithmetic. A mini block bit width above 64, which failed with an `ArrayIndexOutOfBoundsException`, is now rejected with a `ParquetDecodingException` wh [...]
- **An empty column chunk.** The constructor's `totalValueCount == 0` error, which a corrupt footer can cause, now names the column as well.
Each wrapper keeps the original message and adds a single layer to the cause chain.
These cases are not covered:
- An out-of-range dictionary id fails with an `ArrayIndexOutOfBoundsException` from `Dictionary.decodeTo*`, either in `readBatch` or, with lazy dictionary decoding, after `readBatch` returns. Reporting the column for it needs a check of every id against the dictionary size, so it is left to SPARK-60317.
- A dictionary page whose declared size or entry lengths do not fit in the page can end in an `OutOfMemoryError` or, for an uncompressed chunk, return bytes of the next page. That is left to SPARK-60316.
- A corrupt LZ4_RAW or LZ4 page fails with a `RuntimeException` of the codec library, aircompressor's `MalformedInputException` or lz4-java's `LZ4Exception`, which parquet-java does not turn into an `IOException`.
- A negative uncompressed size in the header of a compressed data page fails with a `NegativeArraySizeException`, in `readPage()` for ZSTD and LZ4_RAW and in `readPageV1` or `readPageV2` for the other codecs.
- With a pushed filter, parquet-java's `DictionaryFilter` reads dictionary pages while it filters row groups, before any `VectorizedColumnReader` exists, so a dictionary page that cannot be read there fails without the column.
### Why are the changes needed?
Without the column, a decoding error from a corrupted file does not say which column is affected. For example, a truncated RLE run fails with only `Corrupted RLE data: reading past the end of the encoded values`. When `VectorizedColumnReader` reads a SNAPPY or GZIP data page, a decompression error already reports the column, because the page is decompressed lazily inside the existing catch. ZSTD pages did not report it.
### Does this PR introduce _any_ user-facing change?
Yes, in error messages only. The decoding errors above now name the column, and the page for errors while decoding a batch. The corrupt dictionaries above now fail with a `ParquetDecodingException` that names the column instead of an `IndexOutOfBoundsException`, `NegativeArraySizeException` or `UnsupportedOperationException`, which is kept as the cause. The DELTA_BINARY_PACKED checks now throw a `ParquetDecodingException` instead of the `IllegalArgumentException` of the value count an [...]
### How was this patch tested?
New tests:
- `ParquetVectorizedSuite`:
- A truncated RLE dictionary id section and truncated PLAIN values report the page and the column, with the original `ParquetDecodingException` as the cause.
- A truncated BINARY dictionary, a negative dictionary size and a dictionary page with a non-dictionary encoding report the column.
- A mini block bit width of 65 in a DELTA_BINARY_PACKED page reports the page and the column, like the truncated pages above, and an invalid DELTA_BINARY_PACKED header reports the column.
- A column chunk without values reports the column.
- `ParquetDeltaEncodingSuite`: mini blocks that are not a multiple of 8 values and a page with no values are rejected with a `ParquetDecodingException`.
- `ParquetIOSuite`: a ZSTD data page and a ZSTD dictionary page that cannot be decompressed report the column. The test shares a helper with the SPARK-59832 tests, including their timeout.
All new tests fail without this change. Existing suites also pass: `VectorizedRleValuesReaderSuite`, `ParquetEncodingSuite`, `ParquetVectorizedSuite`, `ParquetColumnIndexSuite`, `ParquetIOSuite`, `ParquetV1QuerySuite`, `ParquetV2QuerySuite`, `ParquetDeltaEncodingSuite`, `ParquetDeltaByteArrayEncodingSuite`, `ParquetDeltaLengthByteArrayEncodingSuite`, `VectorizedPlainValuesReaderSuite`, `ParquetVectorUpdaterSuite` and `ParquetCompressionCodecPrecedenceSuite`.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude Opus 5.5)
This pull request and its description were written by Isaac.
Closes #59542 from viirya/SPARK-59854.
Authored-by: Liang-Chi Hsieh <vi...@gmail.com>
Signed-off-by: Dongjoon Hyun <do...@apache.org>
(cherry picked from commit 8c9aa483c1130c2a451e20d34e7f3ad60d393a88)
Signed-off-by: Dongjoon Hyun <do...@apache.org>
---
.../parquet/VectorizedColumnReader.java | 185 ++++++++++++---------
.../parquet/VectorizedDeltaBinaryPackedReader.java | 30 ++--
.../apache/parquet/column/page/TestDataPage.java | 5 +
.../parquet/ParquetDeltaEncodingSuite.scala | 19 +++
.../datasources/parquet/ParquetIOSuite.scala | 70 +++++---
.../parquet/ParquetVectorizedSuite.scala | 138 ++++++++++++---
6 files changed, 319 insertions(+), 128 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 dfe27c292b7e..15017238c3c0 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
@@ -97,6 +97,12 @@ public class VectorizedColumnReader {
*/
private long pageFirstRowIndex;
+ /**
+ * 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 String currentPage;
+
private final PageReader pageReader;
private final ColumnDescriptor descriptor;
private final LogicalTypeAnnotation logicalTypeAnnotation;
@@ -126,23 +132,33 @@ public class VectorizedColumnReader {
int96RebaseMode,
int96RebaseTz);
- DictionaryPage dictionaryPage = pageReader.readDictionaryPage();
- 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);
+ Dictionary dict = null;
+ try {
+ // This decompresses the page eagerly for some codecs, e.g. ZSTD.
+ DictionaryPage dictionaryPage = pageReader.readDictionaryPage();
+ if (dictionaryPage != null) {
+ dict = dictionaryPage.getEncoding().initDictionary(descriptor, dictionaryPage);
}
- } else {
- this.dictionary = null;
- this.isCurrentPageDictionaryEncoded = false;
+ } 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) {
+ // 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");
+ throw new IOException("totalValueCount == 0 in col " + descriptor);
}
assert "LEGACY".equals(datetimeRebaseMode) || "EXCEPTION".equals(datetimeRebaseMode) ||
"CORRECTED".equals(datetimeRebaseMode);
@@ -252,79 +268,96 @@ public class VectorizedColumnReader {
}
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);
- }
+ // 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 {
- 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);
+ // 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.toString();
this.pageFirstRowIndex = page.getFirstRowIndex().orElse(0L);
return page.accept(new DataPage.Visitor<Integer>() {
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 c04fb49f1add..9e4f0339a2de 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.BytePackerForLong;
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);
@@ -100,14 +101,17 @@ public class VectorizedDeltaBinaryPackedReader extends VectorizedReaderBase {
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 + ")");
}
- double miniSize = (double) blockSizeInValues / miniBlockNumInABlock;
- JavaUtils.checkArgument(miniSize % 8 == 0,
- "miniBlockSize must be multiple of 8, but it's " + miniSize);
- this.miniBlockSizeInValues = (int) miniSize;
+ 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 = blockSizeInValues / miniBlockNumInABlock;
// True value count. May be less than valueCount because of nulls
this.totalValueCount = BytesUtils.readUnsignedVarInt(in);
if (totalValueCount < 0) {
@@ -472,9 +476,15 @@ public class VectorizedDeltaBinaryPackedReader extends VectorizedReaderBase {
* 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 da3081919589..792735b85e51 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 class TestDataPage extends DataPage {
public <T> T accept(Visitor<T> 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 bf2bf5b010fa..eb0574a9bfdc 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 7fc1bf186f6a..4f51c87a0f3f 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.{PageType, Util}
import org.apache.parquet.hadoop._
import org.apache.parquet.hadoop.example.ExampleParquetWriter
import org.apache.parquet.io.ParquetDecodingException
@@ -2488,31 +2490,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: 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(bytes)
Files.write(file.toPath, bytes)
dir.listFiles().filter(_.getName.endsWith(".crc")).foreach(_.delete())
@@ -2528,6 +2517,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
@@ -2569,6 +2581,26 @@ 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") {
+ 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 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()
+ (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)
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 8611970baa12..5749b602a635 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
@@ -54,6 +55,11 @@ class ParquetVectorizedSuite extends ParquetTest with SharedSparkSession {
private val NUM_VALUES: Int = VALUES.length
private val BATCH_SIZE_CONFIGS: Seq[Int] = Seq(1, 3, 5, 7, 10, 20, 40)
private val PAGE_SIZE_CONFIGS: Seq[Seq[Int]] = Seq(Seq(6, 6, 7, 7), Seq(4, 9, 4, 9))
+ // 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)
+ // 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).
+ private val oneValueWithDictionaryId0 = Array[Byte](2, 0, 0, 0, 2, 1, 2, 2, 0)
implicit def toStrings(ints: Seq[Int]): Seq[String] = ints.map(i => ('a' + i).toChar.toString)
@@ -526,29 +532,12 @@ 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(e.getCause.getMessage), e.getMessage)
- assert(e.getCause.isInstanceOf[ParquetDecodingException], e)
- } finally {
- recordReader.close()
- }
+ val e = readCorruptedPage("int32", Some(new DictionaryPage(
+ BytesInput.from(Array[Byte](1, 0, 0, 0, 2, 0, 0, 0)), 3, Encoding.PLAIN_DICTIONARY)),
+ oneValueWithDictionaryId0, 1, Encoding.RLE_DICTIONARY)
+ assert(e.getMessage.contains("could not decode the dictionary for [a]"), e.getMessage)
+ assert(e.getMessage.contains(e.getCause.getMessage), e.getMessage)
+ assert(e.getCause.isInstanceOf[ParquetDecodingException], e)
}
test("SPARK-60262: a column requested twice fails instead of leaving a vector unwritten") {
@@ -585,6 +574,78 @@ class ParquetVectorizedSuite extends ParquetTest with SharedSparkSession {
}
}
+ 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),
+ // 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)
+ 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, which cuts off
+ // the length of the second entry.
+ ("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])
+ ).foreach { case (columnType, dictionaryPage, errorClass) =>
+ val e = readCorruptedPage(columnType, Some(dictionaryPage), oneValueWithDictionaryId0, 1,
+ Encoding.RLE_DICTIONARY)
+ assert(e.getMessage.startsWith(
+ s"could not decode the dictionary for [a] optional $columnType 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("int32", 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 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)
+ }
+
+ 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)
truncateTypeTest("basic struct",
@@ -887,6 +948,37 @@ 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
+ * it with the vectorized reader.
+ */
+ private def readCorruptedPage(
+ columnType: String,
+ dictionaryPage: Option[DictionaryPage],
+ data: Array[Byte],
+ valueCount: Int,
+ valuesEncoding: Encoding): ParquetDecodingException = {
+ val parquetSchema =
+ MessageTypeParser.parseMessageType(s"message root { optional $columnType a; }")
+ 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()
+ }
+ }
+
private def checkAnswer(
totalRowCount: Int,
fileSchema: MessageType,
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org