Skip to content

[SPARK-59854][SQL] Report the column in decoding errors of the vectorized Parquet reader #59542

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
viirya wants to merge 5 commits into
base: master
Choose a base branch
from
+319 −128
Open

[SPARK-59854][SQL] Report the column in decoding errors of the vectorized Parquet reader #59542

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
20ce444
[SPARK-59854][SQL] Report the column in decoding errors of the vector…
viirya Oct 16, 2026
68a4b4e
Address review comments
viirya Oct 16, 2026
862d115
Merge remote-tracking branch 'upstream/master' into SPARK-59854
viirya Oct 17, 2026
3568ed4
Address the second round of review comments
viirya Oct 17, 2026
20fe1ce
Address the third round of review comments
viirya Oct 19, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
185 changes: 109 additions & 76 deletions .../main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -98,6 +98,12 @@ public class VectorizedColumnReader {
*/
private long pageFirstRowIndex;

/**
* The data page the values are being decoded from, for error messages. Only its description is
* kept: for codecs that decompress lazily, the page holds the decompressor stream.
*/
private String currentPage;

private final PageReader pageReader;
private final ColumnDescriptor descriptor;
private final LogicalTypeAnnotation logicalTypeAnnotation;
Expand Down Expand Up @@ -146,23 +152,33 @@ 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) {

@dongjoon-hyun dongjoon-hyun Oct 16, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Nit: because this try is separate from the initDictionary one below, a dictionary page that cannot be decompressed fails differently depending on the codec. With ZSTD it is a ParquetDecodingException, "could not read the dictionary page for ...". With SNAPPY or GZIP, which decompress inside initDictionary, it is an IOException, "could not decode the dictionary for ..." (L167-L168). A log search for one misses the other. Moving pageReader.readDictionaryPage() into the try at L164 would give one message shape, and the NegativeArraySizeException arm would then also cover the eager case.

Item 7 of my first review is still open as well. If the decode block of readBatch (L293-L358) moved into a private method and only that call were wrapped, git blame would keep its history: today about 61 of the changed lines differ only in indentation (git diff -w shows 42+/3- instead of 103+/64-). Either way the wrapper should keep catching only ParquetDecodingException, since ParquetStorageFilter.rethrowIfMustPropagate dispatches on the exact RuntimeException class.

@viirya viirya Oct 17, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Done for the first part: readDictionaryPage() is now in the same try as initDictionary, so a dictionary page that cannot be read fails with could not decode the dictionary for ... for every codec.

For the helper method, I tried it and measured git blame on the decode block. Plain git blame attributes 64 of its lines to this PR in both versions, -w 8 in both, and -w -M 7 with the try and 6 with the method. Moving the block into a method body re-indents it as well (to 4 spaces, not back to the original 6), so it does not keep the history. So I kept the try. Agreed that the wrapper must keep catching only ParquetDecodingException, so that a vector growth failure, which rethrowIfMustPropagate checks by its exact class, is not wrapped.

throw new ParquetDecodingException(
"could not decode the dictionary for " + descriptor + ": " + e.getMessage(), e);
} catch (IndexOutOfBoundsException | NegativeArraySizeException |

@dongjoon-hyun dongjoon-hyun Oct 19, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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"?

@viirya viirya Oct 19, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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".

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");
throw new IOException("totalValueCount == 0 in col " + descriptor);
}
assert "LEGACY".equals(datetimeRebaseMode) || "EXCEPTION".equals(datetimeRebaseMode) ||
"CORRECTED".equals(datetimeRebaseMode);
Expand Down Expand Up @@ -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 {

@dongjoon-hyun dongjoon-hyun Oct 16, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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?

@viirya viirya Oct 16, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

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) {

@dongjoon-hyun dongjoon-hyun Oct 16, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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?

@dongjoon-hyun dongjoon-hyun Oct 16, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

At 68a4b4e this is unchanged. With a 2-entry INT32 dictionary and an RLE run of id 5:

  • Read as LongType (eager), decodeDictionaryIds throws ArrayIndexOutOfBoundsException: Index 5 out of bounds for length 2 inside the new try, and it is not wrapped because it is not a ParquetDecodingException.
  • Read as IntegerType (lazy), readBatch succeeds, and the same exception is thrown later from ParquetDictionary.decodeToInt in the consumer. That is outside FileScanRDD's handling too, so the error names neither the file nor the column.

I understand the description says only ParquetDecodingException is wrapped, and this was also left out of SPARK-59832. Since the RLE reader does not know the dictionary, a check needs some thought about its per-value cost. Could you either handle it here, or list it in the description as a known gap with a separate JIRA?

@viirya viirya Oct 16, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Right, an out-of-range id fails with an ArrayIndexOutOfBoundsException from Dictionary.decodeTo*, and with lazy decoding only after readBatch returns. Covering both needs a check of every id against getMaxId(), which is per-value work on the hot path (the lazy path does not look at the ids at all today), so I'd rather leave it to a follow-up. I narrowed the description: only ParquetDecodingExceptions are covered, and out-of-range ids are listed as not covered.

@viirya viirya Oct 17, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

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) {

@dongjoon-hyun dongjoon-hyun Oct 16, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Minor: this catches only ParquetDecodingException, which ColumnChunkPageReader creates from an IOException of the codec, so other eagerly decompressed pages still fail without the column.

  • parquet-java decompresses LZ4_RAW eagerly in the same branch as ZSTD (codec instanceof ZstandardCodec || codec instanceof Lz4RawCodec in CodecFactory), but aircompressor reports corrupt input with MalformedInputException, a RuntimeException. The new ZSTD test with "compression" -> "lz4_raw" gives columnErrors.size == 0.
  • A negative uncompressed_page_size in a page header makes BytesInput.copy throw a NegativeArraySizeException inside readPage() or readDictionaryPage() for ZSTD and LZ4_RAW, which is not wrapped. For a SNAPPY dictionary page the same exception is thrown inside initDictionary, where the new arm at L172 wraps it, so whether the column is reported depends on the codec.
  • Hadoop LZ4's LZ4Exception in a dictionary page also passes the multi-catch at L172.

Could we also wrap these here and at L159, except ParquetCryptoRuntimeException, or narrow the comment and the description to ZSTD?

@viirya viirya Oct 17, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

throw new ParquetDecodingException(
"could not read the next page in col " + descriptor + ": " + e.getMessage(), e);
}
if (page == null) {
return -1;
}
this.currentPage = page.toString();

@dongjoon-hyun dongjoon-hyun Oct 19, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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"?

@viirya viirya Oct 19, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

this.pageFirstRowIndex = page.getFirstRowIndex().orElse(0L);

return page.accept(new DataPage.Visitor<Integer>() {
Expand Down
30 changes: 20 additions & 10 deletions ...org/apache/spark/sql/execution/datasources/parquet/VectorizedDeltaBinaryPackedReader.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,6 @@
import org.apache.parquet.column.values.bitpacking.Packer;
import org.apache.parquet.io.ParquetDecodingException;

import org.apache.spark.network.util.JavaUtils;
import org.apache.spark.sql.catalyst.util.RebaseDateTime;
import org.apache.spark.sql.execution.datasources.DataSourceUtils;
import org.apache.spark.sql.execution.vectorized.WritableColumnVector;
Expand Down Expand Up @@ -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) {

@dongjoon-hyun dongjoon-hyun Oct 16, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Minor: like the header fields checked here, the mini block bit widths come from the file, but they are never range-checked, so a corrupt width still fails without the page or the column. readBitWidthsForMiniBlocks (L496-L503) reads each width as 0 to 255, and unpackMiniBlock passes it to Packer.LITTLE_ENDIAN.newBytePackerForLong (L480-L481), which indexes a table of 65 packers. With the bit widths 65 0 0 0 in a block header, reading the second INT32 value throws ArrayIndexOutOfBoundsException: Index 65 out of bounds for length 65. That is not a ParquetDecodingException, so the new readBatch wrapper does not add the column. For DELTA_LENGTH_BYTE_ARRAY and DELTA_BYTE_ARRAY, whose lengths are decoded in initFromPage, the readPageV1 and readPageV2 wrappers do not either.

The spec allows arbitrary widths for the unused mini blocks of the last block, so the check has to be where a width is used: if (bitWidth > 64) throw new ParquetDecodingException(...) in unpackMiniBlock, like the width > 32 check in VectorizedRleValuesReader. Since this PR already hardens this reader's header, could we add it here?

@viirya viirya Oct 17, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

throw new ParquetDecodingException("Invalid DELTA_BINARY_PACKED page: it must have at "
+ "least one value, but it has " + valueCount);
}
this.in = in;
// Read the header
this.blockSizeInValues = BytesUtils.readUnsignedVarInt(in);
Expand All @@ -100,14 +101,17 @@ public void initFromPage(int valueCount, ByteBufferInputStream in) throws IOExce
throw new ParquetDecodingException("Invalid DELTA_BINARY_PACKED block size in values: "
+ blockSizeInValues + " (must be in [1, " + MAX_BLOCK_SIZE_IN_VALUES + "])");
}
if (miniBlockNumInABlock <= 0 || miniBlockNumInABlock > blockSizeInValues) {
if (miniBlockNumInABlock <= 0) {
throw new ParquetDecodingException("Invalid DELTA_BINARY_PACKED mini block count: "
+ 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);
this.miniBlockSizeInValues = (int) miniSize;
if (blockSizeInValues % miniBlockNumInABlock != 0 ||

@dongjoon-hyun dongjoon-hyun Oct 16, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Nit: this check makes the miniBlockNumInABlock > blockSizeInValues arm at L104 redundant. L100 guarantees blockSizeInValues >= 1, so m > b implies b % m == b != 0, and both checks throw "Invalid DELTA_BINARY_PACKED mini block count: ...". Could we drop that arm from L104? miniBlockNumInABlock <= 0 has to stay, because it guards the division here and rejects negative counts such as -16 (128 % -16 == 0). The contains("mini block count") assertion in ParquetDeltaEncodingSuite still matches.

@viirya viirya Oct 17, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Done, the miniBlockNumInABlock > blockSizeInValues arm is gone and <= 0 stays. The existing craftHeader(128, 256, 1) case still gets the "mini block count" message.

(blockSizeInValues / miniBlockNumInABlock) % 8 != 0) {

@dongjoon-hyun dongjoon-hyun Oct 16, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Minor: this half of the condition has no test. The only new test uses a block of 128 values in 3 mini blocks, which already fails 128 % 3 != 0. If (blockSizeInValues / miniBlockNumInABlock) % 8 != 0 were dropped, every test would still pass. A header with 128 values in 32 mini blocks would then allocate unpackedValuesBuffer = new long[4] at L122 and fail in unpack8Values with an ArrayIndexOutOfBoundsException, without the column.

ParquetDeltaEncodingSuite already unit-tests these header checks in "reject invalid DELTA_BINARY_PACKED page header", with a craftHeader helper, for both INT32 and INT64. Could we add craftHeader(128, 3, 1), craftHeader(128, 32, 1) and a zero value count there, and drop the direct initFromPage(0, ...) check from ParquetVectorizedSuite (L636-L641) together with its ByteBufferInputStream import?

@viirya viirya Oct 17, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Done. ParquetDeltaEncodingSuite now checks 128 values in 3 and in 32 mini blocks and a page with no values, and I dropped the direct initFromPage(0, ...) check and its import from ParquetVectorizedSuite.

throw new ParquetDecodingException("Invalid DELTA_BINARY_PACKED mini block count: "
+ miniBlockNumInABlock + " (block size in values: " + blockSizeInValues
+ ", the mini block size must be a multiple of 8)");
}
this.miniBlockSizeInValues = blockSizeInValues / miniBlockNumInABlock;
// True value count. May be less than valueCount because of nulls
this.totalValueCount = BytesUtils.readUnsignedVarInt(in);
if (totalValueCount < 0) {
Expand Down Expand Up @@ -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) {

@dongjoon-hyun dongjoon-hyun Oct 19, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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"?

@viirya viirya Oct 19, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

throw new ParquetDecodingException("Invalid DELTA_BINARY_PACKED mini block bit width: "
+ bitWidth + " (must be at most 64)");
}
Arrays.fill(this.unpackedValuesBuffer, 0);
BytePackerForLong packer = Packer.LITTLE_ENDIAN.newBytePackerForLong(
bitWidths[currentMiniBlock]);
BytePackerForLong packer = Packer.LITTLE_ENDIAN.newBytePackerForLong(bitWidth);
for (int j = 0; j < miniBlockSizeInValues; j += 8) {
ByteBuffer buffer = in.slice(packer.getBitWidth());
if (buffer.hasArray()) {
Expand Down
5 changes: 5 additions & 0 deletions sql/core/src/test/java/test/org/apache/parquet/column/page/TestDataPage.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,4 +41,9 @@ public Optional<Integer> getIndexRowCount() {
public <T> T accept(Visitor<T> visitor) {
return wrapped.accept(visitor);
}

@Override
public String toString() {
return wrapped.toString();
}
}
19 changes: 19 additions & 0 deletions .../scala/org/apache/spark/sql/execution/datasources/parquet/ParquetDeltaEncodingSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -204,6 +204,25 @@ abstract class ParquetDeltaEncodingSuite[T] extends ParquetCompatibilityTest
reader.initFromPage(100, craftHeader(128, 4, -1))
}
assert(e3.getMessage.contains("total value count"))

// The mini blocks must have a multiple of 8 values: 128 values do not split into 3 mini
// blocks, and split into 32 mini blocks of 4 values.
Seq(3, 32).foreach { miniBlockNum =>
reader = new VectorizedDeltaBinaryPackedReader
val e4 = intercept[ParquetDecodingException] {
reader.initFromPage(100, craftHeader(128, miniBlockNum, 1))
}
assert(e4.getMessage == s"Invalid DELTA_BINARY_PACKED mini block count: $miniBlockNum " +
"(block size in values: 128, the mini block size must be a multiple of 8)")
}

// A page must have at least one value.
reader = new VectorizedDeltaBinaryPackedReader
val e5 = intercept[ParquetDecodingException] {
reader.initFromPage(0, craftHeader(128, 4, 1))
}
assert(e5.getMessage ==
"Invalid DELTA_BINARY_PACKED page: it must have at least one value, but it has 0")
}

test("skip()") {
Expand Down
Loading