[SQL] SchemaPruning is not idempotent when a query reads `_metadata` and a VARIANT column is only referenced below a nondeterministic filter

[felipepessoto]

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:

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:

// 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:

spark.sql.optimizer.nestedSchemaPruning.enabled = false