felipepessoto opened a new issue, #57659:
URL: https://github.com/apache/spark/issues/57659
### What
The `Early Filter and Projection Push-Down` batch is a `Once` batch, and it
is not idempotent for a particular plan shape. `RuleExecutor` detects this and
throws:
```
org.apache.spark.SparkRuntimeException: Once strategy's idempotence is
broken for batch Early Filter and Projection Push-Down
Aggregate [file_path#4], [file_path#4] Aggregate
[file_path#4], [file_path#4]
+- Project [_metadata#3.file_path AS file_path#4] +- Project
[_metadata#3.file_path AS file_path#4]
+- Filter UDF() +- Filter UDF()
! +- Project [v#9.0 AS v#2, _metadata#3] +- Filter
(isnotnull(v#9) AND (v#9.1 = 3))
! +- Filter (isnotnull(v#9) AND (v#9.1 = 3)) +- Relation
[v#9,_metadata#3] parquet
! +- Relation [v#9,_metadata#3] parquet
```
Applying the batch a second time removes the `Project [v#9.0 AS v#2,
_metadata#3]` node, so the result of the first application is not a fixed point.
### Reproduction
Plain Parquet, no third-party code. This is written as a Spark unit test
because `RuleExecutor` only checks idempotence when `Utils.isTesting`:
```scala
test("SchemaPruning idempotence with variant and _metadata") {
withTempDir { dir =>
val path = new java.io.File(dir, "t").getAbsolutePath
spark.range(0, 10)
.selectExpr("parse_json(cast(id as string)) as v")
.write.parquet(path)
val alwaysTrue = udf(() => true).asNondeterministic()
spark.read.parquet(path)
.where("v::int = 3") // filter references the
variant column
.select(col("_metadata.file_path")) // the only output column
.filter(alwaysTrue()) // nondeterministic, so it is
not pushed down
.distinct()
.collect()
}
}
```
### Why the shape matters
Three ingredients appear to be needed:
1. the query outputs **only** `_metadata.file_path`, so no data column is
required above the scan;
2. the filter references a **VARIANT** column, so it is pushed below the
variant reconstruction projection (`v#9.0 AS v#2`), leaving that projection
unused;
3. a **nondeterministic** filter sits above, which prevents the unused
projection from being collapsed in the same pass.
Removing any one of them makes the failure go away. In particular, a plain
nested `struct` column in place of the variant works fine, as does the same
query without the nondeterministic filter.
### Impact
`RuleExecutor` only runs `checkBatchIdempotence` under `Utils.isTesting`:
```scala
// Check idempotence for Once batches.
if (batch.strategy == Once &&
Utils.isTesting && !excludedOnceBatches.contains(batch.name)) {
checkBatchIdempotence(batch, curPlan)
}
```
So this is not a wrong-results bug for end users: outside tests the batch
runs once and the plan it produces is correct, just with a redundant projection
left in place. The practical impact is that the optimizer leaves a plan it
would keep rewriting, and that any project whose test suites run in a Spark
test JVM and that builds this plan shape fails.
It surfaced in Delta Lake, whose `UPDATE`/`DELETE` identify the files to
rewrite with a query of exactly this shape: they read `_metadata.file_path`,
and they wrap the scan in a nondeterministic filter that increments a SQL
metric.
### Versions
Reproduced with the snippet above on Spark **4.0**, **4.1** and **4.2**, so
this does not look like a recent regression.
(For context, Delta Lake only sees it on 4.1 and 4.2, because the variant
reconstruction projection in the plan depends on shredded variant read support.)
### Workaround
Disabling nested schema pruning for the affected query avoids it:
```scala
spark.sql.optimizer.nestedSchemaPruning.enabled = false
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
