This is an automated email from the ASF dual-hosted git repository.
viirya 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 a63cd8dc7160 [SPARK-58918][SQL][TEST][FOLLOWUP] Cover deeply nested
types in the Arrow cache column-pruning read
a63cd8dc7160 is described below
commit a63cd8dc71605386f78954e94c3ab4fe8fbad891
Author: Liang-Chi Hsieh <[email protected]>
AuthorDate: Sat Aug 22 09:18:06 2026 -0700
[SPARK-58918][SQL][TEST][FOLLOWUP] Cover deeply nested types in the Arrow
cache column-pruning read
### What changes were proposed in this pull request?
Add a test to `ArrowCachedBatchSerializerSuite` covering deeply nested
types in the Arrow cache column-pruning read path added by #58177
(SPARK-58918): `array<struct<...>>`, `struct<struct<struct<...>>>`, and
`map<int, array<int>>`, each selected alone, mixed with primitives, reordered,
and with other columns pruned in between, under both the row and vectorized
read paths.
### Why are the changes needed?
The pruning read copies each selected top-level column's entire buffer
subtree, so `readProjectedRecordBatch`'s buffer-span arithmetic
(`fieldBufferCount` / `fieldNodeCount`) recurses through nested types to
arbitrary depth. The original test only exercised one level of nesting
(`array<int>`, `struct` of primitives), so an off-by-one in the recursive span
over a deeper subtree -- which would shift the buffers of a following selected
column and produce wrong values -- was not covered. [...]
### Does this PR introduce _any_ user-facing change?
No, test-only.
### How was this patch tested?
The new test; full `ArrowCachedBatchSerializerSuite` passes.
### Was this patch authored or co-authored using generative AI tooling?
Yes, this pull request and its description were written by Claude Code.
Closes #58215 from viirya/arrow-cache-nested-projection-test.
Authored-by: Liang-Chi Hsieh <[email protected]>
Signed-off-by: Liang-Chi Hsieh <[email protected]>
(cherry picked from commit 09ebec3cb0d7c5ba38ee49282d95339b1a1fbe34)
Signed-off-by: Liang-Chi Hsieh <[email protected]>
---
.../columnar/ArrowCachedBatchSerializerSuite.scala | 56 ++++++++++++++++++++++
1 file changed, 56 insertions(+)
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/columnar/ArrowCachedBatchSerializerSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/columnar/ArrowCachedBatchSerializerSuite.scala
index 9fe540388e37..b3d6c3539a10 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/columnar/ArrowCachedBatchSerializerSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/columnar/ArrowCachedBatchSerializerSuite.scala
@@ -304,6 +304,62 @@ class ArrowCachedBatchSerializerSuite extends QueryTest
with SharedSparkSession
}
}
+ test("column projection prunes deeply nested columns on load") {
+ // The buffer-span arithmetic must skip (and, when selected, copy) a
column's ENTIRE buffer
+ // subtree, at arbitrary nesting depth. Exercise array<struct>,
struct<struct<struct>>, and
+ // map<int, array<int>>, each selected while pruning neighbours whose own
subtrees have varying
+ // buffer counts, under both read paths, so an off-by-one in the recursive
span would surface
+ // as wrong values in a following column.
+ Seq(false, true).foreach { vectorized =>
+ withSQLConf(SQLConf.CACHE_VECTORIZED_READER_ENABLED.key ->
vectorized.toString) {
+ val schema = new StructType()
+ .add("id", IntegerType)
+ .add("arrOfStruct", ArrayType(new StructType().add("x",
LongType).add("y", StringType)))
+ .add("s", StringType)
+ .add("deepStruct", new StructType()
+ .add("l1", new StructType()
+ .add("l2", new StructType().add("v", LongType).add("t",
StringType))))
+ .add("mapOfArray", MapType(IntegerType, ArrayType(IntegerType)))
+ val rows = (1 to 30).map { i =>
+ Row(i, Seq(Row(i.toLong, s"a$i"), Row((i + 1).toLong, s"b$i")),
s"s$i",
+ Row(Row(Row(i.toLong * 10, s"deep$i"))), Map(i -> Seq(i, i + 1, i
+ 2)))
+ }
+ val df = spark.createDataFrame(
+ spark.sparkContext.parallelize(rows, 1), schema).cache()
+ try {
+ def expected(cols: Seq[String]): Seq[Row] = (1 to 30).map { i =>
+ Row.fromSeq(cols.map {
+ case "id" => i
+ case "arrOfStruct" => Seq(Row(i.toLong, s"a$i"), Row((i +
1).toLong, s"b$i"))
+ case "s" => s"s$i"
+ case "deepStruct" => Row(Row(Row(i.toLong * 10, s"deep$i")))
+ case "mapOfArray" => Map(i -> Seq(i, i + 1, i + 2))
+ })
+ }
+ // Each nested column selected alone: its whole subtree must be
copied, nothing else.
+ checkAnswer(df.select("arrOfStruct"), expected(Seq("arrOfStruct")))
+ checkAnswer(df.select("deepStruct"), expected(Seq("deepStruct")))
+ checkAnswer(df.select("mapOfArray"), expected(Seq("mapOfArray")))
+ // A primitive after a pruned deep column: the skip must span the
whole deep subtree.
+ checkAnswer(df.select("id", "s"), expected(Seq("id", "s")))
+ // Reordered mix of nested and primitive columns, pruning others in
between.
+ checkAnswer(
+ df.select("mapOfArray", "id", "deepStruct"),
+ expected(Seq("mapOfArray", "id", "deepStruct")))
+ checkAnswer(
+ df.select("deepStruct", "arrOfStruct"),
+ expected(Seq("deepStruct", "arrOfStruct")))
+ // Full projection round-trips.
+ checkAnswer(
+ df.select("id", "arrOfStruct", "s", "deepStruct", "mapOfArray"),
+ expected(Seq("id", "arrOfStruct", "s", "deepStruct",
"mapOfArray")))
+ } finally {
+ df.unpersist()
+ }
+ }
+ }
+ }
+
test("caching with multiple batches") {
withSQLConf(SQLConf.ARROW_EXECUTION_MAX_RECORDS_PER_BATCH.key -> "10") {
val df = (1 to 50).map(i => (i, s"str$i")).toDF("a", "b")
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
