From c83cf0a09e6f6517110971fc51f24da57a048414 Mon Sep 17 00:00:00 2001 From: Holden Karau Date: Tue, 1 Sep 2026 04:44:42 +0000 Subject: [PATCH] [SQL] Validate prefix lengths in the vectorized DELTA_BYTE_ARRAY reader 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. Co-authored-by: Cursor --- .../VectorizedDeltaByteArrayReader.java | 22 ++++++++ .../ParquetDeltaByteArrayEncodingSuite.scala | 50 ++++++++++++++++++- 2 files changed, 71 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 1edee60bc5644..209524ba2cb42 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 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 Binary readBinary(int len) { 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 @@ private void readValues(int total, WritableColumnVector c, int rowId) { // 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 @@ private void readGeoData(int total, WritableColumnVector c, int rowId, int srid, 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 void skipBinary(int total) { 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 d62ef9c3f197d..77a2caaa8e197 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,13 @@ */ package org.apache.spark.sql.execution.datasources.parquet -import org.apache.parquet.bytes.DirectByteBufferAllocator +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 +63,37 @@ 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("test lengths") { var reader = new VectorizedDeltaBinaryPackedReader Utils.writeData(writer, values) @@ -100,6 +135,19 @@ class ParquetDeltaByteArrayEncodingSuite extends ParquetCompatibilityTest with S geoType) } + /** Builds a raw DELTA_BYTE_ARRAY page from explicit prefix lengths and suffixes. */ + private def craftPage( + prefixLengths: Array[Int], + suffixes: Array[String]): 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.fromString(s))) + BytesInput.concat(prefixWriter.getBytes, suffixWriter.getBytes).toInputStream + } + private def assertGeoReadWrite( writer: DeltaByteArrayWriter, reader: VectorizedDeltaByteArrayReader,