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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
