You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "viirya (via GitHub)" <gi...@apache.org> on 2026/10/16 07:25:20 UTC
[PR] [SPARK-59854][SQL] Report the column in decoding errors of the vectorized Parquet reader [spark]
viirya opened a new pull request, #59542:
URL: https://github.com/apache/spark/pull/59542
### 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 RLE, PLAIN, DELTA_LENGTH_BYTE_ARRAY, DELTA_BYTE_ARRAY and BYTE_STREAM_SPLIT readers. Only `ParquetDecodingException` is caught, so other exceptions keep their types. `SchemaColumnConvertNotSupportedException` and `SparkUpgradeException`, which `FileDataSourceV2.attachFilePath` dispatches on, are not wrapped.
- **Reading a page.** parquet-java decompresses ZSTD pages eagerly, inside `PageReader.readPage()` and `readDictionaryPage()`, which were outside every catch. Their `ParquetDecodingException` (`could not decompress page`) now gets the column. Only `ParquetDecodingException` is caught here as well, because these calls also decrypt pages, and decryption errors are not decoding errors.
- **Decoding the dictionary.** `Encoding.initDictionary` only parses the bytes of the dictionary page, so a `RuntimeException` from it comes from the page or its metadata. It is now wrapped in a `ParquetDecodingException` with the column. Examples are an `IndexOutOfBoundsException` from a truncated BINARY dictionary, and an `UnsupportedOperationException` from a dictionary page whose encoding has no dictionary.
- **DELTA_BINARY_PACKED header.** `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, so the existing page wrapper adds the page and the column.
Each wrapper keeps the original message and adds a single layer to the cause chain.
### 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`. SNAPPY and GZIP pages already report the column when decompression fails, because they are 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. A corrupt DELTA_BINARY_PACKED header and an undecodable dictionary now fail with a `ParquetDecodingException` that names the column. Before, they failed with an `IllegalArgumentException`, `IndexOutOfBoundsException` or `UnsupportedOperationException`, and the original exception is kept as the cause. Valid files are not affected.
### 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 and a dictionary page with a non-dictionary encoding report the column.
- An invalid DELTA_BINARY_PACKED header reports the column, and a page with no values fails with a `ParquetDecodingException`.
- `ParquetIOSuite`: a ZSTD data page and a ZSTD dictionary page that cannot be decompressed report the column once.
All new tests fail without this change. Existing suites also pass: `VectorizedRleValuesReaderSuite`, `ParquetEncodingSuite`, `ParquetVectorizedSuite`, `ParquetColumnIndexSuite`, `ParquetIOSuite`, `ParquetV1QuerySuite`, `ParquetV2QuerySuite`, `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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59854][SQL] Report the column in decoding errors of the vectorized Parquet reader [spark]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on code in PR #59542:
URL: https://github.com/apache/spark/pull/59542#discussion_r4298490028
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java:
##########
@@ -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) {
Review Comment:
Right, an out-of-range id fails with an `ArrayIndexOutOfBoundsException` from `Dictionary.decodeTo*`, and with lazy decoding only after `readBatch` returns. Covering both needs a check of every id against `getMaxId()`, which is per-value work on the hot path (the lazy path does not look at the ids at all today), so I'd rather leave it to a follow-up. I narrowed the description: only `ParquetDecodingException`s are covered, and out-of-range ids are listed as not covered.
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java:
##########
@@ -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) {
Review Comment:
Good point, and the comment was wrong: for SNAPPY and GZIP the dictionary is decompressed when `initDictionary` reads the bytes. The catch now takes only what a corrupted page or page header causes there: `IndexOutOfBoundsException` (a truncated BINARY dictionary), `NegativeArraySizeException` (a negative dictionary size in the page header) and `UnsupportedOperationException` (an encoding that has no dictionary). The test covers all three, and the comment now says that decompression can happen here. Decompression errors there are `IOException`s, which the existing catch already handles.
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java:
##########
@@ -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;
Review Comment:
Done, `currentPage` is now set before the null check, so it is cleared at the end of the column chunk.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetIOSuite.scala:
##########
@@ -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 =>
Review Comment:
Done. `readCorruptedFile` now takes the writer options and the corruption step; `readCorruptedRleSection` and the ZSTD test both use it, with the 60-second `failAfter` guard.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetIOSuite.scala:
##########
@@ -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 ==
Review Comment:
Done, the test now only checks that the column is reported once, on a `ParquetDecodingException`.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetIOSuite.scala:
##########
@@ -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(
Review Comment:
`ParquetTest.readFooter` reads the `_metadata` summary file of a directory (`new Path(path, PARQUET_METADATA_FILE)`), which is why the tests at L1353 and L1527 write summary files first, so it would not find the footer of this data file. I switched to the non-deprecated `ParquetFooterReader.readFooter(HadoopInputFile.fromPath(...), NO_FILTER)` instead.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59854][SQL] Report the column in decoding errors of the vectorized Parquet reader [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on code in PR #59542:
URL: https://github.com/apache/spark/pull/59542#discussion_r4315099719
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java:
##########
@@ -146,21 +152,31 @@ public 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 |
Review Comment:
Minor: the third bullet of item 5 of my second review (on L373) is still open, and it applies to data pages too. `compression=lz4` maps to Hadoop's `Lz4Codec`, which parquet-java decompresses lazily (`CodecFactory` copies eagerly only for ZSTD and LZ4_RAW). So a page whose LZ4 block is corrupt is decompressed inside `initDictionary` at L160, or by `toInputStream()` at L491 and L518. Hadoop's `Lz4Decompressor` passes lz4-java's `LZ4Exception`, a `RuntimeException`, through, so it gets past the catches here and at L497-L502 and L520-L525, and the read fails without the column.
The description says that LZ4_RAW differs "because parquet-java does not turn it into an `IOException` as it does for the other codecs", which does not hold for LZ4. Could we list LZ4 next to LZ4_RAW in the "not covered" list and drop "as it does for the other codecs"?
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java:
##########
@@ -146,21 +152,31 @@ public 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");
Review Comment:
Nit: this is now the only error in the constructor that does not name the column. It is older than this PR, but a corrupt footer that sets `num_values = 0` for one column chunk of a non-empty row group reaches it: parquet-java then reads no page of that chunk, and the read fails with only `totalValueCount == 0`. Could we add the column here as well, e.g. `throw new IOException("totalValueCount == 0 in col " + descriptor);`?
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetVectorizedSuite.scala:
##########
@@ -887,6 +935,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()
+ }
+ }
Review Comment:
Nit: moving `readCorruptedPage` here left two blank lines above its Scaladoc (L937-L938) and none between its closing brace and `checkAnswer` (L968-L969). Could we move the blank line from L938 to after L968? Also, `ParquetIOSuite.scala:2603` adds a blank line before the closing brace of the class, which the base does not have.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetIOSuite.scala:
##########
@@ -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 in = new ByteArrayInputStream(bytes, ParquetFileWriter.MAGIC.length, bytes.length)
Review Comment:
Nit: `bytes.length` is 4 bytes more than what is left after the offset. It works because `ByteArrayInputStream` caps `count` at `buf.length`, which `pageStart` at L2596 also relies on. Could we pass `bytes.length - ParquetFileWriter.MAGIC.length`, like `MapOutputTracker.scala:1780`?
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetVectorizedSuite.scala:
##########
@@ -585,6 +573,66 @@ 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)
+ ).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: invalid DELTA_BINARY_PACKED data 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)
+
+ // 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,
Review Comment:
Nit: this case repeats the setup and three of the four assertions of "a decoding error in a batch reports the page and the column" (L576-L591), without its check of the cause's message (L589). With 4 values it fits there as a third row, since the second value still unpacks the first mini block:
```scala
("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)
```
Could we move it there, so that this test keeps only the header case?
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java:
##########
@@ -272,79 +288,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);
- }
+ // 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();
Review Comment:
Minor: the second bullet of item 5 of my second review was incomplete. A negative `uncompressed_page_size` also breaks the data pages of the codecs that decompress lazily, SNAPPY (the default), GZIP and LZ4, not only ZSTD and LZ4_RAW. For those, `readPage()` returns normally, because parquet-java checks only `compressed_page_size` and wraps the stream as `BytesInput.from(is, -1)`, so this line records `Page [bytes.size=-1, ...]`. Then `bytes.toInputStream()` at L491, or `page.getData().toInputStream()` at L518 for a V2 page, allocates `new byte[-1]` in `StreamBytesInput.toByteArray`. The `NegativeArraySizeException` passes the catches at L497-L502 and L520-L525, so the error names neither the page nor the column. A dictionary page with the same header is wrapped by the new arm at L167 for every compressed codec.
Could we reword the "not covered" item, e.g. "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"?
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedDeltaBinaryPackedReader.java:
##########
@@ -472,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) {
Review Comment:
Minor: the user-facing section of the description says that the original `IllegalArgumentException`, `ArrayIndexOutOfBoundsException` and the others are kept as the cause. That holds only for the three dictionary types (`VectorizedColumnReader.java:175`). Here and at L91 and L108, a new `ParquetDecodingException` is thrown before an `ArrayIndexOutOfBoundsException` or an `IllegalArgumentException` can occur, so the cause is this reader's own exception, as the new tests check (`ParquetVectorizedSuite.scala:622` and `:633`, versus `:611` for the dictionaries). Also, only the value count and mini block size checks threw an `IllegalArgumentException` before; the other header checks already threw a `ParquetDecodingException`.
Since the description becomes the commit message, could we say so, e.g. "The dictionary exceptions are kept as the cause, and the DELTA_BINARY_PACKED checks now throw a `ParquetDecodingException` instead of an `IllegalArgumentException` or `ArrayIndexOutOfBoundsException`"?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59854][SQL] Report the column in decoding errors of the vectorized Parquet reader [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on code in PR #59542:
URL: https://github.com/apache/spark/pull/59542#discussion_r4294716537
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetIOSuite.scala:
##########
@@ -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 =>
Review Comment:
Minor: this repeats the write, patch, delete-`.crc` and read steps of `readCorruptedRleSection` (L2499-L2531), but without its 60-second `failAfter` guard. If a regression made a corrupt ZSTD page hang, this test would block the suite instead of failing. Could we generalize the helper to take the writer options and the corruption step, and reuse it here?
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java:
##########
@@ -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;
Review Comment:
Minor: `currentPage` is only used for error messages, but it is never cleared. When `pageReader.readPage()` returns null at L370, the last page stays reachable until the reader is dropped. Could we set `currentPage = null` in that branch?
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetIOSuite.scala:
##########
@@ -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(
Review Comment:
Nit: `ParquetFileReader.readFooter(conf, path)` is deprecated, and `ParquetTest` already has `readFooter(path, configuration)` (ParquetTest.scala:145), which this suite uses at L1353 and L1527. Could we use it here?
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java:
##########
@@ -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) {
Review Comment:
**This wrapper catches only `ParquetDecodingException`, so an out-of-range dictionary id still fails without the page or the column.** If the RLE id section is intact but holds an id past the end of the dictionary, e.g. id 5 for a 2-entry dictionary, `updater.decodeDictionaryIds` at L335/L342 calls `Dictionary.decodeToInt(5)` and throws an `ArrayIndexOutOfBoundsException`, which passes through here. On the lazy path the same error is thrown later through `ParquetDictionary`, outside `readBatch`. The PR description says the RLE decoding errors are covered. Could we validate the ids against `dictionary.getMaxId()` and throw a `ParquetDecodingException`, or narrow the description and leave this to a follow-up?
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetIOSuite.scala:
##########
@@ -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 ==
Review Comment:
Minor: this pins two things parquet-java does not promise: the exact messages `could not decompress page` and `Could not decompress dictionary page`, and that `ColumnChunkPageReader` decompresses ZSTD pages eagerly. A parquet-java upgrade that rewords the message, or makes ZSTD lazy like SNAPPY so that the error goes through the `could not read page` wrapper of `readPageV1`, would fail this test although Spark still reports the column. Could we check only that the column is reported once, on a `ParquetDecodingException`?
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedDeltaBinaryPackedReader.java:
##########
@@ -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: "
Review Comment:
Nit: the new message prints the double `miniSize`, e.g. `42.666666666666664` in the new test, while the checks above print the header fields as integers. Could we check with integer arithmetic, e.g. `blockSizeInValues % miniBlockNumInABlock != 0 || (blockSizeInValues / miniBlockNumInABlock) % 8 != 0`, and report `blockSizeInValues` and `miniBlockNumInABlock` like L105-L106?
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java:
##########
@@ -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 {
Review Comment:
Nit: wrapping the block in `try` re-indents about 70 unchanged lines, so `git blame` for the dictionary and lazy-decoding logic (SPARK-16334, the decimal and unsigned handling) now points at this PR. Could we move the block into a private method, e.g. `decodePage(...)`, and wrap only that call?
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java:
##########
@@ -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) {
Review Comment:
**This catch reports every `RuntimeException` of `initDictionary` as a corrupt dictionary, and the comment above is not exact.** For a lazily decompressed dictionary page (SNAPPY or GZIP through a stream `BytesInput`), `initDictionary` is also where the page is decompressed, so it does more than parse the bytes. A bug in Spark or parquet-java, such as an NPE or a `ClassCastException`, would also show up as `could not decode the dictionary`, which points users at the file instead of the reader. Could we catch only the exceptions the tests cover (`IndexOutOfBoundsException`, `UnsupportedOperationException`), or at least reword the comment?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59854][SQL] Report the column in decoding errors of the vectorized Parquet reader [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on PR #59542:
URL: https://github.com/apache/spark/pull/59542#issuecomment-6292593343
**Merge Summary:**
- merged into master https://github.com/apache/spark/commit/8c9aa483c1130c2a451e20d34e7f3ad60d393a88
- merged into branch-4.x https://github.com/apache/spark/commit/50f367f5f77e4bdf8e5f35cc72a38a31ea47d35a
*Posted by `merge_spark_pr.py`*
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59854][SQL] Report the column in decoding errors of the vectorized Parquet reader [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on code in PR #59542:
URL: https://github.com/apache/spark/pull/59542#discussion_r4298474479
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java:
##########
@@ -264,76 +286,93 @@ 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);
+ }
+ this.currentPage = page;
Review Comment:
**`currentPage` keeps each column's decompressor stream alive, about 66 KiB per column with the default SNAPPY codec.**
For the codecs that parquet-java decompresses lazily (SNAPPY, GZIP and Hadoop LZ4), the `DataPage` from `pageReader.readPage()` holds a `StreamBytesInput`, which holds the codec's `DecompressorStream` and its `io.file.buffer.size` buffer. Spark sets that size from `spark.buffer.size`, 64 KiB, and Hadoop LZ4 uses a 256 KiB buffer. Before this PR, the page became garbage as soon as `readPage()` returned, because the value readers keep only the stream over the array that `page.getBytes().toInputStream()` copied. Now every `VectorizedColumnReader` keeps its current page. A scan of 1,000 leaf columns with 8 tasks per executor keeps about 500 MiB more heap live, and 100 columns keep about 50 MiB.
The `null` assignment here does not help flat columns. `ParquetReadState` sets `lastListCompleted` for them, so `readPage()` is not called again after the last page, and that page stays reachable until the next row group.
The message only needs what `DataPageV1.toString()` and `DataPageV2.toString()` print, a few sizes and counts. Could we store `String.valueOf(page)`, or those values or a page ordinal, instead of the page?
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetVectorizedSuite.scala:
##########
@@ -551,6 +551,96 @@ class ParquetVectorizedSuite extends ParquetTest with SharedSparkSession {
}
}
+ /**
Review Comment:
**This spot now conflicts with master, so the second commit has not been tested by CI.**
SPARK-60262 (f095057ea8d) added a test right after the SPARK-59832 test, where these helpers and tests go. GitHub reports the PR as not mergeable. The `Check changes` job of 68a4b4e failed at its merge step with `CONFLICT (content): Merge conflict in .../ParquetVectorizedSuite.scala`, so every build and test job was skipped. Only 20ce444 has a green run, so the narrowed multi-catch, the `NegativeArraySizeException` case, the integer mini block check and the rewritten ZSTD test have not run in CI yet.
`VectorizedColumnReader.java` merges cleanly, and SPARK-60262's `ran out of pages` exception is thrown before the new `try`, so keeping both tests should be enough. Could you rebase and rerun CI?
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java:
##########
@@ -264,76 +286,93 @@ 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) {
Review Comment:
Minor: this catches only `ParquetDecodingException`, which `ColumnChunkPageReader` creates from an `IOException` of the codec, so other eagerly decompressed pages still fail without the column.
- parquet-java decompresses LZ4_RAW eagerly in the same branch as ZSTD (`codec instanceof ZstandardCodec || codec instanceof Lz4RawCodec` in `CodecFactory`), but aircompressor reports corrupt input with `MalformedInputException`, a `RuntimeException`. The new ZSTD test with `"compression" -> "lz4_raw"` gives `columnErrors.size == 0`.
- A negative `uncompressed_page_size` in a page header makes `BytesInput.copy` throw a `NegativeArraySizeException` inside `readPage()` or `readDictionaryPage()` for ZSTD and LZ4_RAW, which is not wrapped. For a SNAPPY dictionary page the same exception is thrown inside `initDictionary`, where the new arm at L172 wraps it, so whether the column is reported depends on the codec.
- Hadoop LZ4's `LZ4Exception` in a dictionary page also passes the multi-catch at L172.
Could we also wrap these here and at L159, except `ParquetCryptoRuntimeException`, or narrow the comment and the description to ZSTD?
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java:
##########
@@ -157,6 +169,16 @@ public VectorizedColumnReader(
} 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
Review Comment:
**Some truncated dictionaries pass `initDictionary`, and an uncompressed chunk can then return bytes of the next page with no error.**
This arm fires only when the truncation cuts a BINARY length prefix. parquet-java's `PlainBinaryDictionary` does not check an entry's length against the page, and its FIXED_LEN_BYTE_ARRAY and INT96 branch reads no bytes at all. For example, take a PLAIN string dictionary with two entries, `02 00 00 00 'a' 'b' 05 00 00 00 'c' 'd'`, in an UNCOMPRESSED chunk:
- `initDictionary` succeeds, although the second entry claims 5 bytes and only 2 are left.
- The dictionary buffer is a slice of the buffer that holds the column chunk, and `Binary.getBytes` copies from the backing array at `arrayOffset() + offset`, so the second value reads as `cd` plus three bytes of the next page header.
With SNAPPY or ZSTD the same entry throws an `ArrayIndexOutOfBoundsException` at decode time, without the column, and a truncated FIXED_LEN_BYTE_ARRAY decimal or INT96 dictionary throws an `IllegalArgumentException` inside `readBatch`.
This is how parquet-java behaved before this PR, but this comment and the description suggest that truncated BINARY dictionaries are covered. Could we validate the entries against the page (every `offset + len` within the page for BINARY, `size * length` for FIXED_LEN_BYTE_ARRAY and INT96), or narrow the comment and leave the silent case to a separate JIRA?
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetIOSuite.scala:
##########
@@ -2569,6 +2583,30 @@ 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(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)
Review Comment:
Minor: this check still depends on eager ZSTD decompression. That is my fault: I suggested "the column is reported once, on a `ParquetDecodingException`" in my first review, and it only holds on the eager path.
- For a lazily decompressed data page (SNAPPY and GZIP today, or ZSTD if parquet-java drops the eager copy in `CodecFactory`), the error goes through the `IOException` "could not read page ... in col ..." of `readPageV1`, and then through the visitor's `new RuntimeException(e)` at `VectorizedColumnReader` L387. `Throwable(Throwable)` copies `cause.toString()` into its message, so two exceptions name the column and `size == 1` fails.
- A lazily decompressed dictionary page fails through the `IOException` at L167-L168, so `isInstanceOf[ParquetDecodingException]` fails.
Spark still reports the column in both cases. Could we check only that some exception in the chain names the column, like `causes(e).exists(...)` in the SPARK-59832 test at L2571?
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedDeltaBinaryPackedReader.java:
##########
@@ -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) {
Review Comment:
Minor: like the header fields checked here, the mini block bit widths come from the file, but they are never range-checked, so a corrupt width still fails without the page or the column. `readBitWidthsForMiniBlocks` (L496-L503) reads each width as 0 to 255, and `unpackMiniBlock` passes it to `Packer.LITTLE_ENDIAN.newBytePackerForLong` (L480-L481), which indexes a table of 65 packers. With the bit widths `65 0 0 0` in a block header, reading the second INT32 value throws `ArrayIndexOutOfBoundsException: Index 65 out of bounds for length 65`. That is not a `ParquetDecodingException`, so the new `readBatch` wrapper does not add the column. For DELTA_LENGTH_BYTE_ARRAY and DELTA_BYTE_ARRAY, whose lengths are decoded in `initFromPage`, the `readPageV1` and `readPageV2` wrappers do not either.
The spec allows arbitrary widths for the unused mini blocks of the last block, so the check has to be where a width is used: `if (bitWidth > 64) throw new ParquetDecodingException(...)` in `unpackMiniBlock`, like the `width > 32` check in `VectorizedRleValuesReader`. Since this PR already hardens this reader's header, could we add it here?
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetVectorizedSuite.scala:
##########
@@ -551,6 +551,96 @@ 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(
Review Comment:
Nit: `readCorruptedPage` is a parameterized copy of the SPARK-59832 test right above it (L527-L552). That test keeps its own copy of the same `MemPageStore`, reader, `initialize`, `intercept` and `close` code, and the same one-value definition levels with the same comment (L536-L538 and L616-L619). It is exactly `readCorruptedPage("message root { optional int32 a; }", Some(<its dictionary page>), Array[Byte](2, 0, 0, 0, 2, 1, 2, 2, 0), 1, Encoding.RLE_DICTIONARY)` plus its asserts. Also, `schema` looks free-form, but L565-L566 hard-code the column `a` with max definition level 1.
The helper and `fourNonNullValues` also sit between the tests, while this suite keeps its vals at the top and its helpers after the tests (from `truncateTypeTest` on), as AGENTS.md asks ("follow the sectioning the file already uses"). Since the rebase touches this exact spot, could we move them and reuse the helper in the SPARK-59832 test?
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedDeltaBinaryPackedReader.java:
##########
@@ -104,10 +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;
- JavaUtils.checkArgument(miniSize % 8 == 0,
- "miniBlockSize must be multiple of 8, but it's " + miniSize);
- this.miniBlockSizeInValues = (int) miniSize;
+ if (blockSizeInValues % miniBlockNumInABlock != 0 ||
Review Comment:
Nit: this check makes the `miniBlockNumInABlock > blockSizeInValues` arm at L104 redundant. L100 guarantees `blockSizeInValues >= 1`, so `m > b` implies `b % m == b != 0`, and both checks throw "Invalid DELTA_BINARY_PACKED mini block count: ...". Could we drop that arm from L104? `miniBlockNumInABlock <= 0` has to stay, because it guards the division here and rejects negative counts such as -16 (`128 % -16 == 0`). The `contains("mini block count")` assertion in `ParquetDeltaEncodingSuite` still matches.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetVectorizedSuite.scala:
##########
@@ -551,6 +551,96 @@ 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)
Review Comment:
Minor: this test does not check the page, although its name says it does. `TestPageReader` returns a `TestDataPage`, which has no `toString`, so `currentPage` prints as `TestDataPage@<hash>`, and `startsWith("could not decode page ")` is the only check of that part. If `this.currentPage = page` were removed from `VectorizedColumnReader` (L375), the message would read `could not decode page null in col ...`, and this test would still pass. Production pages are `DataPageV1` or `DataPageV2`, which do have a useful `toString`. Could we make `TestDataPage.toString` delegate to the wrapped page, and assert on it, e.g. `contains("could not decode page Page [bytes.size=")`?
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java:
##########
@@ -157,6 +169,16 @@ public VectorizedColumnReader(
} catch (ParquetDecodingException e) {
throw new ParquetDecodingException(
"could not decode the dictionary for " + descriptor + ": " + e.getMessage(), e);
+ } catch (IndexOutOfBoundsException | NegativeArraySizeException |
Review Comment:
**A huge positive dictionary size still ends in an `OutOfMemoryError`, which kills the executor.**
This arm turns a negative `num_values` into a `ParquetDecodingException`. However, `PlainValuesDictionary` allocates `new int[size]`, `new Binary[size]` and so on without checking the size against the page. Take an INT32 dictionary page header with `num_values = 0x7FFFFFF0` and an executor heap under 8 GiB:
- `initDictionary` throws `OutOfMemoryError: Java heap space`, which none of these catches match.
- `FileDataSourceV2.attachFilePath` rethrows it without the file path, and the executor exits through `SparkUncaughtExceptionHandler`.
- Every retry kills another executor, and `spark.sql.files.ignoreCorruptFiles` cannot skip the file.
FIXED_LEN_BYTE_ARRAY and INT96 dictionaries fail on any realistic heap, because they create one `Binary` per entry.
This is older than the PR, but it is the other sign of the case this arm handles. A check before `initDictionary`, such as `size < 0 || (long) size * minEntryWidth > pageSize`, would cover both signs with a `ParquetDecodingException` that names the column. Could we add it here, or file a separate JIRA?
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java:
##########
@@ -157,6 +169,16 @@ public VectorizedColumnReader(
} 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
Review Comment:
Minor: the PR description still describes the first commit, and it will become the commit message.
- Its "Decoding the dictionary" item says that `initDictionary` only parses the bytes and that any `RuntimeException` from it is wrapped. This catch now takes only three types, and this comment says that `initDictionary` also decompresses.
- The user-facing change and the test list do not mention `NegativeArraySizeException`.
- "SNAPPY and GZIP pages already report the column" does not hold for filtered reads. With a filter pushed on a dictionary-encoded column (`spark.sql.parquet.filterPushdown` and `parquet.filter.dictionary.enabled` are on by default), parquet-java's `DictionaryFilter` decompresses the dictionary page inside `new ParquetFileReader(...)`, before any `VectorizedColumnReader` exists. Reading the new ZSTD test file with `.where("a = 5")` fails with `ParquetDecodingException: Failed to read dictionary`, and with SNAPPY or GZIP it fails with a plain `RuntimeException`, both without the column.
Could we update the description, and mention that the filter path is out of scope?
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedDeltaBinaryPackedReader.java:
##########
@@ -104,10 +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;
- 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) {
Review Comment:
Minor: this half of the condition has no test. The only new test uses a block of 128 values in 3 mini blocks, which already fails `128 % 3 != 0`. If `(blockSizeInValues / miniBlockNumInABlock) % 8 != 0` were dropped, every test would still pass. A header with 128 values in 32 mini blocks would then allocate `unpackedValuesBuffer = new long[4]` at L122 and fail in `unpack8Values` with an `ArrayIndexOutOfBoundsException`, without the column.
`ParquetDeltaEncodingSuite` already unit-tests these header checks in "reject invalid DELTA_BINARY_PACKED page header", with a `craftHeader` helper, for both INT32 and INT64. Could we add `craftHeader(128, 3, 1)`, `craftHeader(128, 32, 1)` and a zero value count there, and drop the direct `initFromPage(0, ...)` check from `ParquetVectorizedSuite` (L636-L641) together with its `ByteBufferInputStream` import?
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java:
##########
@@ -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) {
Review Comment:
Nit: because this `try` is separate from the `initDictionary` one below, a dictionary page that cannot be decompressed fails differently depending on the codec. With ZSTD it is a `ParquetDecodingException`, "could not read the dictionary page for ...". With SNAPPY or GZIP, which decompress inside `initDictionary`, it is an `IOException`, "could not decode the dictionary for ..." (L167-L168). A log search for one misses the other. Moving `pageReader.readDictionaryPage()` into the `try` at L164 would give one message shape, and the `NegativeArraySizeException` arm would then also cover the eager case.
Item 7 of my first review is still open as well. If the decode block of `readBatch` (L293-L358) moved into a private method and only that call were wrapped, `git blame` would keep its history: today about 61 of the changed lines differ only in indentation (`git diff -w` shows 42+/3- instead of 103+/64-). Either way the wrapper should keep catching only `ParquetDecodingException`, since `ParquetStorageFilter.rethrowIfMustPropagate` dispatches on the exact `RuntimeException` class.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetIOSuite.scala:
##########
@@ -2569,6 +2583,30 @@ 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(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(
Review Comment:
Nit: the footer is read only to get `chunk.getStartingPos`, which is always 4 for this local single-column file, and `hasDictionaryPage`. It is also the only reason `readCorruptedFile` passes a `Path` (the other caller ignores it at L2537), and it adds two imports. Could we start the stream at `ParquetFileWriter.MAGIC.length` and assert the type of the parsed header instead (`DICTIONARY_PAGE` or `DATA_PAGE`)? That would check the page that is actually corrupted.
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java:
##########
@@ -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) {
Review Comment:
At 68a4b4e this is unchanged. With a 2-entry INT32 dictionary and an RLE run of id 5:
- Read as `LongType` (eager), `decodeDictionaryIds` throws `ArrayIndexOutOfBoundsException: Index 5 out of bounds for length 2` inside the new `try`, and it is not wrapped because it is not a `ParquetDecodingException`.
- Read as `IntegerType` (lazy), `readBatch` succeeds, and the same exception is thrown later from `ParquetDictionary.decodeToInt` in the consumer. That is outside `FileScanRDD`'s handling too, so the error names neither the file nor the column.
I understand the description says only `ParquetDecodingException` is wrapped, and this was also left out of SPARK-59832. Since the RLE reader does not know the dictionary, a check needs some thought about its per-value cost. Could you either handle it here, or list it in the description as a known gap with a separate JIRA?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59854][SQL] Report the column in decoding errors of the vectorized Parquet reader [spark]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on code in PR #59542:
URL: https://github.com/apache/spark/pull/59542#discussion_r4318670400
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetIOSuite.scala:
##########
@@ -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 in = new ByteArrayInputStream(bytes, ParquetFileWriter.MAGIC.length, bytes.length)
Review Comment:
Done, it now passes `bytes.length - start`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59854][SQL] Report the column in decoding errors of the vectorized Parquet reader [spark]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on code in PR #59542:
URL: https://github.com/apache/spark/pull/59542#discussion_r4304155581
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java:
##########
@@ -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) {
Review Comment:
Filed SPARK-60317 for it, and the description lists it as not covered. I verified both paths: read as `LongType` the `ArrayIndexOutOfBoundsException` is thrown inside `readBatch` and not wrapped, and read as `IntegerType` `readBatch` succeeds and `ParquetDictionary.decodeToInt` throws later in the consumer. Since a check is needed only when the dictionary is smaller than 2^bitWidth, once per RLE run and once per group of 8 for bit-packed runs, I'd like to do it there with a benchmark.
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java:
##########
@@ -264,76 +286,93 @@ 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);
+ }
+ this.currentPage = page;
Review Comment:
Thanks, confirmed: `StreamBytesInput` keeps the decompressor stream in a field, and `toInputStream()` copies the bytes, so before this PR the page was garbage once `readPage()` returned. `currentPage` is now the `String` of the page, set after the null check, so the reader keeps neither the page nor its stream.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetVectorizedSuite.scala:
##########
@@ -551,6 +551,96 @@ class ParquetVectorizedSuite extends ParquetTest with SharedSparkSession {
}
}
+ /**
Review Comment:
Merged master and kept both tests. The conflict was only in `ParquetVectorizedSuite`, and the SPARK-60262 test passes together with the new ones.
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java:
##########
@@ -157,6 +169,16 @@ public VectorizedColumnReader(
} catch (ParquetDecodingException e) {
throw new ParquetDecodingException(
"could not decode the dictionary for " + descriptor + ": " + e.getMessage(), e);
+ } catch (IndexOutOfBoundsException | NegativeArraySizeException |
Review Comment:
Confirmed: an INT32 dictionary page with `num_values = 0x7FFFFFF0` throws `OutOfMemoryError: Java heap space` from `initDictionary`, which none of the catches match. I filed SPARK-60316 for checking the dictionary page against its size before `initDictionary`, which covers this together with the next comment, and listed it in the description as not covered.
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java:
##########
@@ -157,6 +169,16 @@ public VectorizedColumnReader(
} 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
Review Comment:
Confirmed. With an UNCOMPRESSED chunk the string dictionary `02 00 00 00 'a' 'b' 05 00 00 00 'c' 'd'` reads as `cd` followed by `15 00 15`, the start of the next page header, with both the vectorized and the non-vectorized reader, so this is not specific to this reader. That is SPARK-60316 as well. I narrowed the comment: the `IndexOutOfBoundsException` arm is for a BINARY dictionary that ends within the length of an entry, and an entry that claims more bytes than the page has is not detected here.
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java:
##########
@@ -264,76 +286,93 @@ 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) {
Review Comment:
Both confirmed: LZ4_RAW's `MalformedInputException` is a `RuntimeException` that parquet-java's `Lz4RawDecompressor` does not turn into an `IOException`, and a negative uncompressed size makes `StreamBytesInput.toByteArray` throw a `NegativeArraySizeException` in `readPage()`. I kept the catch to `ParquetDecodingException`, since `readPage()` also decrypts and I'd rather not label everything else as a decoding error, and listed both cases in the description as not covered. For the dictionary page, `readDictionaryPage()` now shares the `try` of `initDictionary` (see the comment at L159), so its `NegativeArraySizeException` is wrapped with the column whatever the codec.
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedDeltaBinaryPackedReader.java:
##########
@@ -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) {
Review Comment:
Done. `unpackMiniBlock` now rejects a bit width above 64 with a `ParquetDecodingException`, where the width is used, and a new test with the widths `65 0 0 0` checks that the batch wrapper adds the page and the column.
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java:
##########
@@ -157,6 +169,16 @@ public VectorizedColumnReader(
} 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
Review Comment:
Updated the description: the dictionary item now matches the narrowed catch, `NegativeArraySizeException` is in the user-facing change and the tests, and a "not covered" list names SPARK-60317, SPARK-60316, LZ4_RAW, a negative uncompressed size of a data page, and the `DictionaryFilter` path of filtered reads. The SNAPPY and GZIP sentence now says it is about pages that `VectorizedColumnReader` reads.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetIOSuite.scala:
##########
@@ -2569,6 +2583,30 @@ 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(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)
Review Comment:
Done, the test now checks only that some exception in the chain names the column.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59854][SQL] Report the column in decoding errors of the vectorized Parquet reader [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on PR #59542:
URL: https://github.com/apache/spark/pull/59542#issuecomment-6292612310
This landed to branch-4.x according to the JIRA info.
<img width="701" height="249" alt="Screenshot 2026-10-20 at 6 58 26 AM" src="https://github.com/user-attachments/assets/2cae82bc-76fb-4e8c-b1d9-836161da188c" />
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59854][SQL] Report the column in decoding errors of the vectorized Parquet reader [spark]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on code in PR #59542:
URL: https://github.com/apache/spark/pull/59542#discussion_r4304156546
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetVectorizedSuite.scala:
##########
@@ -551,6 +551,96 @@ 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)
Review Comment:
Done. `TestDataPage.toString` now delegates to the wrapped page, and the test checks `could not decode page Page [bytes.size=`, so it would fail if the page were missing.
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedDeltaBinaryPackedReader.java:
##########
@@ -104,10 +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;
- 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) {
Review Comment:
Done. `ParquetDeltaEncodingSuite` now checks 128 values in 3 and in 32 mini blocks and a page with no values, and I dropped the direct `initFromPage(0, ...)` check and its import from `ParquetVectorizedSuite`.
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java:
##########
@@ -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) {
Review Comment:
Done for the first part: `readDictionaryPage()` is now in the same `try` as `initDictionary`, so a dictionary page that cannot be read fails with `could not decode the dictionary for ...` for every codec.
For the helper method, I tried it and measured `git blame` on the decode block. Plain `git blame` attributes 64 of its lines to this PR in both versions, `-w` 8 in both, and `-w -M` 7 with the `try` and 6 with the method. Moving the block into a method body re-indents it as well (to 4 spaces, not back to the original 6), so it does not keep the history. So I kept the `try`. Agreed that the wrapper must keep catching only `ParquetDecodingException`, so that a vector growth failure, which `rethrowIfMustPropagate` checks by its exact class, is not wrapped.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetVectorizedSuite.scala:
##########
@@ -551,6 +551,96 @@ 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(
Review Comment:
Done. The SPARK-59832 test now uses `readCorruptedPage`, which takes the column type instead of a schema, so it no longer looks free-form. The helper sits with the other helpers after `writeDataPage`, and the definition level vals are at the top of the suite.
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedDeltaBinaryPackedReader.java:
##########
@@ -104,10 +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;
- JavaUtils.checkArgument(miniSize % 8 == 0,
- "miniBlockSize must be multiple of 8, but it's " + miniSize);
- this.miniBlockSizeInValues = (int) miniSize;
+ if (blockSizeInValues % miniBlockNumInABlock != 0 ||
Review Comment:
Done, the `miniBlockNumInABlock > blockSizeInValues` arm is gone and `<= 0` stays. The existing `craftHeader(128, 256, 1)` case still gets the "mini block count" message.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetIOSuite.scala:
##########
@@ -2569,6 +2583,30 @@ 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(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(
Review Comment:
Done. The test parses the page header at `ParquetFileWriter.MAGIC.length` and asserts its type, `DATA_PAGE` or `DICTIONARY_PAGE`, and `readCorruptedFile` no longer passes the path.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59854][SQL] Report the column in decoding errors of the vectorized Parquet reader [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun closed pull request #59542: [SPARK-59854][SQL] Report the column in decoding errors of the vectorized Parquet reader
URL: https://github.com/apache/spark/pull/59542
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59854][SQL] Report the column in decoding errors of the vectorized Parquet reader [spark]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on code in PR #59542:
URL: https://github.com/apache/spark/pull/59542#discussion_r4318668599
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java:
##########
@@ -146,21 +152,31 @@ public 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 |
Review Comment:
Confirmed: with `compression=lz4`, a corrupt LZ4 block in a data page and in a dictionary page both fail with only `net.jpountz.lz4.LZ4Exception: Error decoding offset 10 of input buffer`. The "not covered" list now has LZ4 next to LZ4_RAW, without "as it does for the other codecs".
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java:
##########
@@ -272,79 +288,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);
- }
+ // 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();
Review Comment:
Confirmed: with a SNAPPY data page whose `uncompressed_page_size` is -65, `readPage()` returns and the read fails with only `NegativeArraySizeException: -65`. I reworded the item as you suggested.
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedDeltaBinaryPackedReader.java:
##########
@@ -472,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) {
Review Comment:
Right, thanks. The user-facing section now says that the dictionary exceptions are kept as the cause, and that the DELTA_BINARY_PACKED checks throw a `ParquetDecodingException` instead of the `IllegalArgumentException` of the value count and mini block size checks or the `ArrayIndexOutOfBoundsException` of a bit width above 64.
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java:
##########
@@ -146,21 +152,31 @@ public 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");
Review Comment:
Done, it now reads `totalValueCount == 0 in col <column>`, and a new test reads a column chunk without pages.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetVectorizedSuite.scala:
##########
@@ -585,6 +573,66 @@ 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)
+ ).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: invalid DELTA_BINARY_PACKED data 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)
+
+ // 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,
Review Comment:
Done, the bit width case is now the third row of the batch test, which also checks the cause's message, and the DELTA_BINARY_PACKED test keeps only the header case.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetVectorizedSuite.scala:
##########
@@ -887,6 +935,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()
+ }
+ }
Review Comment:
Done, fixed both.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59854][SQL] Report the column in decoding errors of the vectorized Parquet reader [spark]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on PR #59542:
URL: https://github.com/apache/spark/pull/59542#issuecomment-6295432604
Thank you @dongjoon-hyun !
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59854][SQL] Report the column in decoding errors of the vectorized Parquet reader [spark]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on code in PR #59542:
URL: https://github.com/apache/spark/pull/59542#discussion_r4298491407
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java:
##########
@@ -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 {
Review Comment:
A private method would re-indent the same lines too: the block is at 6 spaces inside the `while`, 8 inside the `try`, and would be at 4 in a method body. Moving it would also make `git blame` need `-M` rather than just `-w`. So I kept the `try`; `git blame -w` and the diff with `?w=1` skip the re-indentation.
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedDeltaBinaryPackedReader.java:
##########
@@ -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: "
Review Comment:
Done. The check now uses integer arithmetic and reports the mini block count and the block size, like the checks above.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org