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: [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]
