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