[
https://issues.apache.org/jira/browse/SPARK-59107?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Yang Jie resolved SPARK-59107.
------------------------------
Fix Version/s: 4.3.0
Resolution: Fixed
Issue resolved by pull request 58472
[https://github.com/apache/spark/pull/58472]
> Subplan merging must not widen the columns a projection-sensitive parser reads
> ------------------------------------------------------------------------------
>
> Key: SPARK-59107
> URL: https://issues.apache.org/jira/browse/SPARK-59107
> Project: Spark
> Issue Type: Bug
> Components: SQL
> Affects Versions: 5.0.0
> Reporter: Yang Jie
> Assignee: Yang Jie
> Priority: Major
> Labels: pull-request-available
> Fix For: 4.3.0
>
>
> On the DataSource V1 file source path, subplan merging can widen the set of
> columns a scan reads, and for CSV and JSON that changes which records the
> parser treats as malformed. One subquery's result then depends on what a
> sibling subquery projects.
> Reproduction, spark-shell on default configurations, so every read below is
> V1. Each shape is run twice, once as-is and once with subplan merging
> switched off, and every value is measured.
> {code}
> import java.nio.file.{Files, Paths}
> def merging(on: Boolean): Unit =
> spark.conf.set("spark.sql.optimizer.excludedRules",
> if (on) "" else "org.apache.spark.sql.execution.planmerging.MergeSubplans")
> val csv = "/tmp/spark-v1-merge-csv"
> Files.createDirectories(Paths.get(csv))
> Files.write(Paths.get(csv, "data.csv"),
> "0,0\n1,10\n2,BAD\n3,30\n4,40".getBytes)
> // 1. DROPMALFORMED: the record malformed in b is dropped for sum(a) too.
> spark.read.schema("a long, b long").option("mode", "DROPMALFORMED").csv(csv)
> .createOrReplaceTempView("t1")
> val q1 = "SELECT (SELECT sum(a) FROM t1), (SELECT sum(b) FROM t1)"
> merging(true); sql(q1).show() // [8, 80]
> merging(false); sql(q1).show() // [10, 80]
> // 2. PERMISSIVE, the default mode, with the corrupt-record column in the
> schema.
> spark.read.schema("a long, b long, _corrupt_record string").option("mode",
> "PERMISSIVE")
> .option("columnNameOfCorruptRecord", "_corrupt_record").csv(csv)
> .createOrReplaceTempView("t2")
> val q2 = "SELECT (SELECT count(_corrupt_record) FROM t2 WHERE a >= 0), " +
> "(SELECT sum(b) FROM t2 WHERE a >= 0)"
> merging(true); sql(q2).show() // [1, 80]
> merging(false); sql(q2).show() // [0, 80]
> // 3. FAILFAST with a row carrying fewer tokens than the schema has columns.
> val short = "/tmp/spark-v1-merge-short"
> Files.createDirectories(Paths.get(short))
> Files.write(Paths.get(short, "data.csv"), "0,0\n1,10\n2\n3,30\n4,40".getBytes)
> spark.read.schema("a long, b long").option("mode", "FAILFAST").csv(short)
> .createOrReplaceTempView("t3")
> val q3 = "SELECT (SELECT sum(a) FROM t3), (SELECT sum(b) FROM t3)"
> merging(true); sql(q3).show() // throws MALFORMED_RECORD_IN_PARSING
> merging(false); sql(q3).show() // [10, 80]
> // 4. ignoreCorruptFiles, parquet: not about parsing, so this reaches every
> format.
> val pq = "/tmp/spark-v1-merge-parquet"
> spark.range(0, 10).selectExpr("id AS a", "cast(id AS string) AS b")
> .write.mode("overwrite").parquet(pq)
> spark.conf.set("spark.sql.files.ignoreCorruptFiles", "true")
> spark.read.schema("a long, b long").parquet(pq).createOrReplaceTempView("t4")
> val q4 = "SELECT (SELECT sum(a) FROM t4), (SELECT count(b) FROM t4)"
> merging(true); sql(q4).show() // [null, 0]
> merging(false); sql(q4).show() // [45, 0]
> {code}
> Shapes 1 to 3 are CSV, and JSON behaves the same way with equivalent records.
> In shape 1 the scan that reads only a never parses b, so the record malformed
> in b is not dropped for sum(a) until the merge widens the read set. In shape
> 2 the merged scan parses b, so the corrupt-record column is populated for a
> row the first subquery had counted as clean. In shape 3, with column pruning
> on, UnivocityParser.parsedSchema is the pruned schema, so a one-column scan
> matches a one-token row and only the merged two-column scan trips the
> tokens.length != parsedSchema.length branch. In shape 4 sum(a) touches only
> healthy data and is correct at 45 until the merge widens the read set;
> FilePartitionReader.next and FileScanRDD both route the failure through
> DataSourceUtils.shouldIgnoreCorruptFileException, which accepts any
> RuntimeException, so the rest of that file's rows go with it.
> Mechanism. There is no scan-merge operation on the V1 path. Two
> LogicalRelations over the same paths are canonically equal regardless of
> projection, because top-level column pruning happens in physical planning
> (FileSourceStrategy computes readDataColumns from filterAttributes ++
> projects), so PlanMerger's identical-plan path reuses one of them and the
> union of the columns is produced one level up in mergeNamedExpressions. The
> planmerging package has no per-source predicate on the V1 path; the
> SCAN_MERGING capability gate in PlanMerger applies only to DSv2 scans.
> Scope. Declining the merge needs two things. First, a signal reachable from
> HadoopFsRelation.fileFormat saying the format's parsing depends on the
> projected columns; DataSourceUtils.supportNestedPredicatePushdown and
> SchemaPruning.canPruneDataSchema are the closest existing shapes. Second, the
> decline point has to see whether the projections actually differ: the most
> local seam, a (LogicalRelation, LogicalRelation) arm in
> PlanMerger.tryMergePlans, cannot, because only MergeContext threads downward
> and both leaves carry the full schema. Without that distinction a decline
> also gives up the identical-projection case, which is zero-IO pure reuse.
> Note also that "the projections differ" is not the same as "the scan reads
> more columns", since symmetric filter propagation can add columns
> independently of the projections.
> The alternative fix, making malformed-record detection independent of the
> projection, changes user-visible behavior for a plain df.select("a") on the
> same file in three of four configurations and would have to be a config-gated
> mode; CSVSuite pins the current behavior for SPARK-29101 (count() is 4 with
> CSV column pruning on and 3 with it off). Turning
> spark.sql.csv.parser.columnPruning.enabled off is not a fix either: it
> addresses only the arity check, while the type-error path is decided by a
> conversion loop over the required schema.
> Context. SPARK-57205 declares SCAN_MERGING per format on the DSv2 read path
> and deliberately leaves CSV and JSON out, and withholds it from any file
> table whose reads are not strict, so V2 does not copy any of these
> behaviours. That leaves V1 and V2 disagreeing on these shapes until this is
> fixed.
> The merge itself is not new: non-correlated scalar subqueries have been
> merged since SPARK-34079 (3.4.0), and SPARK-44571 handed that to
> MergeSubplans. Everything measured above is on master.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
