You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by ho...@apache.org on 2026/09/03 03:25:34 UTC
(spark) branch branch-4.x updated: [MINOR][SQL] Validate DELTA_BINARY_PACKED page header in the vectorized reader
This is an automated email from the ASF dual-hosted git repository.
asf-gitbox-commits pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 1bcb58b0070d [MINOR][SQL] Validate DELTA_BINARY_PACKED page header in the vectorized reader
1bcb58b0070d is described below
commit 1bcb58b0070d9e3f7aaec10c7060d71bb1fbb38e
Author: Holden Karau <ho...@snowflake.com>
AuthorDate: Tue Sep 1 02:04:03 2026 +0000
[MINOR][SQL] Validate DELTA_BINARY_PACKED page header in the vectorized reader
Check the block size, mini block count, and total value count from the
page header before sizing scratch buffers from them, so a corrupt page
fails with ParquetDecodingException instead of driving oversized
allocations.
Co-authored-by: Cursor <cu...@cursor.com>
(cherry picked from commit 701d5e9768dd0502be2a3c0678de6d3c534135a8)
---
.../parquet/VectorizedDeltaBinaryPackedReader.java | 19 +++++++++++
.../parquet/ParquetDeltaEncodingSuite.scala | 38 ++++++++++++++++++++--
2 files changed, 55 insertions(+), 2 deletions(-)
diff --git a/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedDeltaBinaryPackedReader.java b/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedDeltaBinaryPackedReader.java
index 0912ae534b9b..c04fb49f1add 100644
--- a/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedDeltaBinaryPackedReader.java
+++ b/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedDeltaBinaryPackedReader.java
@@ -47,6 +47,12 @@ import org.apache.spark.sql.execution.vectorized.WritableColumnVector;
*/
public class VectorizedDeltaBinaryPackedReader extends VectorizedReaderBase {
+ // Upper bound accepted for the block size declared in the page header. Parquet writers use
+ // small blocks (parquet-java defaults to 128 values per block), so this generous ceiling only
+ // exists to keep a corrupt header from blowing up the scratch-buffer allocations in
+ // initFromPage: 2^20 values caps those buffers at ~12 MiB.
+ private static final int MAX_BLOCK_SIZE_IN_VALUES = 1 << 20;
+
// header data
private int blockSizeInValues;
private int miniBlockNumInABlock;
@@ -89,12 +95,25 @@ public class VectorizedDeltaBinaryPackedReader extends VectorizedReaderBase {
// Read the header
this.blockSizeInValues = BytesUtils.readUnsignedVarInt(in);
this.miniBlockNumInABlock = BytesUtils.readUnsignedVarInt(in);
+ // Validate the block layout from the page header before sizing any buffers from it.
+ if (blockSizeInValues <= 0 || blockSizeInValues > MAX_BLOCK_SIZE_IN_VALUES) {
+ 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) {
+ 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;
// True value count. May be less than valueCount because of nulls
this.totalValueCount = BytesUtils.readUnsignedVarInt(in);
+ if (totalValueCount < 0) {
+ throw new ParquetDecodingException(
+ "Invalid DELTA_BINARY_PACKED total value count: " + totalValueCount);
+ }
this.bitWidths = new int[miniBlockNumInABlock];
this.unpackedValuesBuffer = new long[miniBlockSizeInValues];
this.intScratchBuffer = new int[miniBlockSizeInValues];
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetDeltaEncodingSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetDeltaEncodingSuite.scala
index b72b55344507..bf2bf5b010fa 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetDeltaEncodingSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetDeltaEncodingSuite.scala
@@ -16,11 +16,11 @@
*/
package org.apache.spark.sql.execution.datasources.parquet
-import java.io.IOException
+import java.io.{ByteArrayOutputStream, IOException}
import java.nio.ByteBuffer
import java.util.Random
-import org.apache.parquet.bytes.{ByteBufferInputStream, DirectByteBufferAllocator}
+import org.apache.parquet.bytes.{ByteBufferInputStream, BytesUtils, DirectByteBufferAllocator}
import org.apache.parquet.column.values.ValuesWriter
import org.apache.parquet.column.values.delta.{DeltaBinaryPackingValuesWriterForInteger, DeltaBinaryPackingValuesWriterForLong}
import org.apache.parquet.io.ParquetDecodingException
@@ -172,6 +172,40 @@ abstract class ParquetDeltaEncodingSuite[T] extends ParquetCompatibilityTest
}
}
+ test("reject invalid DELTA_BINARY_PACKED page header") {
+ // A corrupt header declaring blockSizeInValues = 2^30 would force multi-gigabyte scratch
+ // allocations from a few-byte page. The reader must reject it before allocating anything.
+ def craftHeader(blockSize: Int, miniBlockNum: Int, totalCount: Int): ByteBufferInputStream = {
+ val out = new ByteArrayOutputStream()
+ BytesUtils.writeUnsignedVarInt(blockSize, out)
+ BytesUtils.writeUnsignedVarInt(miniBlockNum, out)
+ BytesUtils.writeUnsignedVarInt(totalCount, out)
+ out.write(0) // firstValue (zigzag var long 0)
+ ByteBufferInputStream.wrap(ByteBuffer.wrap(out.toByteArray))
+ }
+
+ reader = new VectorizedDeltaBinaryPackedReader
+ val e = intercept[ParquetDecodingException] {
+ reader.initFromPage(100, craftHeader(1 << 30, 1, 1))
+ }
+ assert(e.getMessage.contains("Invalid DELTA_BINARY_PACKED block size"))
+
+ // A mini block count larger than the block size (mini blocks of zero values) is likewise
+ // rejected.
+ reader = new VectorizedDeltaBinaryPackedReader
+ val e2 = intercept[ParquetDecodingException] {
+ reader.initFromPage(100, craftHeader(128, 256, 1))
+ }
+ assert(e2.getMessage.contains("mini block count"))
+
+ // A negative total value count is rejected as well.
+ reader = new VectorizedDeltaBinaryPackedReader
+ val e3 = intercept[ParquetDecodingException] {
+ reader.initFromPage(100, craftHeader(128, 4, -1))
+ }
+ assert(e3.getMessage.contains("total value count"))
+ }
+
test("skip()") {
val data = allocDataArray(5 * blockSize + 1)
for (i <- data.indices) {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org