Jorge Molina created SPARK-58518:
------------------------------------
Summary: 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
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]
