LuciferYang opened a new pull request, #58411: URL: https://github.com/apache/spark/pull/58411
### What changes were proposed in this pull request? `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 decide what counts as a malformed record from the columns it was asked for, which lets a wider read drop or rewrite rows the narrower one returned; CSV, JSON and XML all build their parser from the required schema and take `mode` and the corrupt-record column from it. 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 is the same predicate as `FileScanRDD.hasStrictFileReads`, 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. The three formats are named in `DataSourceUtils` rather than declared by each `FileFormat`, the way `SchemaPruning.canPruneDataSchema` names Parquet and ORC. A capability method on `FileFormat` would have to pick a default: "safe to widen" repeats this same list as the overrides, and "not safe" stops merging for every format that does not override, including Hive's ORC reader and every third-party format, for no correctness gain. A third-party format whose parser is projection-sensitive therefore keeps merging, as it does today. #58340 (SPARK-57205) records this V1 behaviour as a gap on its V2 side, where the same four shapes are declined; with this in, the two read paths agree on them. ### 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 `LogicalRelation`s 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 head, all four shapes, `[merged, with MergeSubplans excluded]`: | 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 first three reproduce for csv, json and xml alike. The fourth is not about parsing, so it reaches every format. ### Does this PR introduce _any_ user-facing change? Yes. A query with two subqueries over the same CSV, JSON or XML relation that project different columns, or over any file relation under `ignoreCorruptFiles` or `ignoreMissingFiles`, now returns what two separate scans return, which is what the same query returned before subplan merging learned to merge it. The cost is one extra scan for those shapes. 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. 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; and eight shapes that must keep merging, namely parquet subqueries projecting different columns, csv subqueries reading the same columns, an identical pair of subqueries, which are reused rather than merged, a csv read where the two sides differ only in a partition column reference, which is not a column read at all, and a csv read the two sides merge through filter propagation, which rebuilds the projection above the relation. 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. Filter propagation is on by default, and it merges two subqueries over such a relation by rebuilding the projection above it. Measured on six query shapes with symmetric propagation on and off, twelve runs in all, the merged answers match the answers with `MergeSubplans` excluded and no scan reads a column neither side asked for, because the rebuilt projection is over the pruned child. One of those shapes is the test above. Beyond the suite, ten shapes with three or four subqueries over one relation, including a self join, two different views, a mix of csv and parquet, the corrupt-record column, a partitioned table and a subquery that reads no column at all, were run with symmetric filter propagation on and off, twenty runs in all: every one returns what the same query returns with `MergeSubplans` excluded. Four mutation checks. Turning `isProjectionSensitiveRead` to false fails 10 of the 17; the 7 that pass are the ones asserting that a merge still happens. Keeping one column set per relation instead of one per occurrence, which is what the first version of this did, fails only the self-join test, with `[84, 84]` where the fix gives `[85, 84]`. Keeping partition columns in the read set fails only the partition-reference test. Deriving the cached side's read set from the merged plan rather than from the record taken when it was cached fails only the three-subquery test, with `[14, 8, 88]` where three separate scans give `[18, 10, 88]`. Regression: `MergeSubplansSuite`, `PlanMergingSuite`, `DSv2PlanMergingSuite`, `SubquerySuite`, `DataFrameSubquerySuite`, `ReuseExchangeAndSubquerySuite`, `ExplainSuite`, `ExplainSuiteAE`, `FileBasedDataSourceSuite` and `InMemoryColumnarQuerySuite`, 419 tests; the csv, json and xml suites, 1185 tests; `catalyst/scalastyle`, `sql/scalastyle` and `sql/Test/scalastyle`. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude Code -- 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]
