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

[LuciferYang]

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 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, measured [10, 100] merged against [10, 10] unmerged. SPARK-59108 (#58409) removes that at the root, which retires the avro arm here along with the AvroTable gate #58340 adds on the V2 side; the landing order is below. 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 FileSourceOptions.hasStrictFileReads, which this PR lifts out of FileScanRDD and points the other two spellings of it at, InMemoryRelation's cache-repeatability check and the capability gate #58340 added to FileTable. 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 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), now in master, recorded 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.

Landing order: this one, then #58409, which deletes the avro arm, the org.apache.spark.sql.avro import it needs, and the AvroV1Suite test. A branch cannot delete code that is not in master yet, which is the only reason the order matters. Neither direction breaks the build: that test compares the merged answer against the same query with MergeSubplans excluded rather than against a literal row, and asserts one column per scan, both of which hold whatever sum(b) returns once #58409 changes it, so #58409 landing first leaves the arm over-conservative rather than turning a test red.

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 master, where all four shapes merge; 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 first two reproduce for csv, json and xml alike, and the third does too with a format-appropriate malformation, a token count being a CSV notion. The fourth is not about parsing, so it reaches every format.

The bug is not new. Measured on the maintenance branches with the same two csv shapes: 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 fix only reaches back to 4.3, though: from 4.2 down the rule lives in sql/catalyst, which cannot see HadoopFsRelation, so those branches would need a different seam.

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, over an avro relation read with positionalFieldMatching, 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; 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: parquet subqueries projecting different columns, csv subqueries reading the same columns, an identical pair of subqueries, which are reused rather than merged, a third subquery that reads the same columns as the merged pair, 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. One test in AvroV1Suite, because avro does not resolve from the sql/core test classpath; it pins both strictness configurations rather than inheriting them, so that positional matching is the only reason its read is projection-sensitive, and it takes its expected values from the same query with MergeSubplans excluded, because the value sum(b) returns is a property of what its own subquery projects and SPARK-59108 changes it.

Every test in the suite 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, each measured rather than inferred. Turning isProjectionSensitiveRead to false fails 11 of the 17 here and the avro test as well; the 6 that pass are the ones whose only assertion is that a merge still happens. Removing the avro case alone fails the avro test. 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 the three-subquery and four-subquery tests, with [14, 8, 88] and [14, 8, 88, 4] where separate scans give [18, 10, 88] and [18, 10, 88, 4].

Regression, re-measured after merging master rather than carried over: the whole planmerging package, which now includes #58340's FileSourceV2PlanMergingSuite, plus SubquerySuite, DataFrameSubquerySuite, ReuseExchangeAndSubquerySuite, ExplainSuite, ExplainSuiteAE, FileBasedDataSourceSuite and InMemoryColumnarQuerySuite, 12 suites and 439 tests; the datasources.csv, datasources.json and datasources.xml packages, 19 suites and 1718 tests; catalyst/scalastyle, sql/scalastyle, sql/Test/scalastyle and avro/Test/scalastyle.

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

Generated-by: Claude Code

[sunchao]

Reviewed 00b48e2 against 024120d with five independent review passes covering merge/cache invariants, read-column tracking and self-joins, parser/file-option semantics, optimizer/AQE/reuse interactions, and tests/compatibility. No actionable P1/P2 findings.

The per-occurrence read sets and preservation of the original cached signature cover the important self-join and multi-subquery cases, while retaining identical-plan reuse and compatible scan sharing.

Verified CI on this exact commit: 29 checks passed and 2 were skipped, with no failing or pending checks. The test report records 72,229 tests run, 1,100 skipped, and 0 failures. Kubernetes integration was skipped. I did not independently execute the tests locally; this approval is based on source review and the verified CI results.

CI run · Test report

LGTM.

[peter-toth]

Thanks for the PR, @LuciferYang!

V1 top-level column pruning happens in physical planning, so two LogicalRelations over the same files canonicalize equal whatever each side projects, and PlanMerger is the only place that can notice. Recording the read column sets per occurrence, and keeping the record taken when the entry was first cached rather than re-deriving it from a merged plan, is what makes the self join and the third-subquery cases come out right. The suite passes 17/17 on 00b48e2 here.

One thing to fix before merge: docs/sql-performance-tuning.md spells out when two subplans merge, and it gains a condition here that it says nothing about. Below that, AvroFileFormat is projection-sensitive under positionalFieldMatching and is not on the list, so that shape still merges and still changes the answer.

Blocking

• 2.The merge conditions in the tuning doc are now incomplete:docs/sql-performance-tuning.md:343 enumerates when two subplans merge and does not learn the new condition. This is a default-on optimization that now silently does not fire, and #58340 updates the same file for its half. [inline: sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/PlanMerger.scala:192]

Non-blocking

• 1.V1 avro still widens a projection-sensitive read:AvroFileFormat builds its deserializer from the pruned requiredSchema, so under positionalFieldMatching a column's values depend on the projection, measured [10, 100] merged against [10, 10] unmerged. Not a regression, since positional matching against a pruned schema is already wrong, so this is a silent result change rather than correct-to-wrong. [inline: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSourceUtils.scala:193]
• 3.Mutation-check counts in the description: with isProjectionSensitiveRead forced to false the suite reports 11 failed and 6 succeeded, not "10 of the 17" and "the 7 that pass". "Eight shapes that must keep merging" does not line up either, since 6 tests assert a merge still happens and the sentence names 5 of them.

Alternatives

• 4.Name the formats by short name rather than by class:supportNestedPredicatePushdown, twenty lines above, answers the same kind of question with hs.toString against a SQLConf list. That would cover avro and any third-party format without the three imports. [inline: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSourceUtils.scala:192]

Minor

• 5.Dead first clause: two empty maps are already ==, so the nonEmpty disjunction never changes the result. [inline: sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/PlanMerger.scala:287]
• 6.referenced rebuilt per occurrence: it is the same set every time, so a plan with N sensitive relations walks itself N times. [inline: sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/PlanMerger.scala:298]
• 7.options is now a field on FileScanRDD: reading it from a method body retains the constructor parameter, so it goes into every task closure. [inline: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileScanRDD.scala:94]
• 8.Three missing blank lines in the new suite. [inline: sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/FileSourceV1PlanMergingSuite.scala:89]
• 9.This scaladoc claims the swallowing mechanism for both strictness flags: only ignoreCorruptFiles has it, and #58340's FileTable.hasStrictFileReads says so. [inline: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSourceUtils.scala:177]

[peter-toth]

Finding 1.AvroFileFormat.buildReader hands the pruned requiredSchema to AvroDeserializer, and under positionalFieldMatching catalyst field i takes Avro field i of the full file schema: AvroUtils.AvroSchemaHelper.getAvroField returns avroFieldArray.lift(catalystPos) (sql/core/src/main/scala/org/apache/spark/sql/avro/AvroUtils.scala:462). Widening the read therefore changes the values the columns already being read come back with.

Measured on 00b48e24, on the default configuration (avro is in the spark.sql.sources.useV1SourceList default, so this is the V1 path):

spark.range(0, 5).selectExpr("id AS a", "id * 10 AS b").write.format("avro").save(path)
spark.read.option("positionalFieldMatching", "true").format("avro").load(path)
.createOrReplaceTempView("t")
sql("SELECT (SELECT sum(a) FROM t), (SELECT sum(b) FROM t)")
// merged [10, 100] one scan reading {a, b}// MergeSubplans excluded [10, 10] two scans, {a} and {b}

Non-blocking because merging cannot turn a correct answer into a wrong one here. Pruning keeps the fields' relative order, so a pruned schema is a subsequence of the full one, and in the union of two subsequences every retained field's index moves toward its true index without passing it. Both [10, 10] and [10, 100] are wrong against the file, which is SPARK-59108. What the merge does change is that one subquery's values depend on what its sibling projects, and that is the invariant this PR is about.

#58340 withholds SCAN_MERGING from AvroTable under this option, so the two read paths still disagree on it. AvroFileFormat lives in sql/core under org.apache.spark.sql.avro, so the module argument in the description does not cover it and this object can name it directly:

private[sql] defisProjectionSensitiveRead(relation: BaseRelation):Boolean= relation match {
casehs: HadoopFsRelation=>!newFileSourceOptions(hs.options).hasStrictFileReads ||
hasProjectionSensitiveParser(hs.fileFormat, hs.options)
case _ =>false
}
privatedefhasProjectionSensitiveParser(
fileFormat: FileFormat, options: Map[String, String]):Boolean= fileFormat match {
case_: CSVFileFormat|_: JsonFileFormat|_: XmlFileFormat=>true// Positional matching resolves each catalyst field against the Avro field at the same// position, so pruning shifts which Avro field a column reads. See SPARK-59108.case_: AvroFileFormat=>CaseInsensitiveMap(options).get(AvroOptions.POSITIONAL_FIELD_MATCHING).exists(_.toBoolean)
case _ =>false
}

Naming AvroFileFormat unconditionally would do too, at the cost of every avro merge.

[peter-toth]

Finding 2.docs/sql-performance-tuning.md:343 lists what two subplans have to agree on to merge, ending with "and the leaves must read the same input". This line adds a condition that is not there. It is a default-on optimization that now silently does not fire for a whole class of relations, so the tuning page is where someone will look, and #58340 updates the same file for its half of this.

Something like this, appended to that paragraph:

A V1 file relation whose rows depend on the columns the read asked for is merged only when both subplans read the same columns of it. That covers csv, json and xml, whose parsers decide what counts as a malformed record from the required schema, and any file relation read with spark.sql.files.ignoreCorruptFiles or spark.sql.files.ignoreMissingFiles set, where a failure in a column only one side reads is swallowed together with the rest of that file's rows.

[peter-toth]

Finding 4.supportNestedPredicatePushdown at line 158 of this file answers the same kind of question by short name. hs.toString is the format's registered short name, so a list covers every format the same way, including the ones this object cannot import:

privatedefhasProjectionSensitiveParser(hs: HadoopFsRelation):Boolean= {
valsensitive=Utils.stringToSeq(
SQLConf.get.getConf(SQLConf.PROJECTION_SENSITIVE_FILE_SOURCE_LIST).toLowerCase(Locale.ROOT))
sensitive.contains(hs.toString) // "csv", "json", "xml", "avro"
}

That drops the three imports, covers avro (finding 1), and lets someone add a third-party format they know to be sensitive.

The counter-argument is real. NESTED_PREDICATE_PUSHDOWN_FILE_SOURCE_LIST gates an optimization, so shortening it costs performance, while shortening this one returns wrong rows. A hard-coded Set("csv", "json", "xml", "avro") of short names keeps the module independence without that, and is what I would take if you do not want a new conf.

[peter-toth]

Finding 5. Two empty maps are already ==, so this clause never changes the result. When both are empty the comparison is false anyway, and when exactly one is empty the comparison is true and so is the disjunction.

privatedefwidensProjectionSensitiveRead(
reads: Map[LogicalPlan, Seq[Set[String]]],
cachedPlan: MergedPlan):Boolean= reads != cachedPlan.projectionSensitiveReads

[peter-toth]

Finding 6.referenced does not depend on relation, but it is rebuilt for each occurrence, so a plan with N projection-sensitive relations walks itself N times and builds N AttributeSets. Computing it once at the only caller makes the independence visible too:

privatedefcollectProjectionSensitiveReads(
plan: LogicalPlan):Map[LogicalPlan, Seq[Set[String]]] = {
lazyvalreferenced=AttributeSet(plan.flatMap(_.references)) ++AttributeSet(plan.output)
plan.collect {
casel: LogicalRelationifDataSourceUtils.isProjectionSensitiveRead(l.relation) =>
l.canonicalized -> readColumnNames(referenced, l)
}.groupMap(_._1)(_._2)
}

[peter-toth]

Finding 7. Reading options from a method body turns the constructor parameter into a field, so every FileScanRDD now carries a FileSourceOptions into the task closure. javap -p on the two builds:

base private final boolean ...FileScanRDD$$ignoreCorruptFiles;
private final boolean ...FileScanRDD$$ignoreMissingFiles;
head private final org.apache.spark.sql.catalyst.FileSourceOptions options;
(plus both booleans)

Tiny in bytes, since parameters is @transient. Still, the two vals right above already hold what this needs, so the dedup can keep the parameter constructor-scoped:

Suggested change

	private[sql] defhasStrictFileReads:Boolean= options.hasStrictFileReads
	private[sql] valhasStrictFileReads:Boolean= options.hasStrictFileReads

[peter-toth]

Finding 8. Missing blank line after scanColumns. Same at the two other places where a Seq/test follows a closing brace: FileSourceV1PlanMergingSuite.scala:140 and :199.

[peter-toth]

Finding 9. This attributes the swallowing to both halves of hasStrictFileReads, and only ignoreCorruptFiles has it. A missing file is skipped whatever is projected, so it cannot make a read projection-sensitive. Your own FileTable.hasStrictFileReads in #58340 says exactly that:

ignoreMissingFiles drops the same rows whatever is projected, and is included to match FileScanRDD.hasStrictFileReads, the same predicate on the physical side.

Worth carrying that sentence over here, because this PR is what makes hasStrictFileReads shared: FileSourceOptions now answers it for the cache-repeatability question in InMemoryRelation, for the reader in FileScanRDD, and for this one, and only here is one of the two flags carried for consistency rather than for a mechanism. No behaviour change asked for.

[LuciferYang]

Thanks @peter-toth. Finding 2 is fixed and finding 1 took a gate with it.

1. Confirmed, and the module argument in my description was wrong: AvroFileFormat and AvroOptions are both private[sql] in sql/core, so this object can name them. hasProjectionSensitiveParser now takes the options and answers true for avro under positionalFieldMatching, read off the map rather than through AvroOptions, whose constructor resolves avroSchemaUrl and would do I/O here, and read leniently so a malformed value still fails where Avro reports it rather than here. The test is in AvroV1Suite, because avro does not resolve from the sql/core test classpath. SPARK-59108 (#58409) removes the sensitivity at the root, and then this case and #58340's AvroTable gate come off together.

2. Added to the paragraph that enumerates the conditions, naming the three parsers, avro under the option, and both strictness flags.

3. Both numbers were arithmetic rather than measured: I added the seventeenth test and adjusted the counts instead of re-running. Re-measured, 11 of the 17 fail and 6 pass, and the sentence about shapes that keep merging now names the six it means. A pass over the rest of the description turned up more of the same class, all corrected: the table legend collided with its own header, the FAILFAST shape is CSV-specific rather than one of three, the mutation on the cached record fails two tests rather than one ([14, 8, 88] and [14, 8, 88, 4]), and the enumeration of the suite accounted for 15 of its 17 tests.

4. I kept the class match. hs.toString is the registered short name and falls back to HadoopFiles for a format without a DataSourceRegister, so a list keyed on it fails silently in the direction that returns wrong rows, while a missing class case cannot be a typo. With avro named, every in-tree projection-sensitive format is covered, and a third-party one is covered by neither. If you would rather have the configuration so that a user can add one, I will take it.

5. Gone. 6. Computed once at the caller and passed in. 7.val rather than def, which also stops options being retained as a field, so the serialized RDD is back to what master ships plus one boolean. 8. Fixed. 9. Fixed, with your sentence from #58340 carried over.

Locally: the planmerging suites, SubquerySuite, ExplainSuite, ExplainSuiteAE, FileBasedDataSourceSuite and InMemoryColumnarQuerySuite, 423 tests; the whole avro module, 487; catalyst/scalastyle, sql/scalastyle, sql/Test/scalastyle and avro/Test/scalastyle; and the four mutation checks re-measured rather than inferred.

[peter-toth]

Re-checked through 0942427984 - findings 1, 2, 3, 5, 6, 7, 8 and 9 resolved, nothing regressed. I re-ran the suite at this head: 17/17 plus the avro test, and with isProjectionSensitiveRead forced to false 11 of the 17 fail and the avro test with them, so the description's counts hold. javap -p on this build shows no options field on FileScanRDD, only the boolean, so finding 7 landed as you describe.

Finding 4 I withdraw. HadoopFsRelation.toString:61 really does fall back to HadoopFiles for a format with no DataSourceRegister, so a short name is not an identity and a list keyed on it can only be set too coarsely. On your offer of a configuration: I would not add one. It inherits the same broken key, and a wrong value there returns wrong rows rather than losing an optimization. If a third-party format ever needs this, a FileFormat method defaulting to false is the extension point, and it preserves today's behaviour so it can go in on demand.

Nothing blocking from my side. Both below are small, and 10 is a sequencing decision rather than a defect in this diff.

Non-blocking

• 10.The avro case and its test collide with #58409 (new): Measured at #58409's head, that query returns [10, 100] merged and unmerged, so Row(10L, 10L) fails there and the avro case becomes dead. Say in the description which lands first and what the other one deletes. connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala:3747

Minor

• 11.MergedPlan's record has an unused default (new): Both construction sites pass it, and = Map.empty is the value that lets a merge through. sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/PlanMerger.scala:68

Thanks for working through all nine of the round-1 findings, @LuciferYang - nothing left open from that round.

[peter-toth]

Finding 10. Your description says #58409 "removes that at the root, and this case can go with it". It takes this test with it too, and not by making it redundant - by failing it.

I ran this exact query in a worktree at #58409's head a3358708e60:

merged = [10, 100]
unmerged = [10, 100]

sum(b) on its own is 100 there, because the fix resolves b against its position in the data schema rather than in the projection. So Row(10L, 10L) fails, and the avro arm of hasProjectionSensitiveParser is dead at the same moment - the read stops being projection-sensitive, which is exactly what those two numbers being equal shows.

Whichever of the two lands second pays for it, so it is worth deciding now and saying so in the description. Two ways out:

• Land [SPARK-59108][SQL] Fix Avro positional matching under column pruning #58409 first and drop the avro arm plus this test from here. That also drops the only reason DataSourceUtils imports org.apache.spark.sql.avro.
• Keep them, land this first, and have [SPARK-59108][SQL] Fix Avro positional matching under column pruning #58409 delete both.

If you would rather keep the test through the transition, the assertion that survives is the one the sql/core suite already uses - the columns each scan reads, which stay Seq(Seq("a"), Seq("b")) under the gate whatever sum(b) evaluates to. AvroV1Suite would need spark.sql.adaptive.enabled off for that, for the reason FileSourceV1PlanMergingSuite:369 gives.

[peter-toth]

Finding 11. Nothing uses the default. MergedPlan(plan, false, projectionSensitiveReads) at line 210 is the only construction, and the two other sites are mp.copy(...), which carries the record over.

The value it defaults to is also the permissive one: an entry built without a record compares equal to any plan that reads no projection-sensitive relation, so the merge goes through. Dropping the default makes a future construction site say what it means.

caseclassMergedPlan(
plan: LogicalPlan,
merged: Boolean,
projectionSensitiveReads: Map[LogicalPlan, Seq[Set[String]]])

[LuciferYang]

Thanks for re-running it, @peter-toth. Both new items are in at efdd55d4f13.

10. I kept the test and kept a value assertion with it, by taking the expected row from the same query with MergeSubplans excluded instead of from a literal [10, 10]. It then says what this PR guarantees, that merging does not change the answer, and it survives the [10, 100] you measured on both sides at #58409's head. With isProjectionSensitiveRead forced to false it fails with correct answer [10, 10] against Spark answer [10, 100], which also rules out a vacuous comparison: one taken from a still-merged run would pass. It asserts the scan columns as well, with spark.sql.adaptive.enabled off as you said, and I checked that with it on the collection is empty rather than merely different. The two strictness flags are pinned rather than inherited, so positional matching is the only thing making that read projection-sensitive.

The order is now in all three descriptions: this one, then #58340, then #58409, which deletes the avro arm, the org.apache.spark.sql.avro import DataSourceUtils needs for it, and this test. It only matters because a branch cannot delete code that is not yet in master; with the assertions above, neither direction turns anything red. #58340's AvroV2Suite case needs nothing: it asserts the capability, the withholding under the option, and a data query with the option off, and none of the three depends on how a pruned positional read resolves.

11. Dropped. 210 is the only construction site and the other two are mp.copy, so nothing else moved.

4. Agreed, no configuration. If a third-party format ever needs one, the FileFormat method you describe is where it belongs.

[uros-b]

+1, thank you @LuciferYang and @sunchao@peter-toth!

[LuciferYang]

Merge Summary:

• merged into master b82f987
• merged into branch-4.x d9176bf

Posted by merge_spark_pr.py

[LuciferYang]

Thank you @peter-toth@uros-b@sunchao

There were conflicts against branch‑4.3, so I will need to open a separate PR manually for the fix.