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