[
https://issues.apache.org/jira/browse/SPARK-58518?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Josh Rosen updated SPARK-58518:
-------------------------------
Labels: correctness (was: )
> DataSource.checkAndGlobPathIfNecessary duplicates every input path when
> globbing is disabled, so an unpartitioned relation returns duplicate rows
> -------------------------------------------------------------------------------------------------------------------------------------------------
>
> Key: SPARK-58518
> URL: https://issues.apache.org/jira/browse/SPARK-58518
> Project: Spark
> Issue Type: Bug
> Components: SQL
> Affects Versions: 5.0.0
> Reporter: Jorge Molina
> Priority: Major
> Labels: correctness
>
> h2. Summary
> {{DataSource.checkAndGlobPathIfNecessary}} duplicates its input paths when
> globbing is
> disabled. The duplicated paths become the {{rootPaths}} of the file index,
> and for an
> unpartitioned relation they reach {{FileScanRDD}} intact, so *the same file
> is read several
> times and the query returns duplicate rows*. No error, no warning.
> The highest-impact caller is the Structured Streaming file source, which
> disables globbing
> unconditionally: a microbatch containing one file whose *name* holds a glob
> metacharacter
> emits every other file in that batch more than once.
> h2. Root cause
> {{sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSource.scala}},
> in
> {{checkAndGlobPathIfNecessary}} — the lambda runs *per element* of
> {{globPaths}}, but the
> {{else}} branch returns {{qualifiedPaths}}, the whole input list:
> {code:scala}
> val (globPaths, nonGlobPaths) =
> qualifiedPaths.partition(SparkHadoopUtil.get.isGlobPath)
> val globbedPaths =
> try {
> ThreadUtils.parmap(globPaths, "globPath", numThreads) { globPath =>
> val fs = globPath.getFileSystem(hadoopConf)
> val globResult = if (enableGlobbing) {
> SparkHadoopUtil.get.globPath(fs, globPath)
> } else {
> qualifiedPaths // <-- the WHOLE list, once per glob-looking
> path
> }
> ...
> globResult
> }.flatten
> {code}
> The result is then concatenated with {{nonGlobPaths}} again:
> {{val allPaths = globbedPaths ++ nonGlobPaths}}.
> So with {{G}} paths that *look* like globs and {{n}} that do not,
> {{allPaths}} holds
> {{G*(G+n)+n}} entries instead of {{G+n}}.
> "Looks like a glob" is a bare character scan with no notion of escaping
> ({{SparkHadoopUtil.scala}}):
> {code:scala}
> def isGlobPath(pattern: Path): Boolean = {
> pattern.toString.exists("{}[]*?\\".toSet.contains)
> }
> {code}
> Filenames containing {{[}}, {{]}}, {{\{}}, {{\}}} are legal on HDFS and S3
> and ordinary in real
> ingest zones, so the branch is reached by data, not by misuse.
> h2. Reproduction
> Three single-row CSV files, one of them named {{weird_[x].csv}}:
> {code:python}
> paths = ["plain_a.csv", "plain_b.csv", "weird_[x].csv"] # one row each
> spark.read.csv(paths).count()
> # AnalysisException: [PATH_NOT_FOUND] ... weird_[x].csv
> # -- which is exactly why __globPaths__ exists (SPARK-32810)
> spark.read.options(**{"__globPaths__": "false"}).csv(paths).count()
> # 5 <-- three files, five rows
> # x2 ('0', 'row_from_plain_a.csv') DUPLICATED
> # x2 ('1', 'row_from_plain_b.csv') DUPLICATED
> # x1 ('2', 'row_from_weird_[x].csv')
> {code}
> h3. The predicted count holds exactly
> {{G*(G+n)+n}} is not a description of one observation; it predicts every case
> measured
> (pyspark 3.5.9):
> || G || n || files || predicted || observed ||
> | 1 | 0 | 1 | 1 | 1 |
> | 1 | 2 | 3 | 5 | 5 |
> | 2 | 0 | 2 | 4 | 4 |
> | 2 | 2 | 4 | 10 | 10 |
> | 3 | 1 | 4 | 13 | 13 |
> h3. Format-agnostic, as a defect in {{DataSource}} must be
> Three files, one with a metacharacter in its name,
> {{\_\_globPaths\_\_=false}}:
> || format || rows || expected ||
> | text | 5 | 3 |
> | json | 5 | 3 |
> | csv | 5 | 3 |
> | parquet | 5 | 3 |
> h3. Scoped honestly: it needs an unpartitioned relation
> {{PartitioningAwareFileIndex.allFiles()}} takes a different branch when
> partition columns are
> discovered, and that branch returns a {{Map}}'s values, which silently
> absorbs the duplicates.
> Measured on the same data — three partition directories, one renamed to
> {{part=[9]}}:
> || read || rows ||
> | without {{basePath}} (unpartitioned) | 15 — duplicated |
> | with {{basePath}} (partitions discovered) | 9 — correct |
> So the correctness impact is on unpartitioned relations. It is still a real
> waste of I/O on
> partitioned ones.
> h3. Structured Streaming, the caller that makes this matter
> {{FileStreamSource}} sets {{DataSource.GLOB_PATHS_KEY -> "false"}}
> unconditionally and hands
> the whole microbatch's file list to a {{DataSource}}. Measured with three
> files in the source
> directory, one named {{s_weird_[z].txt}}:
> {code}
> rows emitted: 5 (expected 3)
> x2 stream_line_0 DUPLICATED
> x2 stream_line_1 DUPLICATED
> x1 stream_line_2
> {code}
> Other callers that disable globbing and pass multiple paths — so schema
> inference re-reads
> files as well — include {{MLUtils.parseLibSVMFile}} and the
> {{TextInput*.infer}} paths.
> h2. Why it survived since 2020
> The branch was introduced by SPARK-32810 (PR #29659), whose tests all read a
> *single* path —
> {{G=1, n=0}}, where {{qualifiedPaths == Seq(globPath)}} and the bug is
> invisible. The table
> above confirms it: that row is the one case that comes back correct. There is
> no test with
> {{enableGlobbing = false}} and more than one path.
> Nor is there any dedup downstream to mask it: {{DataSource.scala}} contains
> no {{distinct}},
> and {{PartitioningAwareFileIndex}} does {{rootPaths.flatMap}} without one.
> h2. Suggested fix
> Return the path being processed rather than the whole list:
> {code:scala}
> } else {
> Seq(globPath)
> }
> {code}
> That restores what the parameter's own documentation says
> ({{DataSource.scala}}: "These will be
> globbed before if the '\_\_globPaths\_\_' option is true") and what a
> per-element lambda must
> mean. For the case SPARK-32810 introduced the option to serve — a single path
> — the two
> expressions are the same value, so every scenario it fixed behaves
> identically.
> {{globResult}} stays non-empty, so the {{checkEmptyGlobPath}} check cannot
> begin firing
> {{PATH_NOT_FOUND}} where it did not before. The only observable differences
> are fewer rows —
> the correct count — and less I/O.
> h2. Prior art checked
> * SPARK-32810 / PR #29659 — introduced the branch. Not a report of this
> defect.
> * SPARK-32815 — LibSVM with glob metacharacters; establishes that such
> filenames are meant to
> work, but a different bug.
> * SPARK-28266 "data duplication when {{path}} serde property is present" —
> *related in symptom,
> different in cause*, and worth naming so it is not mistaken for a
> duplicate: there a Hive
> serde property repeated the table LOCATION, and it was fixed in PR #33328
> ({{convertToLogicalRelation should not interpret {{path}} property}}),
> which does not touch
> {{checkAndGlobPathIfNecessary}}. It does establish that Spark treats
> duplicated input paths
> as a genuine correctness bug.
> * SPARK-47833, SPARK-29089, SPARK-56919 all touch this method for unrelated
> reasons.
> * No JIRA matches {{checkAndGlobPathIfNecessary}} + duplication, and no PR in
> {{apache/spark}}
> mentions {{enableGlobbing}} apart from #29659.
> PR to follow, with a regression test in {{DataSourceSuite}} — which currently
> has no case
> exercising {{enableGlobbing = false}} at all.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
