[SPARK-59854][SQL] Report the column in decoding errors of the vectorized Parquet reader by viirya · Pull Request #59542 · apache/spark

viirya · 2026-10-16T07:25:17Z

What changes were proposed in this pull request?

This is a follow-up of SPARK-59832 (#59117). That change added the page and the column to errors from initializing a data page in VectorizedColumnReader. This PR adds the column to the decoding errors that still do not report it:

Decoding a batch. readBatch wraps a ParquetDecodingException thrown while decoding values or levels in a ParquetDecodingException that names the page and the column: could not decode page <page> in col <column>: <original message>. This covers the ParquetDecodingExceptions of the RLE, PLAIN, DELTA_BINARY_PACKED, DELTA_LENGTH_BYTE_ARRAY, DELTA_BYTE_ARRAY and BYTE_STREAM_SPLIT readers. Only ParquetDecodingException is caught, so other exceptions keep their types: SchemaColumnConvertNotSupportedException and SparkUpgradeException, which FileDataSourceV2.attachFilePath dispatches on, and the RuntimeException of a vector that cannot grow, which ParquetStorageFilter.rethrowIfMustPropagate dispatches on. The reader keeps only the description of the current page, not the page, because for codecs that decompress lazily the page holds the decompressor stream.
Reading a data page. parquet-java decompresses ZSTD pages eagerly, inside PageReader.readPage(), which was outside every catch. Its ParquetDecodingException (could not decompress page) now gets the column. Only ParquetDecodingException is caught here, because readPage() also decrypts pages, and decryption errors are not decoding errors.
Reading the dictionary. readDictionaryPage() and Encoding.initDictionary now share one try, so a dictionary page that cannot be read fails with could not decode the dictionary for <column> whether it is decompressed eagerly (ZSTD) or when initDictionary reads its bytes (SNAPPY, GZIP). Besides IOException and ParquetDecodingException, it wraps the three runtime exceptions a corrupted dictionary page or page header causes there: 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. Other runtime exceptions are not wrapped, so a bug in the reader is not reported as a corrupt dictionary.
DELTA_BINARY_PACKED. VectorizedDeltaBinaryPackedReader.initFromPage checked the page value count and the mini block size with checkArgument, which throws an IllegalArgumentException. These checks now throw a ParquetDecodingException, like the block size checks next to them, and the mini block size is checked with integer arithmetic. A mini block bit width above 64, which failed with an ArrayIndexOutOfBoundsException, is now rejected with a ParquetDecodingException when the mini block is unpacked. The existing page wrappers and the new batch wrapper then add the page and the column.
An empty column chunk. The constructor's totalValueCount == 0 error, which a corrupt footer can cause, now names the column as well.

Each wrapper keeps the original message and adds a single layer to the cause chain.

These cases are not covered:

An out-of-range dictionary id fails with an ArrayIndexOutOfBoundsException from Dictionary.decodeTo*, either in readBatch or, with lazy dictionary decoding, after readBatch returns. Reporting the column for it needs a check of every id against the dictionary size, so it is left to SPARK-60317.
A dictionary page whose declared size or entry lengths do not fit in the page can end in an OutOfMemoryError or, for an uncompressed chunk, return bytes of the next page. That is left to SPARK-60316.
A corrupt LZ4_RAW or LZ4 page fails with a RuntimeException of the codec library, aircompressor's MalformedInputException or lz4-java's LZ4Exception, which parquet-java does not turn into an IOException.
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.
With a pushed filter, parquet-java's DictionaryFilter reads dictionary pages while it filters row groups, before any VectorizedColumnReader exists, so a dictionary page that cannot be read there fails without the column.

Why are the changes needed?

Without the column, a decoding error from a corrupted file does not say which column is affected. For example, a truncated RLE run fails with only Corrupted RLE data: reading past the end of the encoded values. When VectorizedColumnReader reads a SNAPPY or GZIP data page, a decompression error already reports the column, because the page is decompressed lazily inside the existing catch. ZSTD pages did not report it.

Does this PR introduce any user-facing change?

Yes, in error messages only. The decoding errors above now name the column, and the page for errors while decoding a batch. The corrupt dictionaries above now fail with a ParquetDecodingException that names the column instead of an IndexOutOfBoundsException, NegativeArraySizeException or UnsupportedOperationException, which is kept as the cause. The DELTA_BINARY_PACKED checks now 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, and the page and the column are added to it. A ZSTD dictionary page that cannot be decompressed now fails with could not decode the dictionary for <column>. Valid files are not affected.

How was this patch tested?

New tests:

ParquetVectorizedSuite:
- A truncated RLE dictionary id section and truncated PLAIN values report the page and the column, with the original ParquetDecodingException as the cause.
- A truncated BINARY dictionary, a negative dictionary size and a dictionary page with a non-dictionary encoding report the column.
- A mini block bit width of 65 in a DELTA_BINARY_PACKED page reports the page and the column, like the truncated pages above, and an invalid DELTA_BINARY_PACKED header reports the column.
- A column chunk without values reports the column.
ParquetDeltaEncodingSuite: mini blocks that are not a multiple of 8 values and a page with no values are rejected with a ParquetDecodingException.
ParquetIOSuite: a ZSTD data page and a ZSTD dictionary page that cannot be decompressed report the column. The test shares a helper with the SPARK-59832 tests, including their timeout.

All new tests fail without this change. Existing suites also pass: VectorizedRleValuesReaderSuite, ParquetEncodingSuite, ParquetVectorizedSuite, ParquetColumnIndexSuite, ParquetIOSuite, ParquetV1QuerySuite, ParquetV2QuerySuite, ParquetDeltaEncodingSuite, ParquetDeltaByteArrayEncodingSuite, ParquetDeltaLengthByteArrayEncodingSuite, VectorizedPlainValuesReaderSuite, ParquetVectorUpdaterSuite and ParquetCompressionCodecPrecedenceSuite.

Was this patch authored or co-authored using generative AI tooling?

Generated-by: Claude Code (Claude Opus 5.5)

This pull request and its description were written by Isaac.

…ized Parquet reader Add the column to decoding errors of the vectorized Parquet reader that did not report it: - ParquetDecodingExceptions thrown while decoding a batch in VectorizedColumnReader.readBatch. - ParquetDecodingExceptions thrown by PageReader.readPage and readDictionaryPage, which decompress ZSTD pages eagerly. - RuntimeExceptions thrown by Encoding.initDictionary, e.g. for a truncated BINARY dictionary. - Invalid DELTA_BINARY_PACKED headers, which now fail with a ParquetDecodingException instead of an IllegalArgumentException.

dongjoon-hyun

Thank you for the fix, @viirya. I reviewed the latest commit, 20ce444, and left 8 inline comments. Here is a summary, ordered by priority.

Correctness

VectorizedColumnReader L354: out-of-range dictionary ids still fail with an ArrayIndexOutOfBoundsException that names no page or column, because the new wrapper catches only ParquetDecodingException.
VectorizedColumnReader L172: the new catch (RuntimeException e) reports any runtime error of initDictionary, including a bug, as a corrupt dictionary, and its comment that initDictionary only parses the bytes is not exact.

Memory and performance

VectorizedColumnReader L373: currentPage keeps the last page reachable after the column chunk is exhausted.

Tests

ParquetIOSuite L2581: the ZSTD test copies the scaffolding of readCorruptedRleSection without its 60-second failAfter guard.
ParquetIOSuite L2607: the test pins the internal messages of parquet-java and its eager ZSTD decompression.
ParquetIOSuite L2589: the deprecated ParquetFileReader.readFooter is used instead of ParquetTest.readFooter.

Cleanup

VectorizedColumnReader L287: about 70 unchanged lines are re-indented; a helper method would keep the diff and git blame small.
VectorizedDeltaBinaryPackedReader L110: the new message prints a fractional double such as 42.666666666666664.

Items 1 and 2 are the most important, since they decide which corruptions report the column and which errors are blamed on the file.

dongjoon-hyun · 2026-10-16T12:37:04Z

+              column, valuesReader, updater);
+          }
        }
+      } catch (ParquetDecodingException e) {


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?

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?

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.

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.

dongjoon-hyun · 2026-10-16T12:37:04Z

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


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?

Good point, and the comment was wrong: for SNAPPY and GZIP the dictionary is decompressed when initDictionary reads the bytes. The catch now takes only what a corrupted page or page header causes there: IndexOutOfBoundsException (a truncated BINARY dictionary), NegativeArraySizeException (a negative dictionary size in the page header) and UnsupportedOperationException (an encoding that has no dictionary). The test covers all three, and the comment now says that decompression can happen here. Decompression errors there are IOExceptions, which the existing catch already handles.

dongjoon-hyun · 2026-10-16T12:37:04Z

    if (page == null) {
      return -1;
    }
+    this.currentPage = page;


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?

Done, currentPage is now set before the null check, so it is cleared at the end of the column chunk.

dongjoon-hyun · 2026-10-16T12:37:04Z

+      (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 =>


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?

Done. readCorruptedFile now takes the writer options and the corruption step; readCorruptedRleSection and the ZSTD test both use it, with the 60-second failAfter guard.

dongjoon-hyun · 2026-10-16T12:37:04Z

+          val columnErrors =
+            causes(e).filter(c => Option(c.getMessage).exists(_.contains("optional int32 a")))
+          assert(columnErrors.size == 1, e)
+          assert(columnErrors.head.getMessage ==


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?

Done, the test now only checks that the column is reported once, on a ParquetDecodingException.

dongjoon-hyun · 2026-10-16T12:37:04Z

+          .option("compression", "zstd")
+          .parquet(path)
+        val file = dir.listFiles().filter(_.getName.endsWith(".parquet")).head
+        val footer = ParquetFileReader.readFooter(


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?

ParquetTest.readFooter reads the _metadata summary file of a directory (new Path(path, PARQUET_METADATA_FILE)), which is why the tests at L1353 and L1527 write summary files first, so it would not find the footer of this data file. I switched to the non-deprecated ParquetFooterReader.readFooter(HadoopInputFile.fromPath(...), NO_FILTER) instead.

dongjoon-hyun · 2026-10-16T12:37:04Z

+      // 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 {


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?

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.

dongjoon-hyun · 2026-10-16T12:37:04Z

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


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?

Done. The check now uses integer arithmetic and reports the mini block count and the block size, like the checks above.

- Wrap only the IndexOutOfBoundsException, NegativeArraySizeException and UnsupportedOperationException of initDictionary, and fix its comment. - Clear currentPage at the end of the column chunk. - Check the DELTA_BINARY_PACKED mini block size with integer arithmetic. - Share the corrupted file helper of ParquetIOSuite, with its timeout, in the ZSTD test, and check only that the column is reported once. Co-authored-by: Isaac <no-reply@databricks.com>

dongjoon-hyun

Thank you for the update, @viirya. Commit 68a4b4e handles items 2, 4, 5 and 8 of my first review, and item 3 for repeated columns (see 1 below). For item 6, you were right not to use ParquetTest.readFooter: it reads the _metadata summary file, so my suggestion would not have worked, and ParquetFooterReader.readFooter is the right call.

I reviewed the whole PR again at this head and left 14 inline comments, plus a reply in the thread of item 1 of my first review, which has no answer yet. Items 3 to 7 below are older than this PR, but they sit next to the new catches. Here is a summary, ordered by priority.

Memory and merge state

VectorizedColumnReader.java L375: currentPage keeps each column's SNAPPY or GZIP decompressor stream, with its 64 KiB buffer, alive for the whole scan.
ParquetVectorizedSuite.scala L554: the PR conflicts with SPARK-60262 on master, so the CI run of 68a4b4e stopped at the merge step and no test ran on the second commit.

Corrupt dictionaries

VectorizedColumnReader.java L172: a huge positive dictionary size still ends in an OutOfMemoryError that kills the executor.
VectorizedColumnReader.java L177: some truncated dictionaries pass initDictionary, and an uncompressed chunk can then return bytes of the next page.

Errors that still lack the column

VectorizedColumnReader.java L371: corrupt LZ4_RAW pages, and eagerly decompressed pages with a negative uncompressed size.
VectorizedDeltaBinaryPackedReader.java L91: a DELTA_BINARY_PACKED mini block bit width above 64.
VectorizedColumnReader.java L359 (reply in the earlier thread): dictionary ids beyond the end of the dictionary.

Description

VectorizedColumnReader.java L174: the description still describes the first commit, and its statement about SNAPPY and GZIP does not hold for filtered reads.

Tests

ParquetIOSuite.scala L2606: the "reported once, on a ParquetDecodingException" check that I suggested still depends on eager ZSTD decompression.
ParquetVectorizedSuite.scala L597: the test that says it reports the page never checks the page.
VectorizedDeltaBinaryPackedReader.java L109: the multiple-of-8 half of the new check has no test.

Cleanup

VectorizedColumnReader.java L159: the same dictionary page that cannot be decompressed fails with a different type and message per codec, and item 7 of my first review is still open.
ParquetVectorizedSuite.scala L558: readCorruptedPage copies the SPARK-59832 test above it, and sits between the tests.
VectorizedDeltaBinaryPackedReader.java L108: the new check makes the miniBlockNumInABlock > blockSizeInValues arm of the previous check redundant.
ParquetIOSuite.scala L2593: the ZSTD test reads the footer only to find offset 4.

Items 1 and 2 matter most: 1 is a memory regression with the default codec, and 2 blocks the merge. If items 3 to 7 are out of scope, separate JIRAs are fine with me, but it would help to say so in the description.

dongjoon-hyun · 2026-10-16T19:24:44Z

+      throw new ParquetDecodingException(
+        "could not read the next page in col " + descriptor + ": " + e.getMessage(), e);
+    }
+    this.currentPage = page;


currentPage keeps each column's decompressor stream alive, about 66 KiB per column with the default SNAPPY codec.

For the codecs that parquet-java decompresses lazily (SNAPPY, GZIP and Hadoop LZ4), the DataPage from pageReader.readPage() holds a StreamBytesInput, which holds the codec's DecompressorStream and its io.file.buffer.size buffer. Spark sets that size from spark.buffer.size, 64 KiB, and Hadoop LZ4 uses a 256 KiB buffer. Before this PR, the page became garbage as soon as readPage() returned, because the value readers keep only the stream over the array that page.getBytes().toInputStream() copied. Now every VectorizedColumnReader keeps its current page. A scan of 1,000 leaf columns with 8 tasks per executor keeps about 500 MiB more heap live, and 100 columns keep about 50 MiB.

The null assignment here does not help flat columns. ParquetReadState sets lastListCompleted for them, so readPage() is not called again after the last page, and that page stays reachable until the next row group.

The message only needs what DataPageV1.toString() and DataPageV2.toString() print, a few sizes and counts. Could we store String.valueOf(page), or those values or a page ordinal, instead of the page?

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.

dongjoon-hyun · 2026-10-16T19:24:44Z

    }
  }

+  /**


This spot now conflicts with master, so the second commit has not been tested by CI.

SPARK-60262 (f095057) added a test right after the SPARK-59832 test, where these helpers and tests go. GitHub reports the PR as not mergeable. The Check changes job of 68a4b4e failed at its merge step with CONFLICT (content): Merge conflict in .../ParquetVectorizedSuite.scala, so every build and test job was skipped. Only 20ce444 has a green run, so the narrowed multi-catch, the NegativeArraySizeException case, the integer mini block check and the rewritten ZSTD test have not run in CI yet.

VectorizedColumnReader.java merges cleanly, and SPARK-60262's ran out of pages exception is thrown before the new try, so keeping both tests should be enough. Could you rebase and rerun CI?

Merged master and kept both tests. The conflict was only in ParquetVectorizedSuite, and the SPARK-60262 test passes together with the new ones.

dongjoon-hyun · 2026-10-16T19:24:44Z

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


A huge positive dictionary size still ends in an OutOfMemoryError, which kills the executor.

This arm turns a negative num_values into a ParquetDecodingException. However, PlainValuesDictionary allocates new int[size], new Binary[size] and so on without checking the size against the page. Take an INT32 dictionary page header with num_values = 0x7FFFFFF0 and an executor heap under 8 GiB:

initDictionary throws OutOfMemoryError: Java heap space, which none of these catches match.

FileDataSourceV2.attachFilePath rethrows it without the file path, and the executor exits through SparkUncaughtExceptionHandler.

Every retry kills another executor, and spark.sql.files.ignoreCorruptFiles cannot skip the file.

FIXED_LEN_BYTE_ARRAY and INT96 dictionaries fail on any realistic heap, because they create one Binary per entry.

This is older than the PR, but it is the other sign of the case this arm handles. A check before initDictionary, such as size < 0 || (long) size * minEntryWidth > pageSize, would cover both signs with a ParquetDecodingException that names the column. Could we add it here, or file a separate JIRA?

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.

dongjoon-hyun · 2026-10-16T19:24:45Z

+        // 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


Some truncated dictionaries pass initDictionary, and an uncompressed chunk can then return bytes of the next page with no error.

This arm fires only when the truncation cuts a BINARY length prefix. parquet-java's PlainBinaryDictionary does not check an entry's length against the page, and its FIXED_LEN_BYTE_ARRAY and INT96 branch reads no bytes at all. For example, take a PLAIN string dictionary with two entries, 02 00 00 00 'a' 'b' 05 00 00 00 'c' 'd', in an UNCOMPRESSED chunk:

initDictionary succeeds, although the second entry claims 5 bytes and only 2 are left.

The dictionary buffer is a slice of the buffer that holds the column chunk, and Binary.getBytes copies from the backing array at arrayOffset() + offset, so the second value reads as cd plus three bytes of the next page header.

With SNAPPY or ZSTD the same entry throws an ArrayIndexOutOfBoundsException at decode time, without the column, and a truncated FIXED_LEN_BYTE_ARRAY decimal or INT96 dictionary throws an IllegalArgumentException inside readBatch.

This is how parquet-java behaved before this PR, but this comment and the description suggest that truncated BINARY dictionaries are covered. Could we validate the entries against the page (every offset + len within the page for BINARY, size * length for FIXED_LEN_BYTE_ARRAY and INT96), or narrow the comment and leave the silent case to a separate JIRA?

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.

dongjoon-hyun · 2026-10-16T19:24:45Z

+    try {
+      // This decompresses the page eagerly for some codecs, e.g. ZSTD.
+      page = pageReader.readPage();
+    } catch (ParquetDecodingException e) {


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?

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.

dongjoon-hyun · 2026-10-16T19:24:45Z

-        "miniBlockSize must be multiple of 8, but it's " + miniSize);
-    this.miniBlockSizeInValues = (int) miniSize;
+    if (blockSizeInValues % miniBlockNumInABlock != 0 ||
+        (blockSizeInValues / miniBlockNumInABlock) % 8 != 0) {


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?

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.

dongjoon-hyun · 2026-10-16T19:24:45Z

+    try {
+      // This decompresses the page eagerly for some codecs, e.g. ZSTD.
+      dictionaryPage = pageReader.readDictionaryPage();
+    } catch (ParquetDecodingException e) {


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.

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.

dongjoon-hyun · 2026-10-16T19:24:45Z

+   * Reads a row group of one page of `valueCount` values of the column `a` in `schema`, after
+   * an optional dictionary page, and returns the error of reading it.
+   */
+  private def readCorruptedPage(


Nit: readCorruptedPage is a parameterized copy of the SPARK-59832 test right above it (L527-L552). That test keeps its own copy of the same MemPageStore, reader, initialize, intercept and close code, and the same one-value definition levels with the same comment (L536-L538 and L616-L619). It is exactly readCorruptedPage("message root { optional int32 a; }", Some(<its dictionary page>), Array[Byte](2, 0, 0, 0, 2, 1, 2, 2, 0), 1, Encoding.RLE_DICTIONARY) plus its asserts. Also, schema looks free-form, but L565-L566 hard-code the column a with max definition level 1.

The helper and fourNonNullValues also sit between the tests, while this suite keeps its vals at the top and its helpers after the tests (from truncateTypeTest on), as AGENTS.md asks ("follow the sectioning the file already uses"). Since the rebase touches this exact spot, could we move them and reuse the helper in the SPARK-59832 test?

Done. The SPARK-59832 test now uses readCorruptedPage, which takes the column type instead of a schema, so it no longer looks free-form. The helper sits with the other helpers after writeDataPage, and the definition level vals are at the top of the suite.

dongjoon-hyun · 2026-10-16T19:24:45Z

-    JavaUtils.checkArgument(miniSize % 8 == 0,
-        "miniBlockSize must be multiple of 8, but it's " + miniSize);
-    this.miniBlockSizeInValues = (int) miniSize;
+    if (blockSizeInValues % miniBlockNumInABlock != 0 ||


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.

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

dongjoon-hyun · 2026-10-16T19:24:45Z

+        ParquetOutputFormat.ENABLE_DICTIONARY -> dictionary.toString,
+        "compression" -> "zstd")
+      val e = readCorruptedFile("cast(id % 10 as int) as a", options) { (file, bytes) =>
+        val footer = ParquetFooterReader.readFooter(


Nit: the footer is read only to get chunk.getStartingPos, which is always 4 for this local single-column file, and hasDictionaryPage. It is also the only reason readCorruptedFile passes a Path (the other caller ignores it at L2537), and it adds two imports. Could we start the stream at ParquetFileWriter.MAGIC.length and assert the type of the parsed header instead (DICTIONARY_PAGE or DATA_PAGE)? That would check the page that is actually corrupted.

Done. The test parses the page header at ParquetFileWriter.MAGIC.length and asserts its type, DATA_PAGE or DICTIONARY_PAGE, and readCorruptedFile no longer passes the path.

Co-authored-by: Isaac <no-reply@databricks.com>

- Keep only the description of the current page, so that the page and its decompressor stream are not kept alive. - Read the dictionary page in the same try as initDictionary, and narrow its comment. - Reject a DELTA_BINARY_PACKED mini block bit width above 64, and drop a redundant check. - Test the page in the batch error message, the mini block size checks and the bit width check; share the test helpers and simplify the ZSTD test. Co-authored-by: Isaac <no-reply@databricks.com>

dongjoon-hyun

Thank you for addressing items 1-15 of my second review, @viirya. I reviewed the latest commit, 3568ed4, and left 7 inline comments, continuing the numbering from items 1-15. Here is a summary, ordered by priority.

Before merging: the PR description

VectorizedColumnReader.java L167: a corrupt Hadoop LZ4 page still fails without the column, and the "not covered" list leaves LZ4 out. This is the third bullet of item 5, which is still open.
VectorizedColumnReader.java L380: a negative uncompressed page size also fails without the column for SNAPPY, GZIP and LZ4 data pages, which my item 5 missed.
VectorizedDeltaBinaryPackedReader.java L482: the original exception is kept as the cause only for the dictionary errors.

Optional

VectorizedColumnReader.java L181: totalValueCount == 0 is the only error in the constructor that still does not name the column.
ParquetVectorizedSuite.scala L627: the bit width case fits as a third row of the batch test.
ParquetVectorizedSuite.scala L968: the blank lines around the moved readCorruptedPage.
ParquetIOSuite.scala L2593: the stream length is 4 bytes more than what is left after the offset.

+1, LGTM. CI is green on 3568ed4. Items 16-18 only need the PR description, which becomes the commit message, so could you update it before merging? Items 19-22 are optional.

dongjoon-hyun · 2026-10-19T14:35:37Z

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


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

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

dongjoon-hyun · 2026-10-19T14:35:37Z

    if (page == null) {
      return -1;
    }
+    this.currentPage = page.toString();


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

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.

dongjoon-hyun · 2026-10-19T14:35:37Z

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


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

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.

dongjoon-hyun · 2026-10-19T14:35:37Z

+    this.dictionary = dict;
+    this.isCurrentPageDictionaryEncoded = dict != null;
    if (pageReader.getTotalValueCount() == 0) {
      throw new IOException("totalValueCount == 0");


Nit: this is now the only error in the constructor that does not name the column. It is older than this PR, but a corrupt footer that sets num_values = 0 for one column chunk of a non-empty row group reaches it: parquet-java then reads no page of that chunk, and the read fails with only totalValueCount == 0. Could we add the column here as well, e.g. throw new IOException("totalValueCount == 0 in col " + descriptor);?

Done, it now reads totalValueCount == 0 in col <column>, and a new test reads a column chunk without pages.

dongjoon-hyun · 2026-10-19T14:35:37Z

+    // Two non-null values, so the second one unpacks the first mini block. The header has a
+    // block of 128 values in 4 mini blocks, 2 values and the first value 0. The block has the
+    // min delta 0 and the mini block bit widths 65, 0, 0 and 0.
+    val e2 = readCorruptedPage("int32", None,


Nit: this case repeats the setup and three of the four assertions of "a decoding error in a batch reports the page and the column" (L576-L591), without its check of the cause's message (L589). With 4 values it fits there as a third row, since the second value still unpacks the first mini block:

("Invalid DELTA_BINARY_PACKED mini block bit width: 65 (must be at most 64)", None, Array[Byte](0x80.toByte, 1, 4, 4, 0, 0, 65, 0, 0, 0), Encoding.DELTA_BINARY_PACKED)

Could we move it there, so that this test keeps only the header case?

Done, the bit width case is now the third row of the batch test, which also checks the cause's message, and the DELTA_BINARY_PACKED test keeps only the header case.

dongjoon-hyun · 2026-10-19T14:35:37Z

+    } finally {
+      recordReader.close()
+    }
+  }


Nit: moving readCorruptedPage here left two blank lines above its Scaladoc (L937-L938) and none between its closing brace and checkAnswer (L968-L969). Could we move the blank line from L938 to after L968? Also, ParquetIOSuite.scala:2603 adds a blank line before the closing brace of the class, which the base does not have.

Done, fixed both.

dongjoon-hyun · 2026-10-19T14:35:37Z

+          "compression" -> "zstd")
+        val e = readCorruptedFile("cast(id % 10 as int) as a", options) { bytes =>
+          // The first page of the only column chunk starts right after the magic number.
+          val in = new ByteArrayInputStream(bytes, ParquetFileWriter.MAGIC.length, bytes.length)


Nit: bytes.length is 4 bytes more than what is left after the offset. It works because ByteArrayInputStream caps count at buf.length, which pageStart at L2596 also relies on. Could we pass bytes.length - ParquetFileWriter.MAGIC.length, like MapOutputTracker.scala:1780?

Done, it now passes bytes.length - start.

- Name the column in the totalValueCount == 0 error of VectorizedColumnReader. - Move the DELTA_BINARY_PACKED bit width case into the batch test, and fix blank lines and a stream length in the tests. Co-authored-by: Isaac <no-reply@databricks.com>

dongjoon-hyun reviewed Oct 16, 2026

View reviewed changes

viirya and others added 2 commits October 17, 2026 12:27

Merge remote-tracking branch 'upstream/master' into SPARK-59854

862d115

Co-authored-by: Isaac <no-reply@databricks.com>

dongjoon-hyun approved these changes Oct 19, 2026

View reviewed changes

                   }
                 }
+                /**

Conversation

viirya commented Oct 16, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

Was this patch authored or co-authored using generative AI tooling?

Uh oh!

dongjoon-hyun left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

dongjoon-hyun left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

viirya commented Oct 16, 2026 •

edited

Loading