[SPARK-59107][SQL][4.3] Do not widen a projection-sensitive V1 file read

[LuciferYang]

What changes were proposed in this pull request?

Backport of b82f9872d1cf (#58411) to branch-4.3, with one hunk dropped and one sentence reworded; everything else is byte-identical to what landed on master.

PlanMerger no longer merges two plans that read different sets of columns from a shared V1 file relation when the rows that relation returns depend on which columns the read asked for. DataSourceUtils.isProjectionSensitiveRead answers that question for a HadoopFsRelation, and merge compares, per shared relation and per occurrence of it, the columns each side reads before it tries to merge them. The two records have to match exactly rather than one containing the other: a cache entry's record stays true of it only because every plan merged in read the same columns, so admitting a narrower plan would make the record stale. Reads of the same columns still merge, so plain reuse is untouched.

Two things make a read projection-sensitive. Its parser may resolve or validate a column against the set of columns it was asked for, which lets a wider read drop or rewrite rows the narrower one returned, or return different values for a column it was already reading: CSV, JSON and XML build their parser from the required schema and take mode and the corrupt-record column from it, and Avro under positionalFieldMatching pairs a column with the Avro field at its position in that schema. Or the read may not be strict, in which case a failure in a column that only the wider read touches is swallowed together with the rest of that file's rows, whatever the format. The strictness half becomes FileSourceOptions.hasStrictFileReads, which this lifts out of FileScanRDD so that the cache-repeatability check in InMemoryRelation shares it, and it is evaluated per merge rather than cached, so a relation built before ignoreCorruptFiles was set still answers for the read that is running.

Two differences from the master commit, both because SPARK-57205 (#58340) is not on this branch:

• The FileTable.scala hunk is dropped. On master that hunk points FileTable.hasStrictFileReads at the shared predicate; here FileTable has no such method, and no built-in file table declares the SCAN_MERGING capability, so a V2 file scan is never merged on this branch and needs no gate. The predicate's callers here are FileScanRDD, InMemoryRelation and the new method, which is exactly what its scaladoc names.
• The scaladoc sentence about the Avro case calls SPARK-59108 a proposal rather than a landed fix, and names the arm it would retire, since that fix is on no branch and a maintenance branch should not carry an instruction whose precondition may never hold.

Why are the changes needed?

Top-level column pruning for a V1 file source happens in physical planning, from the attributes referenced above the relation (FileSourceStrategy computes readDataColumns from filterAttributes ++ projects), so two LogicalRelations over the same files canonicalize equal whatever each side projects. PlanMerger's identical-plan path therefore reuses one of them and the union of the two column sets is formed one level up, which means one subquery's result can depend on what a sibling subquery projects. Measured on this branch with the gate forced off, which is what 4.3.0 does; each cell holds the two subqueries' values, and the second column is what this head returns:

shape	merged	not merged
mode=DROPMALFORMED, a record malformed only in b: SELECT (SELECT sum(a) FROM t), (SELECT sum(b) FROM t)	[8, 80]	[10, 80]
PERMISSIVE with _corrupt_record in the schema: count(_corrupt_record) beside sum(b)	[1, 80]	[0, 80]
FAILFAST with a CSV row carrying fewer tokens than the schema has columns	throws	[10, 80]
spark.sql.files.ignoreCorruptFiles=true, parquet, b written as a string and read as a long: sum(a) beside count(b)	[null, 0]	[45, 0]

The bug is old rather than new, and 4.3.0 shipped it. Measured with the first two shapes on the other maintenance branches as well: on branch-4.2, where the rule is catalyst.optimizer.MergeSubplans, DROPMALFORMED answers [8, 80] against [10, 80] with the rule excluded and PERMISSIVE answers [1, 80] against [0, 80]; on branch-3.5, where it is MergeScalarSubqueries, the two-subquery query answers [8, 80] and [1, 80] while the first subquery run on its own answers 10 and 0. branch-4.0 and branch-4.1 carry the same rule as 3.5 and were not run. This patch does not reach those branches: from 4.2 down the rule lives in sql/catalyst, which cannot see HadoopFsRelation, so they would need a different seam.

Does this PR introduce any user-facing change?

Yes, and on a maintenance branch that is worth spelling out.

A query with two subqueries over the same CSV, JSON or XML relation that project different columns, over an avro relation read with positionalFieldMatching, or over any file relation under ignoreCorruptFiles or ignoreMissingFiles, now returns what two separate scans return. That is what the same query returned before subplan merging learned to merge it, and what 4.3.0 does not return, so the values those shapes answer with change in 4.3.1.

The cost is one extra scan for those shapes. MergeSubplans runs unconditionally, and the gate declines merging for any V1 file relation whose reads are not strict, which includes parquet and orc under ignoreMissingFiles where the scaladoc itself concedes there is no correctness mechanism, only predicate parity with the reader. Users who have either flag on therefore lose scan sharing they had in 4.3.0. The direction is safe, since declining a merge cannot change an answer, but it is the part most likely to be noticed.

Everything else keeps merging, including two subqueries over the same CSV relation that read the same columns.

How was this patch tested?

New suite FileSourceV1PlanMergingSuite, 17 tests, and one test in AvroV1Suite, both as they landed on master; the whole test diff is byte-identical to b82f9872d1cf. The four shapes above, with DROPMALFORMED covered for csv, json and xml alike; a read that is not strict on both configurations, with the temp view built outside the configuration scope so that a cached answer would fail the test; a self join, where each of the two reads of the relation has to be compared on its own; a third subquery that reads a column the merged pair does not, and a fourth that joins the entry the refused third one opened; and six shapes that must keep merging, so that the gate is not simply switching merging off.

Every test asserts the columns each FileSourceScanExec in the plan reads, rather than a scan count: the count depends on which scans physical reuse hid behind a leaf node, while the columns are the property this change is about. Finding a FileSourceScanExec at all is also what pins these tests to the V1 path.

Run on this branch: FileSourceV1PlanMergingSuite 17 of 17, the AvroV1Suite case, and catalyst/scalastyle, sql/scalastyle, sql/Test/scalastyle and avro/Test/scalastyle. Turning isProjectionSensitiveRead to false here fails 11 of the 17, with the values in the table above, and the 6 that pass are the ones whose only assertion is that a merge still happens; that is also where the table's numbers come from. The remaining mutation checks and the multi-subquery sweeps behind the design are on #58411 and were not re-run, since the code under them is identical.

What was re-verified for this branch rather than carried over: the optimizer reruns ColumnPruning after the MergeSubplans batch, tryMergePlans pairs relation occurrences in plan order, every configuration the suite sets exists here with the same default, MergeSubplans is excludable so the avro test's expected value really is the unmerged one, and javap shows the def-to-val change on FileScanRDD retains one boolean rather than the options map.

Was this patch authored or co-authored using generative AI tooling?

Generated-by: Claude Code

[uros-b]

LGTM, thank you @LuciferYang!

[LuciferYang]

Merge Summary:

• merged into branch-4.3 04cdb66

Posted by merge_spark_pr.py

[LuciferYang]

Thank you @dongjoon-hyun @uros-b