Skip to content

Commit 809b7e7

Browse files
goutamadwanturos-b
authored andcommitted
[SPARK-59171][SQL] Make SchemaPruning idempotent after variant pushdown
### What changes were proposed in this pull request? Run `SchemaPruning` after `PushVariantIntoScan` in the early scan pushdown batch. This removes variant reconstruction projections that become unnecessary during variant pushdown and allows the `Once` batch to reach the same plan on its first application. Add a Parquet V1 regression test covering a query that reads `_metadata.file_path` while a VARIANT column is referenced below a nondeterministic filter. Closes #57659. Tracks SPARK-59171. ### Why are the changes needed? `PushVariantIntoScan` can make a variant reconstruction projection unnecessary after the earlier schema-pruning passes have completed. Reapplying the optimizer batch then removes that projection, so the first application is not a fixed point. `RuleExecutor` checks `Once` batch idempotence only when `Utils.isTesting` is true. This causes affected Spark and Delta tests to fail, but production queries do not throw this error or return incorrect results. Outside test mode, the batch runs once and leaves a correct plan with a redundant reconstruction projection. Running `SchemaPruning` once after variant pushdown makes the first application reach the stable plan. This change targets `master`; no backports are requested. ### Does this PR introduce _any_ user-facing change? No. Production query results are unchanged. The patch removes a redundant reconstruction projection and prevents the testing-only `Once` batch idempotence failure. ### How was this patch tested? The new regression test was confirmed to fail before the optimizer change and pass afterward. The following tests and checks passed: - `./build/sbt 'sql/testOnly org.apache.spark.sql.execution.datasources.parquet.ParquetV1SchemaPruningSuite -- -z "SPARK-59171"'` - `./build/sbt 'sql/testOnly org.apache.spark.sql.execution.datasources.parquet.ParquetV1SchemaPruningSuite org.apache.spark.sql.execution.datasources.PushVariantIntoScanSuite org.apache.spark.sql.execution.datasources.PushVariantIntoScanVectorizedSuite'` (349 tests) - `./build/sbt 'sql/scalastyle' 'sql/Test/scalastyle'` - `git diff --check` The full `./dev/run-tests` suite was not run locally. ### Was this patch authored or co-authored using generative AI tooling? AI was used to review the code and understand the existing codebase. This contribution is my original work, and I license it under the project's open source license. Closes #57809 from goutamadwant/fix-schema-pruning-idempotence. Authored-by: Goutam Adwant <workwithgoutam@gmail.com> Signed-off-by: Uros Bojanic <221401595+uros-b@users.noreply.github.com> (cherry picked from commit a5a9e7b) Signed-off-by: Uros Bojanic <221401595+uros-b@users.noreply.github.com>
1 parent 144e227 commit 809b7e7

2 files changed

Lines changed: 22 additions & 1 deletion

File tree

sql/core/src/main/scala/org/apache/spark/sql/execution/SparkOptimizer.scala

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -53,7 +53,10 @@ class SparkOptimizer(
5353
V2ScanPartitioningAndOrdering,
5454
V2Writes,
5555
PruneFileSourcePartitions,
56-
PushVariantIntoScan)
56+
PushVariantIntoScan,
57+
// Variant pushdown can make a reconstruction projection unnecessary. Prune again so this
58+
// Once batch reaches the same plan on its first application as it would on a second one.
59+
SchemaPruning)
5760

5861
override def preCBORules: Seq[Rule[LogicalPlan]] =
5962
Seq(OptimizeMetadataOnlyDeleteFromTable)

sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetSchemaPruningSuite.scala

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,7 @@ import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
2424
import org.apache.spark.sql.execution.datasources.SchemaPruningSuite
2525
import org.apache.spark.sql.execution.datasources.v2.BatchScanExec
2626
import org.apache.spark.sql.execution.datasources.v2.parquet.ParquetScan
27+
import org.apache.spark.sql.functions.{col, udf}
2728
import org.apache.spark.sql.internal.SQLConf
2829
import org.apache.spark.tags.ExtendedSQLTest
2930

@@ -42,6 +43,23 @@ class ParquetV1SchemaPruningSuite extends ParquetSchemaPruningSuite {
4243
super
4344
.sparkConf
4445
.set(SQLConf.USE_V1_SOURCE_LIST, "parquet")
46+
47+
test("SPARK-59171: SchemaPruning is idempotent with metadata and a filtered variant") {
48+
withTempPath { path =>
49+
spark.range(10)
50+
.selectExpr("parse_json(cast(id as string)) as v")
51+
.write.parquet(path.getCanonicalPath)
52+
53+
val alwaysTrue = udf(() => true).asNondeterministic()
54+
val query = spark.read.parquet(path.getCanonicalPath)
55+
.where("v::int = 3")
56+
.select(col("_metadata.file_path"))
57+
.filter(alwaysTrue())
58+
.distinct()
59+
60+
assert(query.collect().nonEmpty)
61+
}
62+
}
4563
}
4664

4765
@ExtendedSQLTest

0 commit comments

Comments
 (0)