[SPARK-57205][SQL] Declare SCAN_MERGING on the built-in file sources

[LuciferYang]

What changes were proposed in this pull request?

FileTable gains a supportsScanMerging seam, and ParquetTable, OrcTable, TextTable and AvroTable override it. Those four then take part in the DSv2 scan merging added by #57360 (SPARK-40259): PlanMerger drives V2ScanRelationPushDown.rebuildScan to rebuild a merged scan, and a source supplies no merge logic of its own. It only declares that widening the set of columns pruned on its builder, with the scan options and pushed filters held constant, changes neither which rows the scan returns nor the values it returns for the columns it was already asked for; it may at most surface a read error. TableCapability.SCAN_MERGING's javadoc stated only a determinism contract, which a CSV table satisfies as written, since its rows are fully determined by the pruned column set and that is exactly the dependence, so this monotonicity criterion is now stated there too, where a connector author will read it.

CSVTable and JsonTable do not override it. Their parsers are handed the columns the scan asked for and decide from that set what counts as a malformed record, so a merged scan reading the union of two column sets can drop or rewrite rows the narrower scan returned. Measured, csv and json alike: with mode=DROPMALFORMED and a record malformed only in the other subquery's column, sum(a) is 8 where two separate scans give 10; with PERMISSIVE, the default, and _corrupt_record in the schema, the column is populated for a row the narrow scan counted as clean; with FAILFAST and a CSV row carrying fewer tokens than the schema has columns, the merged scan throws where the unmerged one returned rows. Those numbers come from the V1 path, which merges all three shapes today and is where SPARK-59107 (#58411) fixes them. The seam defaults to false, because a format that does not merge misses an optimization while a format that merges when its parser is projection-sensitive returns wrong rows.

Two further gates keep the contract true of the formats that do declare it.

FileTable withholds the capability when spark.sql.files.ignoreCorruptFiles or spark.sql.files.ignoreMissingFiles is set, matching FileScanRDD.hasStrictFileReads on the physical side. Under a non-strict read a failure in a column that only the sibling subquery projects is swallowed and the rest of that file's rows go with it, so the merged scan returns fewer rows than the narrow one did. Measured on parquet, V1 and V2 alike: SELECT (SELECT sum(a) FROM t), (SELECT count(b) FROM t) over data whose b is written as a string and read as a long returns [45, 0] with merging off and [null, 0] with it on. The gate is evaluated per call rather than cached, so a table built before the configuration was set still answers for the read that is running.

AvroTable withholds it under positionalFieldMatching. AvroPartitionReaderFactory builds the deserializer from the pruned read schema while the Avro side stays the full Avro schema, so under that option catalyst field i of the projection takes Avro field i of that schema, and widening the projection changes the values a column comes back with. That is a bug in its own right, filed as SPARK-59108 (#58409), which lands after this one and deletes this gate with it. ORC needs no equivalent gate: OrcUtils.requestedColumnIds maps both the _col* case and orc.force.positional.evolution through dataSchema.fieldIndex(name) and disables pruning in that branch, so its positional path is projection-independent.

The rest is a new test suite, a package-private V2ScanMergingTestHelper shared with DSv2PlanMergingSuite, one test in AvroV2Suite, and one documentation update.

Landing order: #58411 (SPARK-59107) first, then this one, then #58409 (SPARK-59108), which deletes the AvroTable gate. Nothing here depends on the first step, since this PR no longer asserts anything about how V1 merges CSV and JSON. Once #58411 is in, FileTable.hasStrictFileReads should call the FileSourceOptions.hasStrictFileReads it adds rather than spell the predicate out a third time, beside FileScanRDD and the cache-repeatability check in InMemoryRelation.

Why are the changes needed?

Two scans of the same file table that differ only in their projected columns cannot be reused today. A file source folds its data filters into the FileScan object, where they are used to list files and prune row groups, and FileScan.equals compares them, so two subquery scans over the same path are not canonically identical and PlanMerger's identical-plan fast path does not fire. On the V1 path those filters sit in a Filter above an identical LogicalRelation, so it does fire. Declaring the capability closes most of that gap for the formats where the merge is sound.

On TPC-DS at scale factor 100, with spark.sql.sources.useV1SourceList cleared, two queries change: q9 goes from 16 distinct scans to 6 under the default configuration and to 2 with dsv2SymmetricFilterPropagation on, and q28 from 6 to 1 with that configuration on. Of the 99 v1.4 queries, 95 ran and the other 93 of those are unchanged. Wall clock for q9 went from 78.1s to 40.3s by default and to 21.6s with the configuration on, and q28 from 61.3s to 34.4s; measured on local[1] with AQE off, two runs per configuration, against a build with the overrides removed as the baseline. Two samples on one machine put the noise around 20%, so the scan counts are the reproducible part and the timings show the order of magnitude. Both queries read store_sales, which is parquet, so leaving CSV and JSON out does not affect these numbers. Four queries could not be measured here because DataSourceV2Relation.computeStats raises a testing-only assertion when stats are read before pushdown, and q30 does not run.

Does this PR introduce any user-facing change?

Yes, on the V2 file source read path, which a format reaches only when it is removed from spark.sql.sources.useV1SourceList.

Plan shape only: subqueries over the same Parquet, ORC, text or Avro table that differ only in their projected columns now collapse into a single scan reading the union of those columns. No query result changes, which is what the exclusions and the two gates above are for, and why this needs no migration-guide entry. The dsv2SymmetricFilterPropagation entry in docs/sql-performance-tuning.md said "no built-in source does", which this PR falsifies. The four formats and the two withholding rules now sit in the prose of the Merging Subplans section rather than in that entry, since they hold whatever the configuration is set to, and the prose sentence that said the leaves must read the same input is qualified there for a source that declares the capability.

Four gaps are left in place. The first three are follow-ups #57360 already lists.

1. Two scans with different partition filters do not merge, while V1 merges them. A partition filter is fully enforced by the V2 scan and reported as strict, so widening it to OR would leave the merged scan returning rows nothing above it filters out.
2. Parquet and ORC nested columns do not merge while nested schema pruning is on. Each side narrows the struct to the field it reads, so the read column is no longer a same-type subset of the relation's column. V1 does not merge this shape either, because SchemaPruning rewrites each side's dataSchema and the two relations stop being canonically equal.
3. Differing data filters need spark.sql.optimizer.mergeSubplans.filterPropagation.dsv2SymmetricFilterPropagation.enabled, which defaults to false. On the V2 path that configuration alone is enough, where V1 needs the broader symmetricFilterPropagation.
4. CSV and JSON get no merging at all, which is deliberate rather than a follow-up. It costs them two shapes: scans whose projected columns differ, and scans reading the same columns under different filters, since FileScan.equals compares the normalized data filters so those are not canonically equal either and do not fall back to plain reuse. The V1 path merges both today and answers [8, 80] where two separate scans answer [10, 80], so one subquery's result depends on what a sibling subquery projects. That is a V1 bug rather than a target to copy; SPARK-59107 ([SPARK-59107][SQL] Do not widen a projection-sensitive V1 file read #58411) fixes it, and with that in both read paths decline these shapes alike.

How was this patch tested?

New suite FileSourceV2PlanMergingSuite, 16 tests. Every built-in file table is checked on the side it belongs to; a non-strict read is checked to withhold the capability on both strictness configurations, and separately to keep its scans separate and return [45, 0], with the table and the temp view built outside the configuration scope so that a cached gate would fail the test. Those are two tests rather than one so that a mutation names which half it broke, the capability assertion having aborted before the rows could report. Scans differing only in projected columns merge for Parquet and ORC and decline for CSV and JSON in the same shape; text merges the one shape a single-column table can differ in, which also needs both aggregates to be hash-aggregatable or PlanMerger.supportedAggregateMerge declines above the scans; scans over the same partition filter merge with that filter still enforced on the rebuilt scan; three scans merge into one; differing data filters merge only with the dsv2 configuration on, and the merged scan is checked to carry the OR-widened predicate rather than only to exist. Declines are covered too: differing partition filters, nested-pruned columns, a pushed aggregate (with the aggregate itself asserted, not just the scan count), and two different tables holding different rows so that a cross-table merge would change the answer.

Then V1/V2 parity on three shapes and the partition-filter gap pinned as the one shape where they disagree, both on parquet, whose reads this suite pins strict and whose parser does not depend on the projection, so SPARK-59107 leaves them merging on V1. The CSV and JSON parse behaviour is pinned on the V2 side alone, on three shapes: DROPMALFORMED gives [10, 80], PERMISSIVE with _corrupt_record in the schema gives [0, 80], and FAILFAST with a short CSV row returns rows rather than throwing. Subquery counts are asserted alongside the rows, so the result is attributed to the decline rather than inferred from the values, and the configurations the expectations depend on are pinned rather than assumed. These three used to assert the V1 numbers beside the V2 ones, and those arms pinned exactly what SPARK-59107 removes, so they are gone; its own suite covers that side. Measured at #58411's head, with this suite copied in: the CSV and JSON test passes there, and so does the parquet partition-filter one.

Every test asserts which read path the plan took before asserting anything about merging. SQL-on-file and catalog tables resolve to the V1 FileFormat regardless of spark.sql.sources.useV1SourceList, so a suite driven by that configuration alone would quietly run V1 on both sides.

One test in AvroV2Suite, because AvroTable lives in the module that has it on the classpath: it asserts the capability, that positionalFieldMatching withholds it, and that two scans differing only in their projected columns fuse into one reading the union.

Mutation checks. Turning the three sql/core overrides to false fails 8 of the 16 tests; of the 8 that pass, 6 assert a decline, one asserts that V1 merges differing partition filters where V2 does not and V2 declines under the mutation too, and one asserts that a non-strict read keeps its scans separate, which the mutation also produces. Dropping && hasStrictFileReads from capabilities fails both strictness tests, the capability one at its first assertion and the rows one with [null, 0] against [45, 0], which is what splitting them was for.

Regression: the planmerging suites, ExplainSuite and ExplainSuiteAE, FileBasedDataSourceSuite, FileTableSuite, OrcV2SchemaPruningSuite, ParquetV2SchemaPruningSuite, ParquetV2FilterSuite, SubquerySuite, SameResultSuite, ParquetV2AggregatePushDownSuite, OrcV2AggregatePushDownSuite, DataSourceV2Suite and AvroV2Suite.

No golden file or PlanStabilitySuite plan needed regenerating. Those build their tables with CREATE TABLE ... USING <format>, which resolves to the V1 FileFormat, so none of them reaches a V2 file scan.

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

Generated-by: Claude Code

[dongjoon-hyun]

Code review findings from an 8-angle review (correctness, removed-behavior, cross-file trace, reuse, simplification, efficiency, altitude, conventions) with an adversarial verification pass: 8 CONFIRMED, 2 PLAUSIBLE. Posted inline below, most severe first. The mechanical checks all came back clean (expected-answer arithmetic, SQLConf key names, the MergeSubplans excludedRules advice, line lengths, non-ASCII), and the production hot path gains no overhead (the capability is consulted only after two canonically-equal merge candidates are found).

[dongjoon-hyun]

[correctness / CONFIRMED] CSV/JSON in non-PERMISSIVE parse modes violate SCAN_MERGING's documented contract, yet the capability is declared unconditionally. TableCapability.java:144 promises the merged scan reads "a superset of their rows", but with DROPMALFORMED the merged scan parses the union of columns and drops a record malformed only in the other subquery's column, returning fewer rows than the original scan — the new suite's own comment concedes this, and pins sum(a) flipping from 10 to 8 depending on whether an unrelated sibling subquery triggers a merge. Today's only consumer (PlanMerger) survives because each side re-filters above the scan, but a future SCAN_MERGING consumer relying on the superset premise (e.g. reusing a wider scan for a narrower query without re-checking) would silently drop rows for CSV/JSON. CSVTable/JsonTable have the parse mode in hand, so the capability could be withheld when mode is not PERMISSIVE — or at minimum the exception should be carved out in the capability javadoc and this comment rather than living only in a test comment and a migration-guide sentence.

[LuciferYang]

Thanks, this is a real gap, and you have put it more precisely than the suite's own comment does. Agreed on the mechanism: with DROPMALFORMED the merged scan parses the union of the columns, so a record malformed only in the other subquery's column is dropped for both, and the merged scan returns fewer rows than the narrower scan did. That is not a superset.

On withholding the capability when mode is not PERMISSIVE, I measured what that would buy first. Same query as the suite's, CSV and JSON, all three modes, with MergeSubplans on and excluded:

mode	merged	not merged
PERMISSIVE	[10, 80]	[10, 80]
DROPMALFORMED	[8, 80]	[10, 80]
FAILFAST	throws	throws

The V1 path also returns [8, 80] under DROPMALFORMED; the suite asserts it, and it is why the migration guide says the V2 path now matches V1. Gating on the parse mode would leave the same query returning 10 on V2 and 8 on V1, so it trades a violated javadoc premise for a result that depends on spark.sql.sources.useV1SourceList. I would rather the two paths agree.

If the [8, 80] result is itself the thing to fix, that is a question about both paths rather than about this capability: V1 returned 8 before this PR too, and this PR does not touch it. I am happy to take that as a follow-up, either as one change covering both paths or as a fix to V1 that this PR then aligns to. If you would rather see V1 fixed first, I can hold this one until that lands.

PERMISSIVE keeps the fields it did parse, so widening the parsed columns changes nothing there. FAILFAST throws whether or not the scans merged, because the merged scan reads the union of the two column sets and the scan that reads b already throws on its own. Both are now assertions in the suite rather than claims in a comment.

Your second suggestion is the one I would like to act on: the exception belongs in the capability's own javadoc, not only in a test comment and a migration-guide sentence. TableCapability is @Evolving public API and that sentence arrived with #57360, so I would rather not reword it unilaterally.

cc @peter-toth, who wrote #57360 and the plan-merging work under it. Does the "superset of their rows" sentence need a qualifier for sources whose parsing depends on which columns are read, and would you rather that land here or in a follow-up? I am happy either way.

[peter-toth]

Finding 6.@LuciferYang, answering your question directly: no, I would not put a qualifier on "a superset of their rows".

That clause is not decoration, it is the soundness argument. The merge is only a rewrite because a projection and a filter above the merged scan can recover each original scan's result, and that needs the merged scan to still contain every row the narrow scan had. A source that can lose rows when a column joins the projection does not have the property. A qualifier admitting that would mean "declaring this capability permits Spark to change your results", and nothing downstream could rely on the premise again - which is the risk @dongjoon-hyun's comment is about.

Two facts have moved since you decided to keep CSV and JSON in, both measured on this head, both in my reply on the migration-guide thread: PERMISSIVE also changes (with columnNameOfCorruptRecord in the schema, count(_corrupt_record) goes 0 to 1 on csv and json), and FAILFAST also changes (a short CSV row makes the merged query throw where the unmerged one returned rows). So this is not one unusual mode. It reaches the default mode, and it can turn a working query into an error.

So I would take @dongjoon-hyun's shape, and one step further:

• FileTable.CAPABILITIES goes back to EnumSet.of(BATCH_READ, BATCH_WRITE), with a CAPABILITIES_WITH_SCAN_MERGING beside it and a protected def supportsScanMerging: Boolean = false seam choosing between them (see my comment on the new scaladoc).
• ParquetTable, OrcTable, TextTable and AvroTable override it to true. CSVTable and JsonTable do not.

What that buys: the javadoc stays true of every source that declares the capability; no third-party FileTable subclass is opted into a contract its author never saw; and the migration-guide entry can go away, because no user-visible result change is left.

On your objection that a seventh format would then silently not participate - I think that is the right way round. A new format that does not merge is a missed optimization, visible as a slower query. A new format that merges while its parser is projection-sensitive is a wrong answer, visible as nothing. The default should fail in the cheap direction, and the four supportsScanMerging = true overrides sit next to formatName and fallbackFileFormat, which a new format has to fill in anyway.

The honest counter-argument, because it is a real one: this leaves V2 CSV and JSON merging less than V1 for column-only-differing scans, part of the gap this PR set out to close. My read is that V1's behaviour here is a bug rather than a target, since it makes one subquery's result depend on what a sibling subquery projects, and matching a bug is worse than a fourth documented gap. If you would rather fix V1 first and align V2 to it after, as you offered above, that order works for me too. What I would avoid is weakening the capability's contract so that the current shape becomes in-spec.

[LuciferYang]

@peter-toth thanks for measuring this, and sorry for the wrong table above. Both shapes reproduce on this head, on V1 as well as V2:

shape	merged	MergeSubplans excluded
PERMISSIVE, _corrupt_record in schema, csv and json	[1, 80]	[0, 80]
FAILFAST, short CSV row	throws	[10, 80]

My table measured one malformed shape and I generalized it. The FAILFAST reasoning I gave only covers malformedness belonging to a column; a token-count mismatch is a property of how many columns are parsed, and that is what the merge changes. With the short row the a-only scan parses one column against one token, so nothing is malformed until the merge makes it two.

That removes my objection. I weighed one non-default mode against V1 parity; it is a different trade when the default mode changes results and FAILFAST can turn a working query into an error. Your point that the superset clause is the soundness argument rather than decoration is well taken.

Finding 1 is right too: V1 returns [8, 80] as well, so falling back to useV1SourceList restores nothing. Only excluding MergeSubplans does.

I will fix the guide sentence and the same claim in my summary comment, and come back on the shape of the fix including your supportsScanMerging seam.

[dongjoon-hyun]

[documentation / CONFIRMED + PLAUSIBLE] Two issues in this bullet:

1. It spells out the behavior change only for DROPMALFORMED, but the same widened-parsed-columns mechanism changes results in the other two modes as well (FailureSafeParser.scala:41-84): in PERMISSIVE mode, a record malformed only in the other subquery's column flips columnNameOfCorruptRecord from null to populated (and nulls that row's other fields); in FAILFAST mode, a query that previously succeeded now throws malformedRecordsDetectedInRecordParsingError — the harshest flip, with no explicit warning here. Consider enumerating all three modes or saying "with any parse mode".

2. The clause "merging there now matches what the V1 path already did" mildly overclaims: the PR's own test "V1 merges differing partition filters, V2 does not" pins a shape the suite itself labels "Known gap against V1". A qualifier like "for these shapes" would keep a skimming reader from over-generalizing.

[peter-toth]

Finding 2. Adding measurements to your first point, @dongjoon-hyun - both mode claims hold. The rebuttal above missed them because its data only exercises a type error inside one column.

PERMISSIVE, with columnNameOfCorruptRecord in the schema. Schema a long, b long, _corrupt_record string, mode=PERMISSIVE, query SELECT (SELECT count(_corrupt_record) FROM t WHERE a >= 0), (SELECT sum(b) FROM t WHERE a >= 0):

format	default	MergeSubplans excluded
csv, row 2,BAD	[1, 80]	[0, 80]
json, row {"a":2,"b":"BAD"}	[1, 80]	[0, 80]

The narrow scan parses only a, so the record is not corrupt for it. The merged scan parses b as well, so _corrupt_record is populated for a row the first subquery had counted as clean. FailureSafeParser.toResultRow sets the corrupt column whenever the raw parser threw, and which columns get parsed is exactly what the merge widens. Same on V1.

FAILFAST, with a short CSV row. Data Seq("0,0", "1,10", "2", "3,30", "4,40"), schema a long, b long, the suite's sum(a)/sum(b) query:

path	default	MergeSubplans excluded
V2	throws	[10, 80]
V1	throws	[10, 80]

Neither narrow scan is malformed here. With column pruning on, UnivocityParser.parsedSchema is the pruned schema and univocity is handed selectIndexes, so a one-token row matches a one-column parsed schema. The merged scan parses two columns, the tokens.length != parsedSchema.length branch fires (UnivocityParser.scala:406), and the query throws where it previously returned rows. That is the success-into-failure flip you described.

@LuciferYang, two consequences. The guide's "so neither of those modes changes" is not accurate as written, and the suite's intercept[SparkException](rows("FAILFAST", ...)) is two-sided only because 2,BAD makes the merged and the unmerged scan both throw - a short row would make it discriminating.

[dongjoon-hyun]

[altitude / CONFIRMED] Adding the capability to the shared FileTable.CAPABILITIES opts in every subclass at once — including out-of-tree file connectors (the class is public, capabilities() is non-final, and cross-module extension is real: connector/avro's AvroTable). But the determinism this comment cites is a property of the subclasses (newScanBuilder is abstract), which the base class cannot enforce. A third-party FileTable whose ScanBuilder is not deterministic per (options, filters, columns) silently inherits the contract on upgrade and gets its scans fused, with no opt-out other than discovering it must override capabilities(). Mitigating: org.apache.spark.sql.execution is nominally internal (blanket MiMa exclusion), so this is unsupported-but-common extension rather than stable API. Per-format declaration in the six built-in tables — or at least documenting the inherited contract in FileTable's scaladoc for subclass authors — would avoid the silent opt-in.

[dongjoon-hyun]

[correctness / PLAUSIBLE] The determinism justification here holds only per table instance: PlanMerger's gate is canonicalized-plan equality (PlanMerger.scala:680), the table case classes exclude the lazy fileIndex from equality, and InMemoryFileIndex.equals compares root paths only — so scans from two table instances whose listings were taken at different times (df1 read, files land, df2 read) are canonically equal, and the rebuilt merged scan reads one side's snapshot for both. This is largely an extension of long-standing rootPaths-only equality semantics (identical-column scans were already collapsed by the capability-free identical-plan path and ReuseSubquery/ReuseExchange, and V1 merges these shapes the same way), so no action beyond wording may be needed — but the comment overclaims as written; stating it as an instance-level property (or noting the cross-instance caveat) would keep it accurate.

[LuciferYang]

Thanks for tracing this through. The mechanism is right: canonical equality does not distinguish two table instances whose listings were taken at different times, because the table case classes exclude the lazy fileIndex and InMemoryFileIndex.equals compares root paths only.

On the wording, I read the comment as already staying inside instance scope: fileIndex is a lazy val "so every scan built from this table lists the same files", and the sentence after it is about rebuilding a scan from the same builder. I would rather not add a cross-instance caveat, for the reason you gave: two scans over the same paths reading the same columns were already collapsed before this change, by the identical-plan path and by ReuseSubquery/ReuseExchange, and V1 collapses them too. A caveat here would document InMemoryFileIndex's listing semantics rather than anything the capability introduces, and it would be easier to find next to that equality.

One note: the comment has moved since you read it. The contract now lives in a class-level scaladoc on FileTable addressed to subclass authors, and what is left at CAPABILITIES is the two facts that make it hold for the built-in tables. If the new text still reads as overclaiming, point at the clause and I will fix it.

[dongjoon-hyun]

[test-coverage / CONFIRMED]AvroTable inherits SCAN_MERGING through FileTable, but this test enumerates only parquet/orc/json/csv/text and nothing in connector/avro asserts the capability or exercises merging (grep for SCAN_MERGING/PlanMerging/MergeSubplans there returns zero hits). An Avro-specific merge defect — e.g. interaction of the union-column rebuild with positionalFieldMatching or Avro's filter pushdown — would ship enabled-by-default with zero coverage, while the migration guide tells users "the built-in file formats" are included. A small merging test in connector/avro (or at least a capability assertion there) would close the gap.

[dongjoon-hyun]

[test-coverage / CONFIRMED] The nestedPruning=false branch of this assertion is vacuous. With pruning off, both subqueries' scans read the identical whole struct s, and FileScan.equals (FileScan.scala:104-111) compares only fileIndex, readSchema and normalized filters — so even if the identical-plan merge were declined and two separate scan relations remained, they would canonicalize equal and distinctScans would still return 1. The comment says this branch verifies merging "on PlanMerger's identical-plan path", but a regression that stops that merge would go undetected. subqueryCounts (already defined in this suite) would actually distinguish: (1, 1) merged vs (2, 0) unmerged. The nestedPruning=true branch is fine — there the readSchemas differ, so == 2 is meaningful.

[dongjoon-hyun]

[simplification / CONFIRMED] This assert can never fire. All 4 call sites run after checkAnswer, which has already physically planned and executed every subquery — and a leaked bare DataSourceV2Relation has no batch-read physical strategy (DataSourceV2Strategy plans only DataSourceV2ScanRelation for reads), so planning would already have failed inside checkAnswer with QueryPlanner's "No plan for" assertion before this helper's friendlier message could be reached. It was copied from DSv2PlanMergingSuite, where the placement is equally post-checkAnswer and equally dead. Deleting the helper and its 4 calls loses nothing.

[dongjoon-hyun]

[reuse / CONFIRMED]v2Scans is a character-for-character copy of the private helper in DSv2PlanMergingSuite.scala:59-62, and assertNoPlaceholderRelation below duplicates its lines 69-75 — same test package. A future change to how merged V2 scans are collected (e.g. a new wrapper node) must land in both suites or they silently measure different things. A small shared package-private trait would collapse the duplication, and is also a natural home for distinctScans/subqueryCounts (the latter's measure is inlined eleven times in PlanMergingSuite).

[dongjoon-hyun]

[efficiency / CONFIRMED]rows(useV1 = false) is called in both asserts, and each call re-registers the temp view and re-executes the whole two-subquery query via collect() — one full redundant execution per format in CI. Binding val v2Rows = rows(useV1 = false) once and reusing it in both asserts eliminates the waste. Same pattern class elsewhere: withTempPath/writeFlat sit inside the dsv2Symmetric flag loop in the differing-data-filters test, writing identical data twice, where the nested-fields test already hoists the write outside its flag loop.

[LuciferYang]

Thanks for the review @dongjoon-hyun . Eight of the ten findings are addressed; the two on the FileTable comment have replies inline instead.

Fixed as suggested:

• The nestedPruning=false branch was vacuous, as you said. That arm now asserts subqueryCounts, (1, 1) merged against (2, 0) declined, and distinctScans == 2 stays only on the nestedPruning=true arm where the two readSchemas differ. The comment says why the old assertion could not tell the two apart.
• assertNoPlaceholderRelation and its four call sites are gone. Your reasoning holds: DataSourceV2Strategy's only batch-read case matches DataSourceV2ScanRelation, so a leaked bare relation fails planning inside checkAnswer before the helper runs.
• v2Scans no longer exists twice. It moved with distinctScans into a package-private V2ScanMergingTestHelper that both suites mix in. I left PlanMergingSuite's inlined subquery counting alone to keep this diff to the duplication you pointed at.
• The redundant executions are gone: the parse-mode test binds the result once, and the differing-data-filters test writes its data outside the flag loop, matching the nested-fields test.
• connector/avro has coverage now. A test in AvroV2Suite asserts AvroTable carries the capability and that two scans differing only in their projected columns fuse into one reading the union.
• The migration guide says "for these shapes" where it claimed to match V1, so the partition-filter gap the suite pins does not read as closed.

Adjusted rather than taken as written:

• On the shared CAPABILITIES: FileTable now has a class-level scaladoc stating the inherited contract for subclass authors, which was your fallback suggestion. I kept the single declaration rather than repeating it in the six built-in tables, because a seventh format added later would then silently not participate, and that drift is harder to notice than the inherited opt-in.
• On the parse modes: I measured all three rather than extending the guide to cover them. PERMISSIVE returns the same rows merged and unmerged, because the parser keeps the fields it did parse, and FAILFAST throws either way, because the merged scan reads the union of the two column sets and the scan reading the malformed column already throws on its own. The guide now records that neither mode changes instead of widening the warning, and both are assertions in the suite. Numbers are in the reply on the FileTable comment.

FileSourceV2PlanMergingSuite, DSv2PlanMergingSuite and PlanMergingSuite pass, 30 tests, along with the new Avro test and scalastyle on sql and connector/avro.

[LuciferYang]

also cc @cloud-fan

[peter-toth]

Thanks for the PR, @LuciferYang!

The one-line production change holds up for parquet, ORC, text and Avro: fileIndex is a lazy val and newScanBuilder is a fresh builder over mergedOptions(options), so what a scan reads really is a function of the filters pushed and the columns pruned. CSV and JSON are the exception, and a bigger one than the PR says. @dongjoon-hyun's parse-mode point is right and I measured it: PERMISSIVE and FAILFAST both change too, so the guide's "neither of those modes changes" is not accurate. The guide's first remedy is also wrong - going back to useV1SourceList restores nothing, because V1 merges these shapes identically, which your own suite asserts.

You asked me about the SCAN_MERGING javadoc, so: no, I would not qualify "a superset of their rows". That clause is the soundness argument for the rewrite, and weakening it turns the capability into "declaring this lets Spark change your results". I would rather CSV and JSON not declare it; full reasoning in my reply on @dongjoon-hyun's FileTable thread.

Blocking

• 1.Migration guide remedy restores nothing: "add the format back to spark.sql.sources.useV1SourceList" leaves both effects the entry describes in place, because V1 merges these shapes too. Measured [8, 80] on V1 with DROPMALFORMED, same as the new V2 behaviour; only excluding MergeSubplans gives the old [10, 80]. [inline: docs/sql-migration-guide.md:28]
• 2.PERMISSIVE and FAILFAST do change: both of @dongjoon-hyun's mode claims hold. The rebuttal above missed them because the test data only exercises a type error inside one column. Two measured shapes on his thread. [reply on @dongjoon-hyun's docs/sql-migration-guide.md thread]
• 3.Description no longer matches the change: "The rest is a new test suite and two documentation updates" misses V2ScanMergingTestHelper, the DSv2PlanMergingSuite refactor and the new AvroV2Suite test, and "How was this patch tested?" never mentions the Avro test. I also reproduced the 7-of-12 count, but the fifth still-passing test is "V1 merges differing partition filters, V2 does not", not one that "merges through the identical-plan fast path".

Non-blocking

• 4.The parity tests never assert which path they ran on:mergedCounts and the parse-mode rows helper both skip assertUsesFileSourceV2, so "V1 and V2 file sources merge the same subquery shapes" would pass with both arms on V2. The suite scaladoc claims every test checks this. [inline: sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/FileSourceV2PlanMergingSuite.scala:111]
• 5.The documented opt-out has no seam: this scaladoc tells a subclass to override capabilities to drop SCAN_MERGING, but FileTable.CAPABILITIES is private, so it must hand-write the base set and will silently drift when that set grows. [inline: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/FileTable.scala:46]

Alternatives

• 6.Declare SCAN_MERGING per built-in table and leave CSV and JSON out:@dongjoon-hyun's per-format point plus the two measurements that were missing when you decided against it. Also answers your javadoc question. [reply on @dongjoon-hyun's FileTable.scala thread]

Minor

• 7.distinctScans scaladoc overstates: "two means it was declined" is false for two identical scans, as the nested-fields test's own comment explains. [inline: sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/V2ScanMergingTestHelper.scala:37]

[peter-toth]

Finding 1. The first remedy does not restore the previous behaviour.

Both effects this bullet describes are already what V1 does. V1 merges two scans of the same relation that differ only in their projected columns, and the merged required schema is the union. So moving the format back into spark.sql.sources.useV1SourceList hands the user the new rows, not the old ones.

Measured on this PR's head, with the suite's own data (Seq("0,0", "1,10", "2,BAD", "3,30", "4,40"), schema a long, b long, mode=DROPMALFORMED) and the suite's two-subquery query:

path	default	MergeSubplans in excludedRules
V2	[8, 80]	[10, 80]
V1	[8, 80]	[10, 80]

The V1 column is already pinned two files over, at FileSourceV2PlanMergingSuite.scala:476 (assert(rows("DROPMALFORMED", useV1 = true) == droppedV2)), so this sentence contradicts the test that ships with it.

Only the excludedRules remedy works. Suggest ending the bullet with just that, e.g. "To restore the previous behavior, disable subplan merging by adding org.apache.spark.sql.execution.planmerging.MergeSubplans to spark.sql.optimizer.excludedRules; falling back to spark.sql.sources.useV1SourceList does not help, because the V1 path merges these shapes the same way."

[peter-toth]

Finding 4. This helper never checks which read path it ran on, so the parity claim is asserted but not measured.

The suite scaladoc says "every test goes through DataFrameReader and asserts the plan is V2 before asserting anything about merging". mergedCounts and the rows helper in the parse-mode test are the two that do not. In "V1 and V2 file sources merge the same subquery shapes" the only assertions are v1 == v2 and v1 == ((1, 1)); if USE_V1_SOURCE_LIST ever stopped taking effect for a format, both arms would run V2, both would return (1, 1), and the test would pass while measuring nothing about V1.

"V1 merges differing partition filters, V2 does not" is safe already, because its two arms assert different values. Only this one needs the check:

valdf= sql(query)
checkAnswer(df, expected)
if (useV1) assertUsesFileSourceV1(df) else assertUsesFileSourceV2(df)
subqueryCounts(df)

with assertUsesFileSourceV1 the one-line mirror of the existing helper (collectWithSubqueries { case r: LogicalRelation => r }.nonEmpty, and no DataSourceV2ScanRelation). Worth calling assertUsesFileSourceV2 in the V2 arm of the parse-mode rows too.

[peter-toth]

Finding 5. The opt-out this paragraph prescribes cannot reuse the base set.

FileTable.CAPABILITIES is private to the companion, so a subclass following this advice has to write util.EnumSet.of(BATCH_READ, BATCH_WRITE) from scratch. It then silently drops whatever the base set gains later: the next capability added to FileTable would reach every built-in format but not that subclass, and nothing would fail.

A seam on the class makes the opt-out one line, keeps the base set in one place, and is testable:

/** Whether this table meets the `SCAN_MERGING` contract described above. */protecteddefsupportsScanMerging:Boolean=trueoverridedefcapabilities: java.util.Set[TableCapability] =if (supportsScanMerging) FileTable.CAPABILITIES_WITH_SCAN_MERGINGelseFileTable.CAPABILITIES

This is also the seam finding 6 would need, if you take that route.

[peter-toth]

Finding 7. "two means it was declined" is not true in the direction that matters.

A decline between two scans that read the same columns still canonicalizes to one distinct scan, which is exactly the trap the nested-fields test documents at its nestedPruning=false arm. Reading it as stated is how that arm became vacuous in the first place. Suggest: "one distinct scan is consistent with a merge, two means it was declined; two identical scans canonicalize equal either way, so use subqueryCounts when the columns match."

[LuciferYang]

Thanks, this is all taken. The change is now what you proposed: FileTable.CAPABILITIES no longer carries SCAN_MERGING, a CAPABILITIES_WITH_SCAN_MERGING sits beside it, and a protected def supportsScanMerging: Boolean = false seam picks between them. ParquetTable, OrcTable, TextTable and AvroTable override it; CSVTable and JsonTable do not. Pushed as 7dd1833.

1. The migration-guide entry is gone rather than corrected. With CSV and JSON out there is no user-visible result change left to document.
2. Conceded above, and now pinned rather than argued. The parse-mode test asserts all three of your shapes on both read paths: DROPMALFORMED[10, 80] on V2 against [8, 80] on V1, PERMISSIVE with _corrupt_record[0, 80] against [1, 80], and FAILFAST with the short CSV row returning rows on V2 and throwing on V1.
3. The description is rewritten against the current diff, including the helper, the DSv2PlanMergingSuite change and the Avro test. Your correction to the count stands: rerun on the current suite it is 6 of 13 plus the Avro test, and of the 7 that still pass, 4 assert a decline, one asserts different values on the two paths, and two are the new CSV and JSON tests.
4. Taken. assertUsesFileSourceV1 is the mirror you sketched, and both mergedCounts and the parse-mode helper call one or the other. I also took the subqueryCounts suggestion for rows, which earned itself: asserting before collect() reported (2, 0) on V1, because AQE had not finalized and the reuse was not yet in the plan, so it moved after.
5. Taken as written, including the name.
6. Taken. One line added to the criterion, because the FAILFAST measurement taken alone proves more than it should: a corrupt column chunk, or datetimeRebaseModeInRead=EXCEPTION meeting an ancient value, makes any format throw on a column the narrower scan had pruned, so "merging can surface a new error" would empty the list. What separates CSV and JSON is the silent and unrecoverable change to row membership and row content, and FAILFAST is that same projection dependence surfacing as an error. The four override comments say it that way, so a later reader does not ask why Parquet is not out too.
7. Taken, with your wording.

That leaves the V1 side, which I will file separately rather than sequence this behind. Fixing it needs a signal on FileFormat plus plumbing so the decline can see whether the projections differ, which the leaf arm cannot: only MergeContext threads down and both leaves carry the full schema. It also changes behaviour users have today. Your reading that V1 is the bug rather than the target is what the fourth gap in the description now says.

[peter-toth]

Re-checked through 7dd18334 - findings 1, 2, 3, 4, 5 and 7 resolved, 6 taken as the design, nothing regressed. I reproduced the description's new counts in both directions: 6 of 13 plus the Avro test fail with the four overrides flipped to false, and 3 of 13 fail if CSVTable and JsonTable declare it. The seam and the exclusion are both pinned.

The new criterion has one hole, and it is the example you used to justify it. A corrupt column chunk stays an error only while spark.sql.files.ignoreCorruptFiles is false. With it on the error is swallowed and the rest of that file is dropped, so the merged scan reads fewer rows than the narrow one. That is the same silent change to row membership that took CSV and JSON out.

Blocking

• 8.ignoreCorruptFiles breaks the superset premise (new): with spark.sql.files.ignoreCorruptFiles=true, a read failure in a column only the sibling subquery projects is swallowed, and sum(a) goes from 45 to null on the V2 parquet path. FileTable should withhold the capability unless the read is strict, the way InMemoryRelation already does. [inline: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/FileTable.scala:124]

Non-blocking

• 9.AvroTable's stated mechanism is not what the reader does (new):AvroDeserializer is built from readDataSchema, not the full schema, so under positionalFieldMatching=true the projection decides which Avro field feeds which column. Measured [10, 10] unmerged against [10, 1000] merged, which also falsifies "No query result changes". [inline: connector/avro/src/main/scala/org/apache/spark/sql/v2/avro/AvroTable.scala:56]
• 10.Gap 4 mis-states what CSV and JSON lose (new): "Scans whose columns match are unaffected: PlanMerger falls back to plain reuse for canonically equal scans" - but FileScan.equals compares the normalized data filters, so two CSV scans reading the same one column with different filters are not canonically equal. Measured (2, 0) subquery counts there where parquet gets (1, 1), and that merge would be sound, since the parser is handed the same column set on both sides.

Minor

• 11.The four overrides drop protected (new): an override with no modifier is public in Scala, so the seam is protected on FileTable and public on the four that declare it. [inline: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/parquet/ParquetTable.scala:76]
• 12.SPARK-57205's description is stale (new): it still says the change declares the capability on FileTable so "Parquet, ORC, CSV, JSON, text, Avro" take part, and lists two out-of-scope items where the PR now lists four gaps.

[peter-toth]

Finding 8. The criterion holds only while the read is strict.

Your reply on @dongjoon-hyun's thread justifies the four overrides with "a corrupt column chunk, or datetimeRebaseModeInRead=EXCEPTION meeting an ancient value, makes any format throw on a column the narrower scan had pruned". That is true only with spark.sql.files.ignoreCorruptFiles=false. With it on, FilePartitionReader.next catches any RuntimeException or IOException and returns false, dropping the rest of that file (FilePartitionReader.scala:74, via DataSourceUtils.shouldIgnoreCorruptFileException). So the merged scan returns fewer rows than the narrow scan did. That is the "silent and unrecoverable change to row membership" you gave as what separates CSV and JSON, reached by a format that declares the capability.

Measured on this head, parquet, spark.sql.files.ignoreCorruptFiles=true. b is written as a string and read as a long, so the reader throws only when it actually reads b; a corrupt column chunk is the same code path, just harder to build in a test.

spark.range(0, 10).selectExpr("id AS a", "cast(id AS string) AS b").write.parquet(path)
spark.read.schema("a long, b long").parquet(path).createOrReplaceTempView("t")
sql("SELECT (SELECT sum(a) FROM t), (SELECT count(b) FROM t)")

path	default	MergeSubplans excluded
V2	[null, 0]	[45, 0]
V1	[null, 0]	[45, 0]

sum(a) touches only healthy data and is correct at 45 today; after the merge it is null. V1 is the same, so this is V1 parity rather than a V1/V2 split - but it is the parity you declined to copy for CSV and JSON, on the same reasoning. ORC did not reproduce with this particular type mismatch; the exposure is any per-column read failure, so it is not parquet-specific.

The fix has a precedent in the tree: InMemoryRelation refuses to treat a file scan as repeatable under either flag (InMemoryRelation.scala:378-398, and FileScanRDD.hasStrictFileReads).

overridedefcapabilities: java.util.Set[TableCapability] =if (supportsScanMerging && hasStrictFileReads) {
FileTable.CAPABILITIES_WITH_SCAN_MERGING
} else {
FileTable.CAPABILITIES
}
/** * Whether a read of this table is strict. A best-effort read is not reproducible: a failure on a * column only the other scan projects is swallowed, so the merged scan would not read a superset * of either input's rows.*/privatedefhasStrictFileReads:Boolean= {
valfileSourceOptions=newFileSourceOptions(options.asCaseSensitiveMap.asScala.toMap)
!fileSourceOptions.ignoreCorruptFiles &&!fileSourceOptions.ignoreMissingFiles
}

FileSourceOptions is already imported here and it resolves the per-read option over the session conf, so both spellings are covered. Only ignoreCorruptFiles has the failure above - a missing file drops the same rows whatever is projected - so including ignoreMissingFiles is just matching the existing predicate; drop it if you prefer the narrower gate.

If you would rather keep the capability unconditional, then the migration-guide entry has to come back for this shape. Unlike the errors the criterion accepts, this one is a silent result change.

[peter-toth]

Finding 9. The first sentence is not what AvroPartitionReaderFactory does.

The deserializer is built from readDataSchema, i.e. the projection, not the full schema (AvroPartitionReaderFactory.scala:101-111). positionalFieldMatching is passed straight through, and AvroUtils.AvroSchemaHelper.getAvroField then resolves a catalyst field by its position in that projected schema (AvroUtils.scala:463-469). So the projection is exactly what decides which Avro field feeds which column, and widening it re-maps them.

Measured on this head. File fields a, b, c holding id, 100 * id, 10000 * id for ids 0 to 4, read with positionalFieldMatching=true, query SELECT (SELECT sum(a) FROM t), (SELECT sum(c) FROM t):

path	default	MergeSubplans excluded
V2	[10, 1000]	[10, 10]
V1	[10, 1000]	[10, 10]

sum(c) is 100000. Both values are wrong, because Avro positional matching against a pruned schema is already broken without this PR: the c-only scan resolves position 0 and reads a. Merging cannot turn a correct answer into a wrong one here either, since the union of two prefixes is the longer prefix. So this is not a correctness regression, and I am not asking you to exclude Avro.

Two smaller asks. State the mechanism as it is - what makes merging safe for Avro is that the format has no record-level parse verdict, not that decoding ignores the projection - so a later reader does not build on a property the reader does not have. And "No query result changes" in the description needs a qualifier, or the pruning bug needs its own JIRA to point at.

@dongjoon-hyun named positionalFieldMatching as the Avro-specific risk on the coverage thread above; the new AvroV2Suite test does not reach it.

[peter-toth]

Finding 11. This override widens the seam from protected to public.

FileTable.supportsScanMerging is protected, and an override with no access modifier is public in Scala. So ParquetTable, OrcTable, TextTable and AvroTable each expose it as public API while CSVTable and JsonTable keep it protected. Either add the modifier to the four, or drop it from the base if the seam is meant to be part of the format-author surface.

Suggested change

	overridedefsupportsScanMerging:Boolean=true
	overrideprotecteddefsupportsScanMerging:Boolean=true

[dongjoon-hyun]

To @peter-toth , if you don't mind, could you ask your agent to exclude the pre-existing comments or avoid to mention the other GitHub Account IDs more than once? It's because it seems that I received too many pings from your comments while you are talking with others (not me). And, please use direct hyperlinks to the comments instead of referring like someone's thread because there are multiple threads in the single PR.

[peter-toth]

To @peter-toth , if you don't mind, could you ask your agent to exclude the pre-existing comments or avoid to mention the other GitHub Account IDs more than once? It's because it seems that I received too many pings from your comments while you are talking with others (not me). And, please use direct hyperlinks to the comments instead of referring like someone's thread because there are multiple threads in the single PR.

Sorry @dongjoon-hyun for the pings and thanks for the feedback. I will adjust my review skill according that.

[LuciferYang]

Thanks @peter-toth. Finding 8 is taken, and it took Avro with it.

8.capabilities is now supportsScanMerging && hasStrictFileReads, with the predicate as you wrote it. I kept ignoreMissingFiles in it to match FileScanRDD.hasStrictFileReads on the physical side, and the gate is evaluated per call rather than cached, so a table built before the configuration was set still answers for the read that is running. The new test builds the table and the temp view outside the configuration scope for exactly that reason, covers both spellings, and asserts [45, 0] alongside two distinct scans.

9. The comment claimed the opposite of what the reader does, and it is rewritten around the property Avro does have: no record-level parse verdict. I also withheld the capability under positionalFieldMatching, so the contract holds for every source that declares it rather than only where a merge cannot make the answer worse. The option is read off the options map rather than through AvroOptions, whose constructor resolves avroSchemaUrl and would do I/O inside capabilities. The AvroV2Suite test now asserts the capability, the withholding under the option, and a projection-only merge. The underlying bug is SPARK-59108, with a runnable repro on the ticket, and the gate can go once it is fixed. ORC turned out to need no equivalent: OrcUtils.requestedColumnIds maps both positional cases through dataSchema.fieldIndex(name) and reports canPruneCols = false. On "No query result changes": with both gates in place it holds, so I dropped the migration-guide entry instead of qualifying the sentence.

10. Corrected. Gap 4 now says CSV and JSON lose both shapes, differing projections and equal columns under differing filters, with your reason: FileScan.equals compares the normalized data filters. The V1 side is SPARK-59107, which pins four measured shapes, the three parse modes plus ignoreCorruptFiles.

11. Fixed on all four.

12.SPARK-57205's description is rewritten: the seam, the two exclusions, the two gates, the four gaps.

Mutation checks on the current head: flipping the three sql/core overrides to false fails 8 of the 15 new tests, and dropping && hasStrictFileReads fails the strictness test at its first capability assertion. CI is green.

[peter-toth]

Re-checked through 60961fe4e1 - findings 8, 9, 10, 11 and 12 all resolved, nothing regressed. Re-measured at this head: 15/15 in the suite plus the Avro test, and flipping the three sql/core overrides to false fails 8 of the 15, with the 4+1+2 accounting of the seven survivors holding exactly. SPARK-57205's description now matches the diff.

I also got the measurement your mutation note says it could not. With && hasStrictFileReads dropped, the end-to-end half of the strictness test returns [null, 0] against Row(45, 0) and one scan against two, so that half is effective - it just never reports. That is finding 15.

Nothing blocking from my side. The four below are a javadoc fix, a sequencing decision against #58411, and two nits.

Non-blocking

• 13.The SCAN_MERGING javadoc admits exactly what this PR excludes (new): Its determinism clause is satisfied by a CSV table, so the "superset of their rows" claim below it does not follow. The criterion that separates them is the one you wrote in FileTable's scaladoc, and it belongs in TableCapability.java too. FileTable.scala:45
• 14.#58411 breaks this test's V1 arms (new): Measured on that head, this test fails at (2, 0) did not equal (1, 1) on the V1 side. Every useV1 = true expectation here pins behaviour #58411 removes, and gap 4 goes stale with them. FileSourceV2PlanMergingSuite.scala:658

Minor

• 15.The strictness test's first half masks its second (new): Split it in two so a mutation reports both. FileSourceV2PlanMergingSuite.scala:192
• 16.This predicate is the one #58411 lifts onto FileSourceOptions (new): Whichever lands second should call the shared method. FileTable.scala:148

Thanks for working through all twelve of the earlier findings, @LuciferYang - nothing left open from those rounds.

[peter-toth]

Finding 13. This is the right criterion, and it is stricter than the one TableCapability.SCAN_MERGING states. A third-party connector author reads only the javadoc.

TableCapability.java:134-139:

By returning this capability a table declares a determinism contract: holding the scan options constant, the rows and columns a scan reads are fully determined by the filters pushed via SupportsPushDownV2Filters and the columns pruned via SupportsPushDownRequiredColumns.

A CSV table satisfies that as written. Its rows are fully determined by the pruned column set - that is exactly the dependence, and re-pruning to the same set does yield an equivalent scan. So CSVTable could declare the capability without contradicting a word of it, and the next paragraph's "the merged scan reads ... a superset of their rows" then simply does not follow from what the table promised.

What closes the gap is the monotonicity clause you wrote here: widening the pruned set must not change the rows or the values. Suggest adding it to the javadoc, right after the determinism sentence:

 * Determinismaloneisnotenough: wideningthesetofprunedcolumns, withtheoptionsand
* pushedfiltersheldconstant, mustnotchangewhichrowsthescanreturnsnorthevaluesit
* returnsforthecolumnsalreadyaskedfor. Asourcewhoseparserdecideswhatcountsasa
* malformedrecordfromthesetofcolumnsitwasaskedfordoesnotmeetthis.

Different clause from the one at r3879483408 - there I said not to weaken "a superset of their rows", and this asks to strengthen the sentence above it so that claim actually follows. Fine as a follow-up on SPARK-40259 if you would rather not widen this diff, but the capability is @since 4.3.0 and unreleased, so it is cheaper now than after.

[peter-toth]

Finding 14. This test's V1 arms pin the behaviour #58411 removes, so whichever of the two lands second breaks.

Measured rather than reasoned: I copied this file and V2ScanMergingTestHelper.scala into a worktree at #58411's head and ran this one test.

- SPARK-57205: CSV and JSON decline to merge, so their parsing stays per subquery *** FAILED ***
format=csv: (2, 0) did not equal (1, 1) unexpected subquery counts on V1:

It aborts on this line, so the rest never report, but they all move with it: Row(8, 80) becomes Row(10, 80), Row(1, 80) becomes Row(0, 80), and the FAILFAST intercept[SparkException] finds nothing to catch. Six V1 expectations plus the throw.

There is a second casualty in the description. Gap 4 opens with "CSV and JSON do not merge, while V1 does" and rests on "V1 has merged both for years and answers [8, 80]". After #58411 both paths decline and the gap is closed, not open.

The order is yours to pick, but it is worth stating in the description either way:

• [SPARK-59107][SQL] Do not widen a projection-sensitive V1 file read #58411 first - then write these arms against the new V1 behaviour from the start, and gap 4 becomes "was a V1 bug, fixed by SPARK-59107".
• this first - then [SPARK-59107][SQL] Do not widen a projection-sensitive V1 file read #58411 has to come back and change them.

Either way the V1 arms stop being a V1/V2 contrast once both are in, so they are better dropped than inverted; the V2 assertions carry the whole point of the test on their own.

[peter-toth]

Finding 15. This test makes two independent claims - the capability flips with the conf, and the merge that flip prevents would return wrong rows - and the first aborts before the second can report. That is the gap your mutation note records.

I measured the second half on its own. With capabilities changed to if (supportsScanMerging), and the end-to-end block lifted into a scratch suite so nothing aborts first:

SCRATCH rows = [null,0]
SCRATCH distinctScans = 1

against Row(45, 0) and 2. So the assertion is effective and worth keeping; it is only unreportable where it sits.

Splitting the test in two - one on the capability, one on the rows - costs a withTempPath and makes the mutation name which half broke:

 test("SPARK-57205: withhold SCAN_MERGING from a table whose reads are not strict") { ... }
test("SPARK-57205: a non-strict read keeps its scans separate") { ... }

[peter-toth]

Finding 16.#58411 lifts this exact predicate onto FileSourceOptions.hasStrictFileReads and points both existing spellings at it, FileScanRDD and the cache-repeatability check in InMemoryRelation. Once both PRs are in, this is the only place still writing it out.

privatedefhasStrictFileReads:Boolean=newFileSourceOptions(options.asCaseSensitiveMap.asScala.toMap).hasStrictFileReads

Nothing to do now if this lands first - just worth a line in the description so the follow-up is not lost, since the scaladoc here already points at FileScanRDD.hasStrictFileReads as the matching predicate.

[LuciferYang]

Thanks, @peter-toth. All four are in at b6c4b7b748b, and CI is green on it.

13. In this PR rather than as a follow-up. I took your clause and added the second disqualifier from FileTable's scaladoc, so the javadoc now names both reasons a built-in source is held back here: a parser that decides what counts as a malformed record from the set of columns it was asked for, and a source that resolves a column by its position in the projection. Those two are the whole gap between the two texts, and the capability is unreleased, so it is cheaper here than after.

14. Dropped rather than inverted, as you suggested. rows loses its useV1 parameter, the three V1 expectations and the FAILFAST intercept go with it, and the comments now say what a merged scan would do instead of what V1 does. Gap 4 is rewritten around CSV and JSON getting no merging at all being deliberate, with the [8, 80] numbers attributed to the V1 path and to SPARK-59107 fixing them there, after which both paths decline these shapes alike. The order is in all three descriptions now: #58411, then this, then #58409.

One correction to the finding. "Every useV1 = true expectation here" does not hold for the two parquet ones: this suite pins ignoreCorruptFiles and ignoreMissingFiles to false in sparkConf, and parquet's parser does not depend on the projection, so #58411's gate never fires for them. Measured the way you did it, with this suite and the helper copied into a worktree at #58411's head: the CSV and JSON test passes there, and so does "V1 merges differing partition filters, V2 does not". So only the csv and json arms had to go, and the V1/V2 parity pair stays.

15. Split in two. Dropping && hasStrictFileReads now fails both, the capability one at its first assertion and the rows one with correct answer [45, 0] against Spark answer [null, 0], which is the number your scratch suite reported, now in a test that can report it. The three-overrides mutation is 8 of 16, and the eight that pass are six declines, the partition-filter test whose V2 side declines under the mutation too, and the new rows test, whose separate scans the mutation also produces.

16. In the description, since FileSourceOptions.hasStrictFileReads is not in master yet. Whichever way the two land, this ends up calling it.

Two more measurements on the sequencing, since #58409 belongs to the same set. Its five commits cherry-picked onto this head leave the AvroV2Suite case passing, so this PR and #58409 do not interact beyond the gate. And git merge-tree puts the only textual conflict among the three in connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala, between this one and #58411, both of which append a test there, so whichever lands second needs a rebase.

[peter-toth]

Re-checked through b6c4b7b7 - findings 13, 14, 15 and 16 resolved, nothing regressed. Your correction on 14 holds: this suite pins both strictness flags off in sparkConf and parquet's parser does not depend on the projection, so #58411's gate never fires for the two parquet arms and only the csv and json ones had to go.

Two new, both on the one paragraph in docs/sql-performance-tuning.md. Nothing in the code.

Non-blocking

• 17.The "Merging Subplans" prose still says the leaves must read the same input (new): that sentence was accurate while this cell ended with "no built-in source does", and the PR makes it false by default on the V2 file path. The format list and the two withholding rules hold whatever dsv2SymmetricFilterPropagation is set to, so they belong in the prose rather than inside a default-off configuration's cell. inline: docs/sql-performance-tuning.md:387

Minor

• 18.The shared "because" describes ignoreCorruptFiles only (new): a read failure swallowed "in a column that only the other scan projects" is not what ignoreMissingFiles does, as FileTable's own scaladoc says. Not reopening whether the flag belongs in the predicate. inline: docs/sql-performance-tuning.md:387

[peter-toth]

Finding 17. This paragraph is the only place the new behaviour is written down, and it sits in the cell of a configuration that is false by default. That configuration only adds the differing-data-filter shape. The projection-only merge and the two withholding rules hold whatever it is set to.

Meanwhile the prose a reader consults for when merging fires still states the old rule.

docs/sql-performance-tuning.md:343:

Two subplans are merged when their plans match node by node: ... and the leaves must read the same input.

Line 331 opens the section the same way ("subplans that return a single row and read the same input"). Both were accurate while this cell ended with "no built-in source does". After this PR two leaves that read different column sets merge, under the default configuration, for Parquet, ORC, text and Avro on the V2 read path.

Suggest moving the format list and the withholding rules up into the prose, after line 343:

On the DataSource V2 read path a source can go further and declare the `SCAN_MERGING` table capability, which lets two leaves that differ only in their projected columns merge into a single scan reading the union of those columns. Among the built-in file formats Parquet, ORC, text and Avro declare it; a format reaches its V2 read path only when it is removed from `spark.sql.sources.useV1SourceList`. A file table withholds the capability when `spark.sql.files.ignoreCorruptFiles` is true, because a read failure in a column that only the other subplan projects would then be swallowed along with the rest of that file's rows, and when `spark.sql.files.ignoreMissingFiles` is true, to match the strictness predicate the file reader uses. Avro withholds it under `positionalFieldMatching`, which resolves a column by its position in the projection.

and leaving this cell its first three sentences plus one line for what the configuration itself adds:

For a file source the strictly enforced filters are the partition filters, so this configuration is what lets two scans over the same partitions but with different data filters merge.

The ignoreMissingFiles wording above is finding 18's fix, folded in.

[peter-toth]

Finding 18. The "because" covers both flags as written, and only one of them does that.

A file table withholds the capability when spark.sql.files.ignoreCorruptFiles or spark.sql.files.ignoreMissingFiles is true, because a read failure in a column that only the other scan projects would then be swallowed along with the rest of that file's rows

ignoreMissingFiles swallows a FileNotFoundException (sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/FilePartitionReader.scala:51), which is file-level: the same rows go whatever either side projects. Your own scaladoc says it, FileTable.scala:143-145:

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

So the reason belongs to ignoreCorruptFiles, and the second flag needs its own half-clause - "and when spark.sql.files.ignoreMissingFiles is true, to match the strictness predicate the file reader uses". Not reopening whether the flag belongs in the predicate; that is settled.

[LuciferYang]

Thanks, @peter-toth. Both are in at 019cda6c095.

17. Moved as you suggest, with the entry keeping its first three sentences plus your line on what the configuration adds. I named the requirement instead of opening with "the last of those conditions", since #58411 inserts the V1 condition into that same paragraph and the reference would point past it. Line 343 is untouched, and git merge-tree still auto-merges the doc after the insertion, so the only conflict with #58411 stays AvroSuite.scala. Line 331 I left as the section's summary, with the exception in the paragraph that enumerates the conditions; say so if you would rather it carried a pointer.

18. Split, ignoreMissingFiles now reading "to match the strictness predicate the file reader uses". That agrees with FileTable.scala:143-145 and with #58411's wording.

[peter-toth]

Thanks @LuciferYang, nothing blocking from my end.

[LuciferYang]

Thanks for your thorough and detailed review. @peter-toth

[LuciferYang]

@dongjoon-hyun Could you take another look when you get a chance? Or should I merge this PR first?

[LuciferYang]

Merge Summary:

• merged into master 484866b
• merged into branch-4.x 69e1cb1

Posted by merge_spark_pr.py

[LuciferYang]

Thank you @peter-toth@dongjoon-hyun

Since 4.3 has entered the RC phase, this pr has been merged only to master and branch‑4.x. @dongjoon-hyun If you have any new review comments, feel free to leave them, and I will address them in a follow‑up PR.