dongjoon-hyun commented on code in PR #59542:
URL: https://github.com/apache/spark/pull/59542#discussion_r4294716537
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetIOSuite.scala:
##########
@@ -2569,6 +2571,47 @@ class ParquetIOSuite extends ParquetTest with
SharedSparkSession {
val e = readCorruptedRleSection("id >= 0 as b", writerVersion =
"PARQUET_2_0")(setMaxLength)
assertInvalidLengthReportsColumn(e, "b")
}
+
+ test("SPARK-59854: a ZSTD page that cannot be decompressed reports the
column") {
+ // ZSTD pages are decompressed when they are read from the column chunk,
not when decoded.
+ Seq(
+ (false, "could not read the next page in col", "could not decompress
page"),
+ (true, "could not read the dictionary page for", "Could not decompress
dictionary page")
+ ).foreach { case (dictionary, error, cause) =>
+ withTempPath { dir =>
Review Comment:
Minor: this repeats the write, patch, delete-`.crc` and read steps of
`readCorruptedRleSection` (L2499-L2531), but without its 60-second `failAfter`
guard. If a regression made a corrupt ZSTD page hang, this test would block the
suite instead of failing. Could we generalize the helper to take the writer
options and the corruption step, and reuse it here?
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java:
##########
@@ -264,79 +281,96 @@ void readBatch(
}
readState.resetForNewPage(pageValueCount, pageFirstRowIndex);
}
- PrimitiveType.PrimitiveTypeName typeName =
- descriptor.getPrimitiveType().getPrimitiveTypeName();
- if (isCurrentPageDictionaryEncoded) {
- // Save starting offset in case we need to decode dictionary IDs.
- int startOffset = readState.valueOffset;
- // Save starting row index so we can check if we need to eagerly
decode dict ids later
- long startRowId = readState.rowId;
-
- // Read and decode dictionary ids.
- if (readState.maxRepetitionLevel == 0) {
- defColumn.readIntegers(readState, dictionaryIds, column,
definitionLevels,
- (VectorizedValuesReader) dataColumn);
- } else {
- repColumn.readIntegersRepeated(readState, repetitionLevels,
defColumn, definitionLevels,
- dictionaryIds, column, (VectorizedValuesReader) dataColumn);
- }
-
- // TIMESTAMP_MILLIS encoded as INT64 can't be lazily decoded as we
need to post process
- // the values to add microseconds precision.
- if (column.hasDictionary() || (startRowId == pageFirstRowIndex &&
- isLazyDecodingSupported(typeName, column.dataType()))) {
- // Column vector supports lazy decoding of dictionary values so just
set the dictionary.
- // We can't do this if startRowId is not the first row index in the
page AND the column
- // doesn't have a dictionary (i.e. some non-dictionary encoded
values have already been
- // added).
- PrimitiveType primitiveType = descriptor.getPrimitiveType();
-
- // We need to make sure that we initialize the right type for the
dictionary otherwise
- // WritableColumnVector will throw an exception when trying to
decode to an Int when the
- // dictionary is in fact initialized as Long
- LogicalTypeAnnotation typeAnnotation =
primitiveType.getLogicalTypeAnnotation();
- boolean castLongToInt =
- typeAnnotation instanceof DecimalLogicalTypeAnnotation annotation
&&
- annotation.getPrecision() <= Decimal.MAX_INT_DIGITS() &&
- primitiveType.getPrimitiveTypeName() == INT64;
-
- // We require a long value, but we need to use dictionary to decode
the original
- // signed int first
- boolean isUnsignedInt32 =
updaterFactory.isUnsignedIntTypeMatched(32);
-
- // We require a decimal value, but we need to use dictionary to
decode the original
- // signed long first
- boolean isUnsignedInt64 =
updaterFactory.isUnsignedIntTypeMatched(64);
-
- boolean needTransform = castLongToInt || isUnsignedInt32 ||
isUnsignedInt64;
- column.setDictionary(new ParquetDictionary(dictionary,
needTransform));
- } else {
- updater.decodeDictionaryIds(readState.valueOffset - startOffset,
startOffset, column,
- dictionaryIds, dictionary);
- }
- } else {
- if (column.hasDictionary() && readState.valueOffset != 0) {
- // This batch already has dictionary encoded values but this new
page is not. The batch
- // does not support a mix of dictionary and not so we will decode
the dictionary.
- updater.decodeDictionaryIds(readState.valueOffset, 0, column,
dictionaryIds, dictionary);
- }
- column.setDictionary(null);
- VectorizedValuesReader valuesReader = (VectorizedValuesReader)
dataColumn;
- if (readState.maxRepetitionLevel == 0) {
- defColumn.readBatch(readState, column, definitionLevels,
valuesReader, updater);
+ // A corrupted page makes the readers below throw a
ParquetDecodingException without the
+ // column. Only that is wrapped, so that other exceptions, e.g.
SparkUpgradeException,
+ // keep their types.
+ try {
+ PrimitiveType.PrimitiveTypeName typeName =
+ descriptor.getPrimitiveType().getPrimitiveTypeName();
+ if (isCurrentPageDictionaryEncoded) {
+ // Save starting offset in case we need to decode dictionary IDs.
+ int startOffset = readState.valueOffset;
+ // Save starting row index so we can check if we need to eagerly
decode dict ids later
+ long startRowId = readState.rowId;
+
+ // Read and decode dictionary ids.
+ if (readState.maxRepetitionLevel == 0) {
+ defColumn.readIntegers(readState, dictionaryIds, column,
definitionLevels,
+ (VectorizedValuesReader) dataColumn);
+ } else {
+ repColumn.readIntegersRepeated(readState, repetitionLevels,
defColumn, definitionLevels,
+ dictionaryIds, column, (VectorizedValuesReader) dataColumn);
+ }
+
+ // TIMESTAMP_MILLIS encoded as INT64 can't be lazily decoded as we
need to post process
+ // the values to add microseconds precision.
+ if (column.hasDictionary() || (startRowId == pageFirstRowIndex &&
+ isLazyDecodingSupported(typeName, column.dataType()))) {
+ // Column vector supports lazy decoding of dictionary values so
just set the dictionary.
+ // We can't do this if startRowId is not the first row index in
the page AND the column
+ // doesn't have a dictionary (i.e. some non-dictionary encoded
values have already been
+ // added).
+ PrimitiveType primitiveType = descriptor.getPrimitiveType();
+
+ // We need to make sure that we initialize the right type for the
dictionary otherwise
+ // WritableColumnVector will throw an exception when trying to
decode to an Int when the
+ // dictionary is in fact initialized as Long
+ LogicalTypeAnnotation typeAnnotation =
primitiveType.getLogicalTypeAnnotation();
+ boolean castLongToInt =
+ typeAnnotation instanceof DecimalLogicalTypeAnnotation
annotation &&
+ annotation.getPrecision() <= Decimal.MAX_INT_DIGITS() &&
+ primitiveType.getPrimitiveTypeName() == INT64;
+
+ // We require a long value, but we need to use dictionary to
decode the original
+ // signed int first
+ boolean isUnsignedInt32 =
updaterFactory.isUnsignedIntTypeMatched(32);
+
+ // We require a decimal value, but we need to use dictionary to
decode the original
+ // signed long first
+ boolean isUnsignedInt64 =
updaterFactory.isUnsignedIntTypeMatched(64);
+
+ boolean needTransform = castLongToInt || isUnsignedInt32 ||
isUnsignedInt64;
+ column.setDictionary(new ParquetDictionary(dictionary,
needTransform));
+ } else {
+ updater.decodeDictionaryIds(readState.valueOffset - startOffset,
startOffset, column,
+ dictionaryIds, dictionary);
+ }
} else {
- repColumn.readBatchRepeated(readState, repetitionLevels, defColumn,
definitionLevels,
- column, valuesReader, updater);
+ if (column.hasDictionary() && readState.valueOffset != 0) {
+ // This batch already has dictionary encoded values but this new
page is not. The batch
+ // does not support a mix of dictionary and not so we will decode
the dictionary.
+ updater.decodeDictionaryIds(readState.valueOffset, 0, column,
dictionaryIds,
+ dictionary);
+ }
+ column.setDictionary(null);
+ VectorizedValuesReader valuesReader = (VectorizedValuesReader)
dataColumn;
+ if (readState.maxRepetitionLevel == 0) {
+ defColumn.readBatch(readState, column, definitionLevels,
valuesReader, updater);
+ } else {
+ repColumn.readBatchRepeated(readState, repetitionLevels,
defColumn, definitionLevels,
+ column, valuesReader, updater);
+ }
}
+ } catch (ParquetDecodingException e) {
+ throw new ParquetDecodingException("could not decode page " +
currentPage + " in col " +
+ descriptor + ": " + e.getMessage(), e);
}
}
}
private int readPage() {
- DataPage page = pageReader.readPage();
+ DataPage page;
+ try {
+ // This decompresses the page eagerly for some codecs, e.g. ZSTD.
+ page = pageReader.readPage();
+ } catch (ParquetDecodingException e) {
+ throw new ParquetDecodingException(
+ "could not read the next page in col " + descriptor + ": " +
e.getMessage(), e);
+ }
if (page == null) {
return -1;
}
+ this.currentPage = page;
Review Comment:
Minor: `currentPage` is only used for error messages, but it is never
cleared. When `pageReader.readPage()` returns null at L370, the last page stays
reachable until the reader is dropped. Could we set `currentPage = null` in
that branch?
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetIOSuite.scala:
##########
@@ -2569,6 +2571,47 @@ class ParquetIOSuite extends ParquetTest with
SharedSparkSession {
val e = readCorruptedRleSection("id >= 0 as b", writerVersion =
"PARQUET_2_0")(setMaxLength)
assertInvalidLengthReportsColumn(e, "b")
}
+
+ test("SPARK-59854: a ZSTD page that cannot be decompressed reports the
column") {
+ // ZSTD pages are decompressed when they are read from the column chunk,
not when decoded.
+ Seq(
+ (false, "could not read the next page in col", "could not decompress
page"),
+ (true, "could not read the dictionary page for", "Could not decompress
dictionary page")
+ ).foreach { case (dictionary, error, cause) =>
+ withTempPath { dir =>
+ val path = dir.getCanonicalPath
+ spark.range(1000).selectExpr("cast(id % 100 as int) as a").coalesce(1)
+ .write
+ .option(ParquetOutputFormat.ENABLE_DICTIONARY, dictionary.toString)
+ .option("compression", "zstd")
+ .parquet(path)
+ val file = dir.listFiles().filter(_.getName.endsWith(".parquet")).head
+ val footer = ParquetFileReader.readFooter(
Review Comment:
Nit: `ParquetFileReader.readFooter(conf, path)` is deprecated, and
`ParquetTest` already has `readFooter(path, configuration)`
(ParquetTest.scala:145), which this suite uses at L1353 and L1527. Could we use
it here?
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java:
##########
@@ -264,79 +281,96 @@ void readBatch(
}
readState.resetForNewPage(pageValueCount, pageFirstRowIndex);
}
- PrimitiveType.PrimitiveTypeName typeName =
- descriptor.getPrimitiveType().getPrimitiveTypeName();
- if (isCurrentPageDictionaryEncoded) {
- // Save starting offset in case we need to decode dictionary IDs.
- int startOffset = readState.valueOffset;
- // Save starting row index so we can check if we need to eagerly
decode dict ids later
- long startRowId = readState.rowId;
-
- // Read and decode dictionary ids.
- if (readState.maxRepetitionLevel == 0) {
- defColumn.readIntegers(readState, dictionaryIds, column,
definitionLevels,
- (VectorizedValuesReader) dataColumn);
- } else {
- repColumn.readIntegersRepeated(readState, repetitionLevels,
defColumn, definitionLevels,
- dictionaryIds, column, (VectorizedValuesReader) dataColumn);
- }
-
- // TIMESTAMP_MILLIS encoded as INT64 can't be lazily decoded as we
need to post process
- // the values to add microseconds precision.
- if (column.hasDictionary() || (startRowId == pageFirstRowIndex &&
- isLazyDecodingSupported(typeName, column.dataType()))) {
- // Column vector supports lazy decoding of dictionary values so just
set the dictionary.
- // We can't do this if startRowId is not the first row index in the
page AND the column
- // doesn't have a dictionary (i.e. some non-dictionary encoded
values have already been
- // added).
- PrimitiveType primitiveType = descriptor.getPrimitiveType();
-
- // We need to make sure that we initialize the right type for the
dictionary otherwise
- // WritableColumnVector will throw an exception when trying to
decode to an Int when the
- // dictionary is in fact initialized as Long
- LogicalTypeAnnotation typeAnnotation =
primitiveType.getLogicalTypeAnnotation();
- boolean castLongToInt =
- typeAnnotation instanceof DecimalLogicalTypeAnnotation annotation
&&
- annotation.getPrecision() <= Decimal.MAX_INT_DIGITS() &&
- primitiveType.getPrimitiveTypeName() == INT64;
-
- // We require a long value, but we need to use dictionary to decode
the original
- // signed int first
- boolean isUnsignedInt32 =
updaterFactory.isUnsignedIntTypeMatched(32);
-
- // We require a decimal value, but we need to use dictionary to
decode the original
- // signed long first
- boolean isUnsignedInt64 =
updaterFactory.isUnsignedIntTypeMatched(64);
-
- boolean needTransform = castLongToInt || isUnsignedInt32 ||
isUnsignedInt64;
- column.setDictionary(new ParquetDictionary(dictionary,
needTransform));
- } else {
- updater.decodeDictionaryIds(readState.valueOffset - startOffset,
startOffset, column,
- dictionaryIds, dictionary);
- }
- } else {
- if (column.hasDictionary() && readState.valueOffset != 0) {
- // This batch already has dictionary encoded values but this new
page is not. The batch
- // does not support a mix of dictionary and not so we will decode
the dictionary.
- updater.decodeDictionaryIds(readState.valueOffset, 0, column,
dictionaryIds, dictionary);
- }
- column.setDictionary(null);
- VectorizedValuesReader valuesReader = (VectorizedValuesReader)
dataColumn;
- if (readState.maxRepetitionLevel == 0) {
- defColumn.readBatch(readState, column, definitionLevels,
valuesReader, updater);
+ // A corrupted page makes the readers below throw a
ParquetDecodingException without the
+ // column. Only that is wrapped, so that other exceptions, e.g.
SparkUpgradeException,
+ // keep their types.
+ try {
+ PrimitiveType.PrimitiveTypeName typeName =
+ descriptor.getPrimitiveType().getPrimitiveTypeName();
+ if (isCurrentPageDictionaryEncoded) {
+ // Save starting offset in case we need to decode dictionary IDs.
+ int startOffset = readState.valueOffset;
+ // Save starting row index so we can check if we need to eagerly
decode dict ids later
+ long startRowId = readState.rowId;
+
+ // Read and decode dictionary ids.
+ if (readState.maxRepetitionLevel == 0) {
+ defColumn.readIntegers(readState, dictionaryIds, column,
definitionLevels,
+ (VectorizedValuesReader) dataColumn);
+ } else {
+ repColumn.readIntegersRepeated(readState, repetitionLevels,
defColumn, definitionLevels,
+ dictionaryIds, column, (VectorizedValuesReader) dataColumn);
+ }
+
+ // TIMESTAMP_MILLIS encoded as INT64 can't be lazily decoded as we
need to post process
+ // the values to add microseconds precision.
+ if (column.hasDictionary() || (startRowId == pageFirstRowIndex &&
+ isLazyDecodingSupported(typeName, column.dataType()))) {
+ // Column vector supports lazy decoding of dictionary values so
just set the dictionary.
+ // We can't do this if startRowId is not the first row index in
the page AND the column
+ // doesn't have a dictionary (i.e. some non-dictionary encoded
values have already been
+ // added).
+ PrimitiveType primitiveType = descriptor.getPrimitiveType();
+
+ // We need to make sure that we initialize the right type for the
dictionary otherwise
+ // WritableColumnVector will throw an exception when trying to
decode to an Int when the
+ // dictionary is in fact initialized as Long
+ LogicalTypeAnnotation typeAnnotation =
primitiveType.getLogicalTypeAnnotation();
+ boolean castLongToInt =
+ typeAnnotation instanceof DecimalLogicalTypeAnnotation
annotation &&
+ annotation.getPrecision() <= Decimal.MAX_INT_DIGITS() &&
+ primitiveType.getPrimitiveTypeName() == INT64;
+
+ // We require a long value, but we need to use dictionary to
decode the original
+ // signed int first
+ boolean isUnsignedInt32 =
updaterFactory.isUnsignedIntTypeMatched(32);
+
+ // We require a decimal value, but we need to use dictionary to
decode the original
+ // signed long first
+ boolean isUnsignedInt64 =
updaterFactory.isUnsignedIntTypeMatched(64);
+
+ boolean needTransform = castLongToInt || isUnsignedInt32 ||
isUnsignedInt64;
+ column.setDictionary(new ParquetDictionary(dictionary,
needTransform));
+ } else {
+ updater.decodeDictionaryIds(readState.valueOffset - startOffset,
startOffset, column,
+ dictionaryIds, dictionary);
+ }
} else {
- repColumn.readBatchRepeated(readState, repetitionLevels, defColumn,
definitionLevels,
- column, valuesReader, updater);
+ if (column.hasDictionary() && readState.valueOffset != 0) {
+ // This batch already has dictionary encoded values but this new
page is not. The batch
+ // does not support a mix of dictionary and not so we will decode
the dictionary.
+ updater.decodeDictionaryIds(readState.valueOffset, 0, column,
dictionaryIds,
+ dictionary);
+ }
+ column.setDictionary(null);
+ VectorizedValuesReader valuesReader = (VectorizedValuesReader)
dataColumn;
+ if (readState.maxRepetitionLevel == 0) {
+ defColumn.readBatch(readState, column, definitionLevels,
valuesReader, updater);
+ } else {
+ repColumn.readBatchRepeated(readState, repetitionLevels,
defColumn, definitionLevels,
+ column, valuesReader, updater);
+ }
}
+ } catch (ParquetDecodingException e) {
Review Comment:
**This wrapper catches only `ParquetDecodingException`, so an out-of-range
dictionary id still fails without the page or the column.** If the RLE id
section is intact but holds an id past the end of the dictionary, e.g. id 5 for
a 2-entry dictionary, `updater.decodeDictionaryIds` at L335/L342 calls
`Dictionary.decodeToInt(5)` and throws an `ArrayIndexOutOfBoundsException`,
which passes through here. On the lazy path the same error is thrown later
through `ParquetDictionary`, outside `readBatch`. The PR description says the
RLE decoding errors are covered. Could we validate the ids against
`dictionary.getMaxId()` and throw a `ParquetDecodingException`, or narrow the
description and leave this to a follow-up?
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetIOSuite.scala:
##########
@@ -2569,6 +2571,47 @@ class ParquetIOSuite extends ParquetTest with
SharedSparkSession {
val e = readCorruptedRleSection("id >= 0 as b", writerVersion =
"PARQUET_2_0")(setMaxLength)
assertInvalidLengthReportsColumn(e, "b")
}
+
+ test("SPARK-59854: a ZSTD page that cannot be decompressed reports the
column") {
+ // ZSTD pages are decompressed when they are read from the column chunk,
not when decoded.
+ Seq(
+ (false, "could not read the next page in col", "could not decompress
page"),
+ (true, "could not read the dictionary page for", "Could not decompress
dictionary page")
+ ).foreach { case (dictionary, error, cause) =>
+ withTempPath { dir =>
+ val path = dir.getCanonicalPath
+ spark.range(1000).selectExpr("cast(id % 100 as int) as a").coalesce(1)
+ .write
+ .option(ParquetOutputFormat.ENABLE_DICTIONARY, dictionary.toString)
+ .option("compression", "zstd")
+ .parquet(path)
+ val file = dir.listFiles().filter(_.getName.endsWith(".parquet")).head
+ val footer = ParquetFileReader.readFooter(
+ spark.sessionState.newHadoopConf(), new Path(file.getCanonicalPath))
+ val chunk = footer.getBlocks.get(0).getColumns.get(0)
+ assert(chunk.hasDictionaryPage == dictionary)
+ // Invert the first bytes of the first page, the ZSTD frame magic
number.
+ val bytes = Files.readAllBytes(file.toPath)
+ val in = new ByteArrayInputStream(bytes, chunk.getStartingPos.toInt,
bytes.length)
+ Util.readPageHeader(in)
+ val pageStart = bytes.length - in.available()
+ (pageStart until pageStart + 4).foreach(i => bytes(i) =
(~bytes(i)).toByte)
+ Files.write(file.toPath, bytes)
+ dir.listFiles().filter(_.getName.endsWith(".crc")).foreach(_.delete())
+
+ withSQLConf(SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key -> "true") {
+ val e = intercept[SparkException](spark.read.parquet(path).collect())
+ val columnErrors =
+ causes(e).filter(c =>
Option(c.getMessage).exists(_.contains("optional int32 a")))
+ assert(columnErrors.size == 1, e)
+ assert(columnErrors.head.getMessage ==
Review Comment:
Minor: this pins two things parquet-java does not promise: the exact
messages `could not decompress page` and `Could not decompress dictionary
page`, and that `ColumnChunkPageReader` decompresses ZSTD pages eagerly. A
parquet-java upgrade that rewords the message, or makes ZSTD lazy like SNAPPY
so that the error goes through the `could not read page` wrapper of
`readPageV1`, would fail this test although Spark still reports the column.
Could we check only that the column is reported once, on a
`ParquetDecodingException`?
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedDeltaBinaryPackedReader.java:
##########
@@ -105,8 +106,10 @@ public void initFromPage(int valueCount,
ByteBufferInputStream in) throws IOExce
+ miniBlockNumInABlock + " (block size in values: " +
blockSizeInValues + ")");
}
double miniSize = (double) blockSizeInValues / miniBlockNumInABlock;
- JavaUtils.checkArgument(miniSize % 8 == 0,
- "miniBlockSize must be multiple of 8, but it's " + miniSize);
+ if (miniSize % 8 != 0) {
+ throw new ParquetDecodingException("Invalid DELTA_BINARY_PACKED mini
block size: "
Review Comment:
Nit: the new message prints the double `miniSize`, e.g. `42.666666666666664`
in the new test, while the checks above print the header fields as integers.
Could we check with integer arithmetic, e.g. `blockSizeInValues %
miniBlockNumInABlock != 0 || (blockSizeInValues / miniBlockNumInABlock) % 8 !=
0`, and report `blockSizeInValues` and `miniBlockNumInABlock` like L105-L106?
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java:
##########
@@ -264,79 +281,96 @@ void readBatch(
}
readState.resetForNewPage(pageValueCount, pageFirstRowIndex);
}
- PrimitiveType.PrimitiveTypeName typeName =
- descriptor.getPrimitiveType().getPrimitiveTypeName();
- if (isCurrentPageDictionaryEncoded) {
- // Save starting offset in case we need to decode dictionary IDs.
- int startOffset = readState.valueOffset;
- // Save starting row index so we can check if we need to eagerly
decode dict ids later
- long startRowId = readState.rowId;
-
- // Read and decode dictionary ids.
- if (readState.maxRepetitionLevel == 0) {
- defColumn.readIntegers(readState, dictionaryIds, column,
definitionLevels,
- (VectorizedValuesReader) dataColumn);
- } else {
- repColumn.readIntegersRepeated(readState, repetitionLevels,
defColumn, definitionLevels,
- dictionaryIds, column, (VectorizedValuesReader) dataColumn);
- }
-
- // TIMESTAMP_MILLIS encoded as INT64 can't be lazily decoded as we
need to post process
- // the values to add microseconds precision.
- if (column.hasDictionary() || (startRowId == pageFirstRowIndex &&
- isLazyDecodingSupported(typeName, column.dataType()))) {
- // Column vector supports lazy decoding of dictionary values so just
set the dictionary.
- // We can't do this if startRowId is not the first row index in the
page AND the column
- // doesn't have a dictionary (i.e. some non-dictionary encoded
values have already been
- // added).
- PrimitiveType primitiveType = descriptor.getPrimitiveType();
-
- // We need to make sure that we initialize the right type for the
dictionary otherwise
- // WritableColumnVector will throw an exception when trying to
decode to an Int when the
- // dictionary is in fact initialized as Long
- LogicalTypeAnnotation typeAnnotation =
primitiveType.getLogicalTypeAnnotation();
- boolean castLongToInt =
- typeAnnotation instanceof DecimalLogicalTypeAnnotation annotation
&&
- annotation.getPrecision() <= Decimal.MAX_INT_DIGITS() &&
- primitiveType.getPrimitiveTypeName() == INT64;
-
- // We require a long value, but we need to use dictionary to decode
the original
- // signed int first
- boolean isUnsignedInt32 =
updaterFactory.isUnsignedIntTypeMatched(32);
-
- // We require a decimal value, but we need to use dictionary to
decode the original
- // signed long first
- boolean isUnsignedInt64 =
updaterFactory.isUnsignedIntTypeMatched(64);
-
- boolean needTransform = castLongToInt || isUnsignedInt32 ||
isUnsignedInt64;
- column.setDictionary(new ParquetDictionary(dictionary,
needTransform));
- } else {
- updater.decodeDictionaryIds(readState.valueOffset - startOffset,
startOffset, column,
- dictionaryIds, dictionary);
- }
- } else {
- if (column.hasDictionary() && readState.valueOffset != 0) {
- // This batch already has dictionary encoded values but this new
page is not. The batch
- // does not support a mix of dictionary and not so we will decode
the dictionary.
- updater.decodeDictionaryIds(readState.valueOffset, 0, column,
dictionaryIds, dictionary);
- }
- column.setDictionary(null);
- VectorizedValuesReader valuesReader = (VectorizedValuesReader)
dataColumn;
- if (readState.maxRepetitionLevel == 0) {
- defColumn.readBatch(readState, column, definitionLevels,
valuesReader, updater);
+ // A corrupted page makes the readers below throw a
ParquetDecodingException without the
+ // column. Only that is wrapped, so that other exceptions, e.g.
SparkUpgradeException,
+ // keep their types.
+ try {
Review Comment:
Nit: wrapping the block in `try` re-indents about 70 unchanged lines, so
`git blame` for the dictionary and lazy-decoding logic (SPARK-16334, the
decimal and unsigned handling) now points at this PR. Could we move the block
into a private method, e.g. `decodePage(...)`, and wrap only that call?
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java:
##########
@@ -157,6 +169,11 @@ public VectorizedColumnReader(
} catch (ParquetDecodingException e) {
throw new ParquetDecodingException(
"could not decode the dictionary for " + descriptor + ": " +
e.getMessage(), e);
+ } catch (RuntimeException e) {
Review Comment:
**This catch reports every `RuntimeException` of `initDictionary` as a
corrupt dictionary, and the comment above is not exact.** For a lazily
decompressed dictionary page (SNAPPY or GZIP through a stream `BytesInput`),
`initDictionary` is also where the page is decompressed, so it does more than
parse the bytes. A bug in Spark or parquet-java, such as an NPE or a
`ClassCastException`, would also show up as `could not decode the dictionary`,
which points users at the file instead of the reader. Could we catch only the
exceptions the tests cover (`IndexOutOfBoundsException`,
`UnsupportedOperationException`), or at least reword the comment?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [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]
