[jira] [Created] (SPARK-59171) [SQL] SchemaPruning is not idempotent when a query reads _metadata and a VARIANT column is only referenced below a nondeterministic filter

["Felipe (Jira)" <ji...@apache.org>]

Felipe created SPARK-59171:
------------------------------

             Summary: [SQL] SchemaPruning is not idempotent when a query reads _metadata and a VARIANT column is only referenced below a nondeterministic filter
                 Key: SPARK-59171
                 URL: https://issues.apache.org/jira/browse/SPARK-59171
             Project: Spark
          Issue Type: Bug
          Components: Spark Core
    Affects Versions: 4.2.0
            Reporter: Felipe


Originally created at https://github.com/apache/spark/issues/57659
h3. 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.
h3. Reproduction

Plain Parquet, no third-party code. This is written as a Spark unit test because {{RuleExecutor}} only checks idempotence when {{{}Utils.isTesting{}}}:
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()
  }
}
 
h3. Why the shape matters

Three ingredients appear to be needed:
 # the query outputs *only* {{{}_metadata.file_path{}}}, so no data column is required above the scan;
 # 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;
 # 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.
h3. Impact

{{RuleExecutor}} only runs {{checkBatchIdempotence}} under {{{}Utils.isTesting{}}}:
// 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.
h3. 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.)
h3. Workaround

Disabling nested schema pruning for the affected query avoids it:
spark.sql.optimizer.nestedSchemaPruning.enabled = false



--
This message was sent by Atlassian Jira
(v8.20.10#820010)

---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@spark.apache.org
For additional commands, e-mail: issues-help@spark.apache.org