viirya commented on code in PR #59542:
URL: https://github.com/apache/spark/pull/59542#discussion_r4298491407
##########
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:
A private method would re-indent the same lines too: the block is at 6
spaces inside the `while`, 8 inside the `try`, and would be at 4 in a method
body. Moving it would also make `git blame` need `-M` rather than just `-w`. So
I kept the `try`; `git blame -w` and the diff with `?w=1` skip the
re-indentation.
##########
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:
Done. The check now uses integer arithmetic and reports the mini block count
and the block size, like the checks above.
--
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]
