diff --git a/docs/tutorial/files/geotiffmetadata-sedona-spark.md b/docs/tutorial/files/geotiffmetadata-sedona-spark.md index 28f2e923afe..e071174eb87 100644 --- a/docs/tutorial/files/geotiffmetadata-sedona-spark.md +++ b/docs/tutorial/files/geotiffmetadata-sedona-spark.md @@ -77,7 +77,12 @@ Or load a single file: df = sedona.read.format("geotiff.metadata").load("/path/to/image.tiff") ``` -For directory loads, `recursiveFileLookup=true` and a GeoTIFF extension +Glob paths keep native Spark semantics: a file glob such as `/dir/a*.tif` matches +direct children of `/dir` only, and a glob that matches nothing is an error, as in +every other Spark file source. + +For directory loads — including globs that expand to directories, such as +`/data/region*` — `recursiveFileLookup=true` and a GeoTIFF extension `pathGlobFilter` are applied as defaults; an explicit option always wins. In particular, set `recursiveFileLookup=false` to keep Hive-style partition discovery (e.g. `year=2020/` subdirectories become a `year` column): diff --git a/docs/tutorial/files/netcdfmetadata-sedona-spark.md b/docs/tutorial/files/netcdfmetadata-sedona-spark.md index 5f0e032363c..f02c31d6047 100644 --- a/docs/tutorial/files/netcdfmetadata-sedona-spark.md +++ b/docs/tutorial/files/netcdfmetadata-sedona-spark.md @@ -80,7 +80,12 @@ Or use a glob pattern: df = sedona.read.format("netcdf.metadata").load("/path/to/*.nc") ``` -For directory loads, `recursiveFileLookup=true` and a NetCDF extension +Glob paths keep native Spark semantics: a file glob such as `/dir/a*.nc` matches +direct children of `/dir` only, and a glob that matches nothing is an error, as in +every other Spark file source. + +For directory loads — including globs that expand to directories, such as +`/data/region*` — `recursiveFileLookup=true` and a NetCDF extension `pathGlobFilter` are applied as defaults; an explicit option always wins. In particular, set `recursiveFileLookup=false` to keep Hive-style partition discovery (e.g. `year=2020/` subdirectories become a `year` column): diff --git a/spark/common/src/main/scala/org/apache/spark/sql/sedona_sql/io/geotiffmetadata/GeoTiffMetadataDataSource.scala b/spark/common/src/main/scala/org/apache/spark/sql/sedona_sql/io/geotiffmetadata/GeoTiffMetadataDataSource.scala index c22a9910308..0a46a3de3d0 100644 --- a/spark/common/src/main/scala/org/apache/spark/sql/sedona_sql/io/geotiffmetadata/GeoTiffMetadataDataSource.scala +++ b/spark/common/src/main/scala/org/apache/spark/sql/sedona_sql/io/geotiffmetadata/GeoTiffMetadataDataSource.scala @@ -24,12 +24,14 @@ import org.apache.hadoop.mapreduce.Job import org.apache.spark.sql.SparkSession import org.apache.spark.sql.connector.catalog.Table import org.apache.spark.sql.connector.catalog.TableProvider +import org.apache.spark.sql.execution.datasources.DataSource import org.apache.spark.sql.execution.datasources.FileFormat import org.apache.spark.sql.execution.datasources.OutputWriterFactory import org.apache.spark.sql.execution.datasources.v2.FileDataSourceV2 import org.apache.spark.sql.sources.DataSourceRegister import org.apache.spark.sql.types.StructType import org.apache.spark.sql.util.CaseInsensitiveStringMap +import org.apache.spark.util.HadoopFSUtils import scala.collection.JavaConverters._ import scala.util.Try @@ -45,12 +47,10 @@ class GeoTiffMetadataDataSource override def shortName(): String = "geotiff.metadata" - private val loadTifPattern = "(.*)/([^/]*\\*[^/]*\\.(?i:tif|tiff))$".r - private def createTable( options: CaseInsensitiveStringMap, userSchema: Option[StructType] = None): Table = { - var paths = getPaths(options) + val paths = getPaths(options) var optionsWithoutPaths = getOptionsWithoutPaths(options) val tableName = getTableName(options, paths) @@ -62,24 +62,16 @@ class GeoTiffMetadataDataSource val hasUserGlobFilter = optionsWithoutPaths.containsKey("pathGlobFilter") val hasUserRecursive = optionsWithoutPaths.containsKey("recursiveFileLookup") - if (paths.size == 1 && !isDirectory(paths.head, optionsWithoutPaths)) { - // Rewrite glob patterns like /path/to/some*glob*.tif into /path/to with - // pathGlobFilter="some*glob*.tif" to avoid listing .tif files as directories. When the - // user supplied their own pathGlobFilter, keep the glob path untouched so Spark applies - // both constraints natively (the rewrite could only keep one of them). - paths.head match { - case loadTifPattern(prefix, glob) if !hasUserGlobFilter => - paths = Seq(prefix) - newOptions.put("pathGlobFilter", glob) - case _ => - } - } else if (paths.nonEmpty && paths.forall(p => isDirectory(p, optionsWithoutPaths))) { - // Directory roots (single or multiple, with or without trailing slash): default to a - // recursive scan filtered to GeoTIFF extensions. These are defaults only — an explicit - // recursiveFileLookup=false keeps Hive-style partition discovery available. They apply - // only when EVERY root is a directory: Spark applies pathGlobFilter to all roots, so a - // mixed load(dir, explicitFile) must not silently drop an explicitly named file whose - // name does not match the extension filter. + // File paths and file globs are passed to Spark untouched, keeping native listing + // semantics: a glob such as /dir/a*.tif matches direct children of /dir only, with or + // without recursiveFileLookup. Directory roots (single or multiple; plain, with a + // trailing slash, or produced by a glob such as /data/region*) default to a recursive + // scan filtered to GeoTIFF extensions. These are defaults only — an explicit + // recursiveFileLookup=false keeps Hive-style partition discovery available. They apply + // only when EVERY root stands for directories: Spark applies pathGlobFilter to all + // roots, so a mixed load(dir, explicitFile) must not silently drop an explicitly named + // file whose name does not match the extension filter. + if (paths.nonEmpty && paths.forall(p => isDirectoryRoot(p, optionsWithoutPaths))) { if (!hasUserRecursive) { newOptions.put("recursiveFileLookup", "true") } @@ -116,22 +108,53 @@ class GeoTiffMetadataDataSource classOf[GeoTiffMetadataUnsupportedFileFormat] /** - * Check if a path points to a directory. A trailing `/` is treated as a directory without any - * FS call; otherwise, Hadoop `FileSystem.getFileStatus(path).isDirectory` is consulted. Returns - * `false` if the FS call fails (e.g., path does not exist or is a glob pattern). Uses the same - * per-read options (e.g., `fs.s3a.*`) as the scan so directory detection works on the - * configured filesystem. + * Whether a load path stands for directories. A trailing `/` is treated as a directory + * assertion without any FS call. A glob pattern (unless `__globPaths__` disables expansion, + * exactly as `FileTable` honors it) is expanded with Hadoop `FileSystem.globStatus` and must + * match at least one status, with every match visible to Spark's scan being a directory. + * Hidden-named file matches (`_SUCCESS`, dotfiles, `*._COPYING_`) are invisible — Spark's + * listing discards them — so they neither sway the classification nor, when a glob matches + * nothing else, veto the defaults earned by the remaining roots; hidden-named DIRECTORY matches + * count normally, because Spark traverses a directory root regardless of its name. Anything + * else is probed literally with `getFileStatus`; the hidden-name rule never applies here, + * because Spark exempts explicitly given roots from it. Returns `false` when the probe fails or + * a glob matches nothing (Spark reports the missing path itself during listing). Uses the same + * per-read options (e.g., `fs.s3a.*`) as the scan so detection works on the configured + * filesystem. */ - private def isDirectory(pathStr: String, options: CaseInsensitiveStringMap): Boolean = { + private def isDirectoryRoot(pathStr: String, options: CaseInsensitiveStringMap): Boolean = { if (pathStr.endsWith("/")) return true Try { val hadoopConf = sparkSession.sessionState.newHadoopConfWithOptions(options.asScala.toMap) val path = new Path(pathStr) val fs = path.getFileSystem(hadoopConf) - fs.getFileStatus(path).isDirectory + if (globbingEnabled(options) && isGlobPath(path)) { + val matches = Option(fs.globStatus(path)).getOrElse(Array.empty[FileStatus]) + // An unmatched glob is not a directory root (Spark itself reports the missing + // path). Otherwise classify by what Spark's listing will actually read: it drops + // hidden-named FILE matches (their own name comes back from listStatus and is + // name-checked) but traverses a directory root regardless of its name — only the + // children are name-checked. Hidden files are therefore invisible to the scan, + // and a glob whose visible contribution is empty (e.g. matching only a _SUCCESS + // marker) must not veto the defaults earned by the other roots. + matches.nonEmpty && matches + .filterNot(m => + !m.isDirectory && HadoopFSUtils.shouldFilterOutPathName(m.getPath.getName)) + .forall(_.isDirectory) + } else { + fs.getFileStatus(path).isDirectory + } }.getOrElse(false) } + + /** Mirrors `FileTable.globPaths`: expansion is on unless `__globPaths__` is set to non-true. */ + private def globbingEnabled(options: CaseInsensitiveStringMap): Boolean = + Option(options.get(DataSource.GLOB_PATHS_KEY)).forall(_ == "true") + + /** The wildcard set of `SparkHadoopUtil.isGlobPath`, which governs Spark's own expansion. */ + private def isGlobPath(path: Path): Boolean = + path.toString.exists("{}[]*?\\".contains(_)) } /** diff --git a/spark/common/src/main/scala/org/apache/spark/sql/sedona_sql/io/netcdfmetadata/NetCdfMetadataDataSource.scala b/spark/common/src/main/scala/org/apache/spark/sql/sedona_sql/io/netcdfmetadata/NetCdfMetadataDataSource.scala index 44236a576ed..fc72f6598fc 100644 --- a/spark/common/src/main/scala/org/apache/spark/sql/sedona_sql/io/netcdfmetadata/NetCdfMetadataDataSource.scala +++ b/spark/common/src/main/scala/org/apache/spark/sql/sedona_sql/io/netcdfmetadata/NetCdfMetadataDataSource.scala @@ -24,12 +24,14 @@ import org.apache.hadoop.mapreduce.Job import org.apache.spark.sql.SparkSession import org.apache.spark.sql.connector.catalog.Table import org.apache.spark.sql.connector.catalog.TableProvider +import org.apache.spark.sql.execution.datasources.DataSource import org.apache.spark.sql.execution.datasources.FileFormat import org.apache.spark.sql.execution.datasources.OutputWriterFactory import org.apache.spark.sql.execution.datasources.v2.FileDataSourceV2 import org.apache.spark.sql.sources.DataSourceRegister import org.apache.spark.sql.types.StructType import org.apache.spark.sql.util.CaseInsensitiveStringMap +import org.apache.spark.util.HadoopFSUtils import scala.collection.JavaConverters._ import scala.util.Try @@ -46,12 +48,10 @@ class NetCdfMetadataDataSource override def shortName(): String = "netcdf.metadata" - private val loadNcPattern = "(.*)/([^/]*\\*[^/]*\\.(?i:nc|nc4|netcdf))$".r - private def createTable( options: CaseInsensitiveStringMap, userSchema: Option[StructType] = None): Table = { - var paths = getPaths(options) + val paths = getPaths(options) var optionsWithoutPaths = getOptionsWithoutPaths(options) val tableName = getTableName(options, paths) @@ -63,24 +63,16 @@ class NetCdfMetadataDataSource val hasUserGlobFilter = optionsWithoutPaths.containsKey("pathGlobFilter") val hasUserRecursive = optionsWithoutPaths.containsKey("recursiveFileLookup") - if (paths.size == 1 && !isDirectory(paths.head, optionsWithoutPaths)) { - // Rewrite glob patterns like /path/to/some*glob*.nc into /path/to with - // pathGlobFilter="some*glob*.nc" to avoid listing .nc files as directories. When the - // user supplied their own pathGlobFilter, keep the glob path untouched so Spark applies - // both constraints natively (the rewrite could only keep one of them). - paths.head match { - case loadNcPattern(prefix, glob) if !hasUserGlobFilter => - paths = Seq(prefix) - newOptions.put("pathGlobFilter", glob) - case _ => - } - } else if (paths.nonEmpty && paths.forall(p => isDirectory(p, optionsWithoutPaths))) { - // Directory roots (single or multiple, with or without trailing slash): default to a - // recursive scan filtered to NetCDF extensions. These are defaults only — an explicit - // recursiveFileLookup=false keeps Hive-style partition discovery available. They apply - // only when EVERY root is a directory: Spark applies pathGlobFilter to all roots, so a - // mixed load(dir, explicitFile) must not silently drop an explicitly named file whose - // name does not match the extension filter. + // File paths and file globs are passed to Spark untouched, keeping native listing + // semantics: a glob such as /dir/a*.nc matches direct children of /dir only, with or + // without recursiveFileLookup. Directory roots (single or multiple; plain, with a + // trailing slash, or produced by a glob such as /data/region*) default to a recursive + // scan filtered to NetCDF extensions. These are defaults only — an explicit + // recursiveFileLookup=false keeps Hive-style partition discovery available. They apply + // only when EVERY root stands for directories: Spark applies pathGlobFilter to all + // roots, so a mixed load(dir, explicitFile) must not silently drop an explicitly named + // file whose name does not match the extension filter. + if (paths.nonEmpty && paths.forall(p => isDirectoryRoot(p, optionsWithoutPaths))) { if (!hasUserRecursive) { newOptions.put("recursiveFileLookup", "true") } @@ -117,22 +109,53 @@ class NetCdfMetadataDataSource classOf[NetCdfMetadataUnsupportedFileFormat] /** - * Check if a path points to a directory. A trailing `/` is treated as a directory without any - * FS call; otherwise, Hadoop `FileSystem.getFileStatus(path).isDirectory` is consulted. Returns - * `false` if the FS call fails (e.g., path does not exist or is a glob pattern). Uses the same - * per-read options (e.g., `fs.s3a.*`) as the scan so directory detection works on the - * configured filesystem. + * Whether a load path stands for directories. A trailing `/` is treated as a directory + * assertion without any FS call. A glob pattern (unless `__globPaths__` disables expansion, + * exactly as `FileTable` honors it) is expanded with Hadoop `FileSystem.globStatus` and must + * match at least one status, with every match visible to Spark's scan being a directory. + * Hidden-named file matches (`_SUCCESS`, dotfiles, `*._COPYING_`) are invisible — Spark's + * listing discards them — so they neither sway the classification nor, when a glob matches + * nothing else, veto the defaults earned by the remaining roots; hidden-named DIRECTORY matches + * count normally, because Spark traverses a directory root regardless of its name. Anything + * else is probed literally with `getFileStatus`; the hidden-name rule never applies here, + * because Spark exempts explicitly given roots from it. Returns `false` when the probe fails or + * a glob matches nothing (Spark reports the missing path itself during listing). Uses the same + * per-read options (e.g., `fs.s3a.*`) as the scan so detection works on the configured + * filesystem. */ - private def isDirectory(pathStr: String, options: CaseInsensitiveStringMap): Boolean = { + private def isDirectoryRoot(pathStr: String, options: CaseInsensitiveStringMap): Boolean = { if (pathStr.endsWith("/")) return true Try { val hadoopConf = sparkSession.sessionState.newHadoopConfWithOptions(options.asScala.toMap) val path = new Path(pathStr) val fs = path.getFileSystem(hadoopConf) - fs.getFileStatus(path).isDirectory + if (globbingEnabled(options) && isGlobPath(path)) { + val matches = Option(fs.globStatus(path)).getOrElse(Array.empty[FileStatus]) + // An unmatched glob is not a directory root (Spark itself reports the missing + // path). Otherwise classify by what Spark's listing will actually read: it drops + // hidden-named FILE matches (their own name comes back from listStatus and is + // name-checked) but traverses a directory root regardless of its name — only the + // children are name-checked. Hidden files are therefore invisible to the scan, + // and a glob whose visible contribution is empty (e.g. matching only a _SUCCESS + // marker) must not veto the defaults earned by the other roots. + matches.nonEmpty && matches + .filterNot(m => + !m.isDirectory && HadoopFSUtils.shouldFilterOutPathName(m.getPath.getName)) + .forall(_.isDirectory) + } else { + fs.getFileStatus(path).isDirectory + } }.getOrElse(false) } + + /** Mirrors `FileTable.globPaths`: expansion is on unless `__globPaths__` is set to non-true. */ + private def globbingEnabled(options: CaseInsensitiveStringMap): Boolean = + Option(options.get(DataSource.GLOB_PATHS_KEY)).forall(_ == "true") + + /** The wildcard set of `SparkHadoopUtil.isGlobPath`, which governs Spark's own expansion. */ + private def isGlobPath(path: Path): Boolean = + path.toString.exists("{}[]*?\\".contains(_)) } /** diff --git a/spark/common/src/test/scala/org/apache/sedona/sql/geotiffMetadataTest.scala b/spark/common/src/test/scala/org/apache/sedona/sql/geotiffMetadataTest.scala index 28f329c6283..9d442827e61 100644 --- a/spark/common/src/test/scala/org/apache/sedona/sql/geotiffMetadataTest.scala +++ b/spark/common/src/test/scala/org/apache/sedona/sql/geotiffMetadataTest.scala @@ -328,7 +328,7 @@ class geotiffMetadataTest extends TestBaseScala with BeforeAndAfterAll { it("should apply both a glob path and an explicit pathGlobFilter") { // Native Spark semantics: both constraints apply, so a*.tiff restricted to b*.tiff is - // empty. The internal glob-to-filter rewrite must not discard either constraint. + // empty. val dir = new File(tempDir, "globAndFilter") dir.mkdirs() FileUtils.copyFile(new File(singleFileLocation), new File(dir, "a.tiff")) @@ -346,6 +346,139 @@ class geotiffMetadataTest extends TestBaseScala with BeforeAndAfterAll { assertEquals(1L, globOnly.count()) } + it("should not widen a file glob into nested files under explicit recursion") { + // GH-3131: the glob used to be rewritten to path=dir + pathGlobFilter, and + // pathGlobFilter matches leaf names at any depth, so recursiveFileLookup=true pulled + // nested a2.tiff into a load of dir/a*.tiff. The glob path now reaches Spark natively + // and matches direct children of the directory only. + val dir = new File(tempDir, "globRecursive") + val sub = new File(dir, "sub") + sub.mkdirs() + FileUtils.copyFile(new File(singleFileLocation), new File(dir, "a1.tiff")) + FileUtils.copyFile(new File(singleFileLocation), new File(sub, "a2.tiff")) + + val df = sparkSession.read + .format("geotiff.metadata") + .option("recursiveFileLookup", "true") + .load(dir.getAbsolutePath + "/a*.tiff") + assertEquals(1L, df.count()) + assert(df.select("path").first().getString(0).endsWith("a1.tiff")) + } + + it("should apply the directory-scan defaults to a glob that matches directories") { + // GH-3131: region* expands to directories, which must get the same defaults as loading + // them by explicit path — recursive lookup plus the extension filter. + val root = new File(tempDir, "dirGlob") + val region1 = new File(root, "region1") + val region2Sub = new File(root, "region2/sub") + region1.mkdirs() + region2Sub.mkdirs() + FileUtils.copyFile(new File(singleFileLocation), new File(region1, "x.tiff")) + FileUtils.copyFile(new File(singleFileLocation), new File(region2Sub, "y.tiff")) + // The extension filter must exclude this; without it the reader would surface a row + FileUtils.writeStringToFile(new File(region1, "readme.txt"), "not a geotiff", "UTF-8") + + val df = sparkSession.read + .format("geotiff.metadata") + .load(root.getAbsolutePath + "/region*") + assertEquals(2L, df.count()) + } + + it("should keep the directory defaults when a glob also matches a _SUCCESS marker") { + // Entries Spark's own listing discards (_SUCCESS, dotfiles, *._COPYING_) must not + // sway the all-directories classification: root/* still stands for directories, so + // the recursive-scan and extension-filter defaults apply and the nested file is found. + val root = new File(tempDir, "markerGlob") + val region1Sub = new File(root, "region1/sub") + region1Sub.mkdirs() + FileUtils.copyFile(new File(singleFileLocation), new File(root, "region1/x.tiff")) + FileUtils.copyFile(new File(singleFileLocation), new File(region1Sub, "y.tiff")) + FileUtils.writeStringToFile(new File(root, "_SUCCESS"), "", "UTF-8") + + val df = sparkSession.read.format("geotiff.metadata").load(root.getAbsolutePath + "/*") + assertEquals(2L, df.count()) + } + + it("should keep the defaults when a glob matches a hidden-named directory") { + // Spark's listing drops hidden-named files but traverses a directory root regardless + // of its name (only children are name-checked), so /root/_* matching _staging/ must + // still receive the directory-scan defaults. + val root = new File(tempDir, "hiddenDirGlob") + val staging = new File(root, "_staging") + val sub = new File(staging, "sub") + sub.mkdirs() + FileUtils.copyFile(new File(singleFileLocation), new File(staging, "x.tiff")) + FileUtils.copyFile(new File(singleFileLocation), new File(sub, "y.tiff")) + + val df = sparkSession.read.format("geotiff.metadata").load(root.getAbsolutePath + "/_*") + assertEquals(2L, df.count()) + } + + it("should not let a glob matching only ignored files veto another root's defaults") { + // The first glob matches only a _SUCCESS marker, which Spark's listing discards — + // its visible contribution is empty, so it must not flip the all-directories + // classification that gives the second root its recursive scan. + val root = new File(tempDir, "ignoredOnlyGlob") + val markerDir = new File(root, "markers") + val dataSub = new File(root, "data/sub") + markerDir.mkdirs() + dataSub.mkdirs() + FileUtils.writeStringToFile(new File(markerDir, "_SUCCESS"), "", "UTF-8") + FileUtils.copyFile(new File(singleFileLocation), new File(root, "data/x.tiff")) + FileUtils.copyFile(new File(singleFileLocation), new File(dataSub, "y.tiff")) + + val df = sparkSession.read + .format("geotiff.metadata") + .load(markerDir.getAbsolutePath + "/*", new File(root, "data").getAbsolutePath) + assertEquals(2L, df.count()) + } + + it("should apply the directory defaults to an explicitly loaded underscore directory") { + // Spark's hidden-name rule applies to listing entries, never to explicitly given + // roots — so probing a literal _staging root must not filter it either. + val root = new File(tempDir, "underscoreDir") + val staging = new File(root, "_staging") + val sub = new File(staging, "sub") + sub.mkdirs() + FileUtils.copyFile(new File(singleFileLocation), new File(staging, "x.tiff")) + FileUtils.copyFile(new File(singleFileLocation), new File(sub, "y.tiff")) + + val df = sparkSession.read.format("geotiff.metadata").load(staging.getAbsolutePath) + assertEquals(2L, df.count()) + } + + it("should probe the path literally when glob expansion is disabled") { + // With __globPaths__=false Spark treats the path literally, so a directory whose + // name contains glob metacharacters must be probed with getFileStatus, not expanded + // as a pattern — otherwise it loses the directory-scan defaults. + val root = new File(tempDir, "literalGlobChars") + val dir = new File(root, "[region]") + val sub = new File(dir, "sub") + sub.mkdirs() + FileUtils.copyFile(new File(singleFileLocation), new File(dir, "x.tiff")) + FileUtils.copyFile(new File(singleFileLocation), new File(sub, "y.tiff")) + + val df = sparkSession.read + .format("geotiff.metadata") + .option("__globPaths__", "false") + .load(dir.getAbsolutePath) + assertEquals(2L, df.count()) + } + + it("should report a file glob that matches nothing as a missing path") { + // Native Spark glob semantics, aligned with every other file source: an empty glob is + // an error, not a silently empty result (the pre-GH-3131 rewrite returned no rows). + val dir = new File(tempDir, "emptyGlob") + dir.mkdirs() + FileUtils.copyFile(new File(singleFileLocation), new File(dir, "a.tiff")) + intercept[org.apache.spark.sql.AnalysisException] { + sparkSession.read + .format("geotiff.metadata") + .load(dir.getAbsolutePath + "/nomatch*.tiff") + .count() + } + } + it("should filter non-GeoTIFF files when loading multiple directories") { val dirA = new File(tempDir, "multiA") val dirB = new File(tempDir, "multiB") diff --git a/spark/common/src/test/scala/org/apache/sedona/sql/netcdfMetadataTest.scala b/spark/common/src/test/scala/org/apache/sedona/sql/netcdfMetadataTest.scala index 664f0c2da1d..9ae1efcb99f 100644 --- a/spark/common/src/test/scala/org/apache/sedona/sql/netcdfMetadataTest.scala +++ b/spark/common/src/test/scala/org/apache/sedona/sql/netcdfMetadataTest.scala @@ -1118,7 +1118,6 @@ class netcdfMetadataTest extends TestBaseScala with BeforeAndAfterAll { it("should apply both a glob path and an explicit pathGlobFilter") { // Native Spark semantics: both constraints apply, so a*.nc restricted to b*.nc is empty. - // The internal glob-to-filter rewrite must not discard either constraint. val dir = new File(tempDir, "globAndFilter") dir.mkdirs() FileUtils.copyFile(new File(singleFileLocation), new File(dir, "a.nc")) @@ -1136,6 +1135,139 @@ class netcdfMetadataTest extends TestBaseScala with BeforeAndAfterAll { assertEquals(1L, globOnly.count()) } + it("should not widen a file glob into nested files under explicit recursion") { + // GH-3131: the glob used to be rewritten to path=dir + pathGlobFilter, and + // pathGlobFilter matches leaf names at any depth, so recursiveFileLookup=true pulled + // nested a2.nc into a load of dir/a*.nc. The glob path now reaches Spark natively and + // matches direct children of the directory only. + val dir = new File(tempDir, "globRecursive") + val sub = new File(dir, "sub") + sub.mkdirs() + FileUtils.copyFile(new File(singleFileLocation), new File(dir, "a1.nc")) + FileUtils.copyFile(new File(singleFileLocation), new File(sub, "a2.nc")) + + val df = sparkSession.read + .format("netcdf.metadata") + .option("recursiveFileLookup", "true") + .load(dir.getAbsolutePath + "/a*.nc") + assertEquals(1L, df.count()) + assert(df.select("path").first().getString(0).endsWith("a1.nc")) + } + + it("should apply the directory-scan defaults to a glob that matches directories") { + // GH-3131: region* expands to directories, which must get the same defaults as loading + // them by explicit path — recursive lookup plus the extension filter. + val root = new File(tempDir, "dirGlob") + val region1 = new File(root, "region1") + val region2Sub = new File(root, "region2/sub") + region1.mkdirs() + region2Sub.mkdirs() + FileUtils.copyFile(new File(singleFileLocation), new File(region1, "x.nc")) + FileUtils.copyFile(new File(singleFileLocation), new File(region2Sub, "y.nc")) + // The extension filter must exclude this; without it the reader would surface a row + FileUtils.writeStringToFile(new File(region1, "readme.txt"), "not netcdf", "UTF-8") + + val df = sparkSession.read + .format("netcdf.metadata") + .load(root.getAbsolutePath + "/region*") + assertEquals(2L, df.count()) + } + + it("should keep the directory defaults when a glob also matches a _SUCCESS marker") { + // Entries Spark's own listing discards (_SUCCESS, dotfiles, *._COPYING_) must not + // sway the all-directories classification: root/* still stands for directories, so + // the recursive-scan and extension-filter defaults apply and the nested file is found. + val root = new File(tempDir, "markerGlob") + val region1Sub = new File(root, "region1/sub") + region1Sub.mkdirs() + FileUtils.copyFile(new File(singleFileLocation), new File(root, "region1/x.nc")) + FileUtils.copyFile(new File(singleFileLocation), new File(region1Sub, "y.nc")) + FileUtils.writeStringToFile(new File(root, "_SUCCESS"), "", "UTF-8") + + val df = sparkSession.read.format("netcdf.metadata").load(root.getAbsolutePath + "/*") + assertEquals(2L, df.count()) + } + + it("should keep the defaults when a glob matches a hidden-named directory") { + // Spark's listing drops hidden-named files but traverses a directory root regardless + // of its name (only children are name-checked), so /root/_* matching _staging/ must + // still receive the directory-scan defaults. + val root = new File(tempDir, "hiddenDirGlob") + val staging = new File(root, "_staging") + val sub = new File(staging, "sub") + sub.mkdirs() + FileUtils.copyFile(new File(singleFileLocation), new File(staging, "x.nc")) + FileUtils.copyFile(new File(singleFileLocation), new File(sub, "y.nc")) + + val df = sparkSession.read.format("netcdf.metadata").load(root.getAbsolutePath + "/_*") + assertEquals(2L, df.count()) + } + + it("should not let a glob matching only ignored files veto another root's defaults") { + // The first glob matches only a _SUCCESS marker, which Spark's listing discards — + // its visible contribution is empty, so it must not flip the all-directories + // classification that gives the second root its recursive scan. + val root = new File(tempDir, "ignoredOnlyGlob") + val markerDir = new File(root, "markers") + val dataSub = new File(root, "data/sub") + markerDir.mkdirs() + dataSub.mkdirs() + FileUtils.writeStringToFile(new File(markerDir, "_SUCCESS"), "", "UTF-8") + FileUtils.copyFile(new File(singleFileLocation), new File(root, "data/x.nc")) + FileUtils.copyFile(new File(singleFileLocation), new File(dataSub, "y.nc")) + + val df = sparkSession.read + .format("netcdf.metadata") + .load(markerDir.getAbsolutePath + "/*", new File(root, "data").getAbsolutePath) + assertEquals(2L, df.count()) + } + + it("should apply the directory defaults to an explicitly loaded underscore directory") { + // Spark's hidden-name rule applies to listing entries, never to explicitly given + // roots — so probing a literal _staging root must not filter it either. + val root = new File(tempDir, "underscoreDir") + val staging = new File(root, "_staging") + val sub = new File(staging, "sub") + sub.mkdirs() + FileUtils.copyFile(new File(singleFileLocation), new File(staging, "x.nc")) + FileUtils.copyFile(new File(singleFileLocation), new File(sub, "y.nc")) + + val df = sparkSession.read.format("netcdf.metadata").load(staging.getAbsolutePath) + assertEquals(2L, df.count()) + } + + it("should probe the path literally when glob expansion is disabled") { + // With __globPaths__=false Spark treats the path literally, so a directory whose + // name contains glob metacharacters must be probed with getFileStatus, not expanded + // as a pattern — otherwise it loses the directory-scan defaults. + val root = new File(tempDir, "literalGlobChars") + val dir = new File(root, "[region]") + val sub = new File(dir, "sub") + sub.mkdirs() + FileUtils.copyFile(new File(singleFileLocation), new File(dir, "x.nc")) + FileUtils.copyFile(new File(singleFileLocation), new File(sub, "y.nc")) + + val df = sparkSession.read + .format("netcdf.metadata") + .option("__globPaths__", "false") + .load(dir.getAbsolutePath) + assertEquals(2L, df.count()) + } + + it("should report a file glob that matches nothing as a missing path") { + // Native Spark glob semantics, aligned with every other file source: an empty glob is + // an error, not a silently empty result (the pre-GH-3131 rewrite returned no rows). + val dir = new File(tempDir, "emptyGlob") + dir.mkdirs() + FileUtils.copyFile(new File(singleFileLocation), new File(dir, "a.nc")) + intercept[org.apache.spark.sql.AnalysisException] { + sparkSession.read + .format("netcdf.metadata") + .load(dir.getAbsolutePath + "/nomatch*.nc") + .count() + } + } + it("should filter non-NetCDF files when loading multiple directories") { val dirA = new File(tempDir, "multiA") val dirB = new File(tempDir, "multiB")