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/24 23:17:39 UTC
(spark) branch branch-4.2 updated: [SPARK-59352][SQL] Validate prefix lengths in the vectorized DELTA_BYTE_ARRAY reader
This is an automated email from the ASF dual-hosted git repository.
asf-gitbox-commits pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.2 by this push:
new d7c7f5f6878d [SPARK-59352][SQL] Validate prefix lengths in the vectorized DELTA_BYTE_ARRAY reader
d7c7f5f6878d is described below
commit d7c7f5f6878d6d9f91f254b5287c6df43366c065
Author: Holden Karau <ho...@pigscanfly.ca>
AuthorDate: Thu Sep 24 16:09:17 2026 -0700
[SPARK-59352][SQL] Validate prefix lengths in the vectorized DELTA_BYTE_ARRAY reader
### What changes were proposed in this pull request?
Validate and fast fail invalid prefix lengths while reading parquet data
### Why are the changes needed?
The prefix length in a DELTA_BYTE_ARRAY page is read from the file and was used to copy bytes out of the previous value without validation. Reject negative prefix lengths and prefix lengths larger than the previous value with a ParquetDecodingException, so corrupt pages fail the read deterministically instead of producing malformed values or hard-to-diagnose low-level exceptions.
### Does this PR introduce _any_ user-facing change?
No
### How was this patch tested?
New unit test
### Was this patch authored or co-authored using generative AI tooling?
Generated by: Claude (mixture of version) & Cursor (with Kimi K3).
Closes #58639 from holdenk/33-parquet-delta-page-validation.
Lead-authored-by: Holden Karau <ho...@pigscanfly.ca>
Co-authored-by: Holden Karau <ho...@snowflake.com>
Signed-off-by: Holden Karau <ho...@snowflake.com>
(cherry picked from commit 08d777aa701262e61afd81a0d02308e7f8c21184)
Signed-off-by: Holden Karau <ho...@snowflake.com>
---
.../parquet/VectorizedDeltaByteArrayReader.java | 22 +++++
.../ParquetDeltaByteArrayEncodingSuite.scala | 96 +++++++++++++++++++++-
2 files changed, 117 insertions(+), 1 deletion(-)
diff --git a/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedDeltaByteArrayReader.java b/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedDeltaByteArrayReader.java
index 1edee60bc564..4d758d2cc59b 100644
--- a/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedDeltaByteArrayReader.java
+++ b/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedDeltaByteArrayReader.java
@@ -22,6 +22,7 @@ import static org.apache.spark.sql.types.DataTypes.IntegerType;
import org.apache.parquet.bytes.ByteBufferInputStream;
import org.apache.parquet.column.values.RequiresPreviousReader;
import org.apache.parquet.column.values.ValuesReader;
+import org.apache.parquet.io.ParquetDecodingException;
import org.apache.parquet.io.api.Binary;
import org.apache.spark.sql.execution.vectorized.OnHeapColumnVector;
import org.apache.spark.sql.execution.vectorized.WritableColumnVector;
@@ -71,6 +72,24 @@ public class VectorizedDeltaByteArrayReader extends VectorizedReaderBase
return Binary.fromConstantByteArray(binaryValVector.getBinary(0));
}
+ /**
+ * The prefix length is read from the file, so validate it against the previous value before
+ * using it to copy bytes out of that value. A corrupt page must fail the read instead of
+ * producing a malformed value.
+ */
+ private void checkPrefixLength(int prefixLength) {
+ if (prefixLength < 0) {
+ throw new ParquetDecodingException(
+ "Corrupted DELTA_BYTE_ARRAY page: negative prefix length: " + prefixLength);
+ }
+ int previousLength = previous == null ? 0 : previous.remaining();
+ if (prefixLength > previousLength) {
+ throw new ParquetDecodingException(
+ "Corrupted DELTA_BYTE_ARRAY page: prefix length " + prefixLength
+ + " is larger than the previous value's length " + previousLength);
+ }
+ }
+
private void readValues(int total, WritableColumnVector c, int rowId) {
for (int i = 0; i < total; i++) {
// NOTE: due to PARQUET-246, it is important that we
@@ -79,6 +98,7 @@ public class VectorizedDeltaByteArrayReader extends VectorizedReaderBase
// value of the page should have an empty prefix, it may not
// because of PARQUET-246.
int prefixLength = prefixLengthVector.getInt(currentRow);
+ checkPrefixLength(prefixLength);
ByteBuffer suffix = suffixReader.getBytes(currentRow);
byte[] suffixArray = suffix.array();
int suffixLength = suffix.limit() - suffix.position();
@@ -120,6 +140,7 @@ public class VectorizedDeltaByteArrayReader extends VectorizedReaderBase
WKBConverterStrategy converter) {
for (int i = 0; i < total; i++) {
int prefixLength = prefixLengthVector.getInt(currentRow);
+ checkPrefixLength(prefixLength);
ByteBuffer suffix = suffixReader.getBytes(currentRow);
int suffixLength = suffix.limit() - suffix.position();
int length = prefixLength + suffixLength;
@@ -165,6 +186,7 @@ public class VectorizedDeltaByteArrayReader extends VectorizedReaderBase
for (int i = 0; i < total; i++) {
int prefixLength = prefixLengthVector.getInt(currentRow);
+ checkPrefixLength(prefixLength);
ByteBuffer suffix = suffixReader.getBytes(currentRow);
byte[] suffixArray = suffix.array();
int suffixLength = suffix.limit() - suffix.position();
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetDeltaByteArrayEncodingSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetDeltaByteArrayEncodingSuite.scala
index d62ef9c3f197..b8ef5dab73ee 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetDeltaByteArrayEncodingSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetDeltaByteArrayEncodingSuite.scala
@@ -16,9 +16,15 @@
*/
package org.apache.spark.sql.execution.datasources.parquet
-import org.apache.parquet.bytes.DirectByteBufferAllocator
+import java.nio.charset.StandardCharsets
+
+import org.apache.parquet.bytes.{ByteBufferInputStream, BytesInput, DirectByteBufferAllocator}
import org.apache.parquet.column.values.Utils
+import org.apache.parquet.column.values.delta.DeltaBinaryPackingValuesWriterForInteger
+import org.apache.parquet.column.values.deltalengthbytearray.DeltaLengthByteArrayValuesWriter
import org.apache.parquet.column.values.deltastrings.DeltaByteArrayWriter
+import org.apache.parquet.io.ParquetDecodingException
+import org.apache.parquet.io.api.Binary
import org.apache.spark.sql.catalyst.util.STUtils
import org.apache.spark.sql.execution.vectorized.{OnHeapColumnVector, WritableColumnVector}
@@ -59,6 +65,58 @@ class ParquetDeltaByteArrayEncodingSuite extends ParquetCompatibilityTest with S
assertReadWriteWithSkipN(writer, reader, randvalues)
}
+ test("prefix length larger than the previous value is rejected") {
+ // Craft a page by hand: a benign 2-byte first value, then a value whose prefix length
+ // claims 65536 bytes of a 2-byte previous value.
+ val is = craftPage(prefixLengths = Array(0, 65536), suffixes = Array("ab", ""))
+ reader.initFromPage(2, is)
+ writableColumnVector = new OnHeapColumnVector(2, StringType)
+ val e = intercept[ParquetDecodingException] {
+ reader.readBinary(2, writableColumnVector, 0)
+ }
+ assert(e.getMessage.contains("prefix length 65536"))
+ }
+
+ test("negative prefix length is rejected") {
+ val is = craftPage(prefixLengths = Array(0, -1), suffixes = Array("ab", "cd"))
+ reader.initFromPage(2, is)
+ writableColumnVector = new OnHeapColumnVector(2, StringType)
+ val e = intercept[ParquetDecodingException] {
+ reader.readBinary(2, writableColumnVector, 0)
+ }
+ assert(e.getMessage.contains("negative prefix length"))
+ }
+
+ test("prefix length larger than the previous value is rejected when skipping") {
+ val is = craftPage(prefixLengths = Array(0, 65536), suffixes = Array("ab", ""))
+ reader.initFromPage(2, is)
+ val e = intercept[ParquetDecodingException] {
+ reader.skipBinary(2)
+ }
+ assert(e.getMessage.contains("prefix length 65536"))
+ }
+
+ test("prefix length one byte past the previous value is rejected") {
+ // With on-heap vectors previous.array() is the whole backing buffer, so without the
+ // validation this off-by-one prefix silently copied a stale byte instead of failing.
+ val is = craftPage(prefixLengths = Array(0, 3), suffixes = Array("ab", ""))
+ reader.initFromPage(2, is)
+ writableColumnVector = new OnHeapColumnVector(2, StringType)
+ val e = intercept[ParquetDecodingException] {
+ reader.readBinary(2, writableColumnVector, 0)
+ }
+ assert(e.getMessage.contains("prefix length 3"))
+ }
+
+ test("prefix length equal to the previous value's length is accepted") {
+ val is = craftPage(prefixLengths = Array(0, 2), suffixes = Array("ab", ""))
+ reader.initFromPage(2, is)
+ writableColumnVector = new OnHeapColumnVector(2, StringType)
+ reader.readBinary(2, writableColumnVector, 0)
+ assert(writableColumnVector.getBinary(0) sameElements "ab".getBytes)
+ assert(writableColumnVector.getBinary(1) sameElements "ab".getBytes)
+ }
+
test("test lengths") {
var reader = new VectorizedDeltaBinaryPackedReader
Utils.writeData(writer, values)
@@ -100,6 +158,42 @@ class ParquetDeltaByteArrayEncodingSuite extends ParquetCompatibilityTest with S
geoType)
}
+ testGeo("geo path rejects prefix length larger than the previous value") { geoType =>
+ // The geo read path copies the prefix with previous.get(wkb, 0, prefixLength) rather
+ // than appendBytes.
+ val is = craftPage(
+ prefixLengths = Array(0, 65536),
+ suffixes = Array(makePointWkb(1, 1), Array.empty[Byte]))
+ reader.initFromPage(2, is)
+ writableColumnVector = new OnHeapColumnVector(2, geoType)
+ val e = intercept[ParquetDecodingException] {
+ geoType match {
+ case _: GeometryType => reader.readGeometry(2, writableColumnVector, 0)
+ case _: GeographyType => reader.readGeography(2, writableColumnVector, 0)
+ }
+ }
+ assert(e.getMessage.contains("prefix length 65536"))
+ }
+
+ /** Builds a raw DELTA_BYTE_ARRAY page from explicit prefix lengths and suffixes. */
+ private def craftPage(
+ prefixLengths: Array[Int],
+ suffixes: Array[String]): ByteBufferInputStream = {
+ craftPage(prefixLengths, suffixes.map(_.getBytes(StandardCharsets.UTF_8)))
+ }
+
+ private def craftPage(
+ prefixLengths: Array[Int],
+ suffixes: Array[Array[Byte]]): ByteBufferInputStream = {
+ val allocator = new DirectByteBufferAllocator
+ val prefixWriter =
+ new DeltaBinaryPackingValuesWriterForInteger(128, 4, 64 * 1024, 64 * 1024, allocator)
+ val suffixWriter = new DeltaLengthByteArrayValuesWriter(64 * 1024, 64 * 1024, allocator)
+ prefixLengths.foreach(prefixWriter.writeInteger)
+ suffixes.foreach(s => suffixWriter.writeBytes(Binary.fromConstantByteArray(s)))
+ BytesInput.concat(prefixWriter.getBytes, suffixWriter.getBytes).toInputStream
+ }
+
private def assertGeoReadWrite(
writer: DeltaByteArrayWriter,
reader: VectorizedDeltaByteArrayReader,
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org