From f14e7edf4c5d0fbc4642f1e917f2147d6a6de662 Mon Sep 17 00:00:00 2001
From: Jorge Molina <70665012+Joorgem@users.noreply.github.com>
Date: Mon, 3 Aug 2026 18:28:46 -0300
Subject: [PATCH] [SPARK-58518][SQL] Do not duplicate input paths when globbing
 is disabled

checkAndGlobPathIfNecessary runs its lambda once per glob-looking path, but the
enableGlobbing == false branch returned qualifiedPaths -- the whole input list. With G
glob-looking paths and n others the result held G*(G+n)+n entries instead of G+n. Those
duplicates become the file index rootPaths and, for an unpartitioned relation, reach
FileScanRDD, so the same file is read repeatedly and the query returns duplicate rows.

The branch now returns the path it was handed. For a single path, which is every case
SPARK-32810 introduced the option for and tested, the two expressions are the same value.

The highest-impact caller is the streaming file source, which disables globbing
unconditionally, so a microbatch containing one file whose name holds a glob metacharacter
emits every other file in that batch more than once.
---
 .../execution/datasources/DataSource.scala    |  2 +-
 .../datasources/DataSourceSuite.scala         | 66 +++++++++++++++++++
 2 files changed, 67 insertions(+), 1 deletion(-)

diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSource.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSource.scala
index 8336e13b59e17..9f0624c2c2e42 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSource.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSource.scala
@@ -825,7 +825,7 @@ object DataSource extends Logging {
           val globResult = if (enableGlobbing) {
             SparkHadoopUtil.get.globPath(fs, globPath)
           } else {
-            qualifiedPaths
+            Seq(globPath)
           }
 
           if (checkEmptyGlobPath && globResult.isEmpty) {
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/DataSourceSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/DataSourceSuite.scala
index 865a89692d6d6..94562cf0a97ef 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/DataSourceSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/DataSourceSuite.scala
@@ -240,6 +240,72 @@ class DataSourceSuite extends SharedSparkSession with PrivateMethodTester {
       parameters = Map("paths" -> "/path1, /path2"))
     )
   }
+
+  test("SPARK-58518: checkAndGlobPathIfNecessary must not duplicate paths " +
+    "when globbing is disabled") {
+    // The lambda runs once per glob-looking path, so returning `qualifiedPaths` -- the whole
+    // input -- multiplies the result: G glob-looking paths and n others give G*(G+n)+n entries
+    // instead of G+n. Those duplicates become the file index's rootPaths.
+    //
+    // checkFilesExist = false keeps this focused on the path arithmetic and avoids depending on
+    // MockFileSystem.exists, which only knows allPathsInFs.
+    val onlyGlobLooking = DataSource.checkAndGlobPathIfNecessary(
+      Seq(path1.toString, path2.toString, globPath1.toString),
+      hadoopConf,
+      checkEmptyGlobPath = true,
+      checkFilesExist = false,
+      enableGlobbing = false
+    )
+
+    // Asserted as a Seq, NOT a Set: `toSet` is what would hide the defect. Order follows
+    // `globbedPaths ++ nonGlobPaths`.
+    assert(onlyGlobLooking === Seq(globPath1, path1, path2))
+
+    // Two glob-looking paths among four is where the multiplication is unmistakable: the buggy
+    // branch yields 2*(2+2)+2 = 10 entries for 4 inputs.
+    val twoGlobLooking = DataSource.checkAndGlobPathIfNecessary(
+      Seq(path1.toString, path2.toString, globPath1.toString, globPath2.toString),
+      hadoopConf,
+      checkEmptyGlobPath = true,
+      checkFilesExist = false,
+      enableGlobbing = false
+    )
+
+    assert(twoGlobLooking === Seq(globPath1, globPath2, path1, path2))
+    assert(twoGlobLooking.length === 4, "globbing is disabled, so each path must appear once")
+  }
+
+  test("SPARK-58518: a filename with a glob metacharacter must not duplicate rows " +
+    "when globbing is disabled") {
+    // isGlobPath is a bare character scan over "{}[]*?\\" with no notion of escaping, so an
+    // ordinary filename takes the branch. Names containing brackets are legal on HDFS and S3,
+    // which is why SPARK-32810 introduced __globPaths__ in the first place.
+    withTempDir { dir =>
+      val plainA = new File(dir, "plain_a.txt")
+      val plainB = new File(dir, "plain_b.txt")
+      val bracketed = new File(dir, "weird_[x].txt")
+      // Files.writeString rather than a PrintWriter: scalastyle bans println, and it is
+      // fully qualified so this test adds no import to the suite.
+      java.nio.file.Files.writeString(plainA.toPath, "row_a\n")
+      java.nio.file.Files.writeString(plainB.toPath, "row_b\n")
+      java.nio.file.Files.writeString(bracketed.toPath, "row_c\n")
+
+      val paths = Seq(plainA, plainB, bracketed).map(_.getCanonicalPath)
+
+      val df = spark.read
+        .option(DataSource.GLOB_PATHS_KEY, "false")
+        .text(paths: _*)
+
+      // Three files, one row each. Before the fix this returned five rows, with the two plain
+      // files duplicated, because the bracketed name made the branch emit all three paths.
+      //
+      // Collected and sorted into an Array rather than compared as a Set, for the same reason
+      // the unit test above avoids toSet: a Set is precisely what would hide duplication. This
+      // also keeps the test free of a `Row` import the suite does not currently carry.
+      assert(df.collect().map(_.getString(0)).sorted === Array("row_a", "row_b", "row_c"))
+      assert(df.count() === 3)
+    }
+  }
 }
 
 object TestPaths {