You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by do...@apache.org on 2026/09/29 02:26:09 UTC
(spark) branch branch-4.2 updated: [SPARK-59828][SQL] Support `skipFixedLenByteArray` in `VectorizedDeltaByteArrayReader`
This is an automated email from the ASF dual-hosted git repository.
dongjoon-hyun 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 e35db51c005a [SPARK-59828][SQL] Support `skipFixedLenByteArray` in `VectorizedDeltaByteArrayReader`
e35db51c005a is described below
commit e35db51c005abfacc0ad94a62c6bc7c8a044613b
Author: Dongjoon Hyun <do...@apache.org>
AuthorDate: Mon Sep 28 11:10:31 2026 -0700
[SPARK-59828][SQL] Support `skipFixedLenByteArray` in `VectorizedDeltaByteArrayReader`
### What changes were proposed in this pull request?
This PR aims to implement `skipFixedLenByteArray` in `VectorizedDeltaByteArrayReader` by delegating to `skipBinary`, because `DELTA_BYTE_ARRAY` encodes `FIXED_LEN_BYTE_ARRAY` values in the same way as `BYTE_ARRAY` values.
### Why are the changes needed?
The Parquet v2 writer encodes `FIXED_LEN_BYTE_ARRAY` columns such as `DECIMAL(38, 18)` with `DELTA_BYTE_ARRAY` when dictionary encoding is disabled. When Parquet column index filtering skips rows of such a column, the vectorized reader fails because `VectorizedDeltaByteArrayReader` does not implement `skipFixedLenByteArray`.
```scala
spark.range(0, 2000).selectExpr("id as _1", "cast(id as decimal(38, 18)) as _2")
.coalesce(1)
.write
.option("parquet.writer.version", "PARQUET_2_0")
.option("parquet.enable.dictionary", "false")
.option("parquet.page.size", "2048")
.option("parquet.page.row.count.limit", "500")
.parquet("t")
spark.read.parquet("t").filter("_1 >= 500 and _1 < 1000").collect()
```
```
org.apache.spark.SparkUnsupportedOperationException: [UNSUPPORTED_CALL.WITHOUT_SUGGESTION] Cannot call the method "skipFixedLenByteArray" of the class "org.apache.spark.sql.execution.datasources.parquet.VectorizedReaderBase". SQLSTATE: 0A000
```
### Does this PR introduce _any_ user-facing change?
No, this fixes a bug that has existed since Apache Spark [v3.3.0](https://github.com/apache/spark/releases/tag/v3.3.0) (2022-06-09) but it didn't work before.
### How was this patch tested?
Pass the CIs with the newly added test cases, which fail without this fix.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 5.5
Closes #59101 from dongjoon-hyun/SPARK-59828.
Authored-by: Dongjoon Hyun <do...@apache.org>
Signed-off-by: Dongjoon Hyun <do...@apache.org>
(cherry picked from commit 9e3396008af900b36978aeff8b4fc96c53fa6090)
Signed-off-by: Dongjoon Hyun <do...@apache.org>
(cherry picked from commit d0ddbccefa7627f5a52cb84c8bc6a1ed8a4cf774)
Signed-off-by: Dongjoon Hyun <do...@apache.org>
---
.../parquet/VectorizedDeltaByteArrayReader.java | 6 ++++++
.../parquet/ParquetColumnIndexSuite.scala | 21 +++++++++++++++++++++
.../ParquetDeltaByteArrayEncodingSuite.scala | 15 +++++++++++++++
3 files changed, 42 insertions(+)
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 4d758d2cc59b..49eacd59df22 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
@@ -207,4 +207,10 @@ public class VectorizedDeltaByteArrayReader extends VectorizedReaderBase
}
}
+ @Override
+ public void skipFixedLenByteArray(int total, int len) {
+ // DELTA_BYTE_ARRAY encodes FIXED_LEN_BYTE_ARRAY values in the same way as BYTE_ARRAY values.
+ skipBinary(total);
+ }
+
}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetColumnIndexSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetColumnIndexSuite.scala
index 9fc0560a4802..44acdb2c3138 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetColumnIndexSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetColumnIndexSuite.scala
@@ -115,4 +115,25 @@ class ParquetColumnIndexSuite extends ParquetTest with SharedSparkSession {
df => df.filter("(s._1 >= 500 and s._1 < 1000) or (s._1 >= 1500 and s._1 < 1600)")
)
}
+
+ test("SPARK-59828: reading unaligned pages - FIXED_LEN_BYTE_ARRAY with DELTA_BYTE_ARRAY") {
+ // Without a dictionary, the v2 writer encodes FIXED_LEN_BYTE_ARRAY as DELTA_BYTE_ARRAY.
+ // The pages of _2 are split by size, and those of _1 by row count because its
+ // DELTA_BINARY_PACKED values are tiny, so that the pages of the two columns are unaligned.
+ val df = spark.range(0, 2000).selectExpr("id as _1", "cast(id as decimal(38, 18)) as _2")
+ withTempPath { file =>
+ df.coalesce(1)
+ .write
+ .option("parquet.writer.version", "PARQUET_2_0")
+ .option("parquet.enable.dictionary", "false")
+ .option("parquet.page.size", "2048")
+ .option("parquet.page.row.count.limit", "500")
+ .parquet(file.getCanonicalPath)
+
+ val parquetDf = spark.read.parquet(file.getCanonicalPath)
+ actions.foreach { action =>
+ checkAnswer(action(parquetDf), action(df))
+ }
+ }
+ }
}
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 b8ef5dab73ee..7ba58b46ab12 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
@@ -65,6 +65,21 @@ class ParquetDeltaByteArrayEncodingSuite extends ParquetCompatibilityTest with S
assertReadWriteWithSkipN(writer, reader, randvalues)
}
+ test("SPARK-59828: fixed-length byte arrays with skipN") {
+ // FIXED_LEN_BYTE_ARRAY values, e.g. decimal(38, 18), can also be DELTA_BYTE_ARRAY encoded.
+ val len = 16
+ val vals = Array.tabulate(1000)(i => f"$i%016d")
+ Utils.writeData(writer, vals)
+ reader.initFromPage(vals.length, writer.getBytes.toInputStream)
+ var i = 0
+ while (i < vals.length) {
+ val skipCount = (vals.length - i) / 2
+ assert(vals(i).getBytes() sameElements reader.readBinary(len).getBytes)
+ reader.skipFixedLenByteArray(skipCount, len)
+ i += skipCount + 1
+ }
+ }
+
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.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org