peter-toth commented on code in PR #58340:
URL: https://github.com/apache/spark/pull/58340#discussion_r3879475838
##########
docs/sql-migration-guide.md:
##########
@@ -25,6 +25,7 @@ license: |
## Upgrading from Spark SQL 4.3 to 4.4
- Since Spark 4.4, for storage-partitioned joins,
`spark.sql.requireAllClusterKeysForCoPartition` requires every join key to be
covered by some partition key instead of matching the partition keys
positionally. As a result, a join-key column partitioned by more than one
transform no longer prevents shuffle elimination, and
`spark.sql.sources.v2.bucketing.allowKeysSubsetOfPartitionKeys.enabled` no
longer additionally requires `spark.sql.requireAllClusterKeysForCoPartition` to
be `false` when the join keys are a subset of the partition keys. As before,
when the partition keys cover only part of the join keys, eliminating the
shuffle still requires `spark.sql.requireAllClusterKeysForCoPartition` to be
`false`.
+- Since Spark 4.4, the built-in file formats declare the `SCAN_MERGING` table
capability on their DataSource V2 read path, so two scans of the same file
table that differ only in their projected columns can be merged into a single
scan reading the union of those columns. A format takes that path only when it
is removed from `spark.sql.sources.useV1SourceList`, and merging there now
matches what the V1 path already did. For CSV and JSON this also changes which
records count as malformed, because the parser is handed only the columns the
scan reads: with `mode` set to `DROPMALFORMED`, a record malformed only in the
columns the other scan reads is now dropped for both. To restore the previous
behavior, add the format back to `spark.sql.sources.useV1SourceList`, or
disable subplan merging entirely by adding
`org.apache.spark.sql.execution.planmerging.MergeSubplans` to
`spark.sql.optimizer.excludedRules`.
Review Comment:
**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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
