(spark) branch master updated: [SPARK-56756][SQL] Add error class for recursiveFileLookup conflict with partitioned data source

[we...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

cloud-fan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git


The following commit(s) were added to refs/heads/master by this push:
     new 051ea2b17989 [SPARK-56756][SQL] Add error class for recursiveFileLookup conflict with partitioned data source
051ea2b17989 is described below

commit 051ea2b17989a2f22c287d4de22af0537e033cff
Author: Mark Jarvin <ma...@databricks.com>
AuthorDate: Fri May 15 01:35:59 2026 +0800

    [SPARK-56756][SQL] Add error class for recursiveFileLookup conflict with partitioned data source
    
    ### What changes were proposed in this pull request?
    
    `PartitioningAwareFileIndex.listFiles` rejects the combination of `recursiveFileLookup=true` and a non-empty `partitionSpec().partitionColumns` by throwing a raw `java.lang.IllegalArgumentException` with the message "Datasource with partition do not allow recursive file loading."
    
    This PR replaces that with a tagged `AnalysisException` using a new error class:
    
    - New error class `RECURSIVE_FILE_LOOKUP_NOT_SUPPORTED_FOR_PARTITIONED_DATA_SOURCE` (`sqlState 0A000`) in `error-conditions.json`.
    - New helper `QueryCompilationErrors.recursiveFileLookupNotSupportedForPartitionedDataSourceError()`.
    - Throw site in `PartitioningAwareFileIndex.scala` updated to use the helper.
    
    ### Why are the changes needed?
    
    The raw `IllegalArgumentException` is unclassified and does not surface as a user-facing error with a clear message. Replacing it with an `AnalysisException` using a proper error class ensures it is correctly classified as a user error with an actionable message.
    
    ### Does this PR introduce _any_ user-facing change?
    
    Yes. Users who hit this error will now see a clearer message:
    
    > Recursive file loading is not supported when the data source has explicit partition columns. Either remove the option "recursiveFileLookup", or read the data without supplying partition columns (for example, do not read a partitioned table or set partition-column options such as "cloudFiles.partitionColumns").
    
    Previously the error was a raw `IllegalArgumentException` with the message "Datasource with partition do not allow recursive file loading."
    
    ### How was this patch tested?
    
    Added `"recursiveFileLookup with a partitioned catalog table is rejected"` in `FileBasedDataSourceSuite`, which creates a partitioned Parquet catalog table, then asserts that reading it with `recursiveFileLookup=true` throws an `AnalysisException` with condition `RECURSIVE_FILE_LOOKUP_NOT_SUPPORTED_FOR_PARTITIONED_DATA_SOURCE`.
    
    ### Was this patch authored or co-authored using generative AI tooling?
    
    Generated-by: Claude (claude-sonnet-4-6)
    
    Closes #55721 from markj-db/recursive-file-lookup-error-class.
    
    Authored-by: Mark Jarvin <ma...@databricks.com>
    Signed-off-by: Wenchen Fan <we...@databricks.com>
---
 .../src/main/resources/error/error-conditions.json |  6 ++++++
 .../spark/sql/errors/QueryCompilationErrors.scala  |  6 ++++++
 .../sql/execution/datasources/DataSource.scala     |  5 +++++
 .../datasources/PartitioningAwareFileIndex.scala   |  4 ++--
 .../spark/sql/FileBasedDataSourceSuite.scala       | 23 ++++++++++++++++++++
 .../sql/execution/datasources/FileIndexSuite.scala | 25 +++++++++++++++++++++-
 6 files changed, 66 insertions(+), 3 deletions(-)

diff --git a/common/utils/src/main/resources/error/error-conditions.json b/common/utils/src/main/resources/error/error-conditions.json
index bbe5c3ced6f5..1d4f4317461f 100644
--- a/common/utils/src/main/resources/error/error-conditions.json
+++ b/common/utils/src/main/resources/error/error-conditions.json
@@ -6039,6 +6039,12 @@
     ],
     "sqlState" : "42836"
   },
+  "RECURSIVE_FILE_LOOKUP_NOT_SUPPORTED_FOR_PARTITIONED_DATA_SOURCE" : {
+    "message" : [
+      "Recursive file loading is not supported when the data source has explicit partition columns. Either remove the option \"recursiveFileLookup\", or read the data without supplying partition columns (for example, do not read a partitioned table)."
+    ],
+    "sqlState" : "0A000"
+  },
   "RECURSIVE_PROTOBUF_SCHEMA" : {
     "message" : [
       "Found recursive reference in Protobuf schema, which can not be processed by Spark by default: <fieldDescriptor>. try setting the option `recursive.fields.max.depth` 1 to 10. Going beyond 10 levels of recursion is not allowed."
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala
index 9cd94e984cae..9b899867a9e3 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala
@@ -3432,6 +3432,12 @@ private[sql] object QueryCompilationErrors extends QueryErrorsBase with Compilat
         "newPath" -> newPath.map(toSQLId).mkString(" -> ")))
   }
 
+  def recursiveFileLookupNotSupportedForPartitionedDataSourceError(): Throwable = {
+    new AnalysisException(
+      errorClass = "RECURSIVE_FILE_LOOKUP_NOT_SUPPORTED_FOR_PARTITIONED_DATA_SOURCE",
+      messageParameters = Map.empty)
+  }
+
   def notAllowedToCreatePermanentViewWithoutAssigningAliasForExpressionError(
       viewNameParts: Seq[String],
       attr: Attribute): Throwable = {
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 9b51d3763abb..4a95f681fb6e 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
@@ -420,6 +420,11 @@ case class DataSource(
           catalogTable.isDefined && catalogTable.get.tracksPartitionsInCatalog &&
           catalogTable.get.partitionColumnNames.nonEmpty
         val (fileCatalog, dataSchema, partitionSchema) = if (useCatalogFileIndex) {
+          if (caseInsensitiveOptions.getOrElse(
+              FileIndexOptions.RECURSIVE_FILE_LOOKUP, "false").toBoolean) {
+            throw QueryCompilationErrors
+              .recursiveFileLookupNotSupportedForPartitionedDataSourceError()
+          }
           val defaultTableSize = conf.defaultSizeInBytes
           val index = new CatalogFileIndex(
             sparkSession,
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/PartitioningAwareFileIndex.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/PartitioningAwareFileIndex.scala
index 1bf0d2f0301f..8cea2c95e694 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/PartitioningAwareFileIndex.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/PartitioningAwareFileIndex.scala
@@ -31,6 +31,7 @@ import org.apache.spark.sql.SparkSession
 import org.apache.spark.sql.catalyst.{expressions, InternalRow}
 import org.apache.spark.sql.catalyst.expressions._
 import org.apache.spark.sql.catalyst.util.CaseInsensitiveMap
+import org.apache.spark.sql.errors.QueryCompilationErrors
 import org.apache.spark.sql.types.StructType
 import org.apache.spark.util.ArrayImplicits._
 
@@ -89,8 +90,7 @@ abstract class PartitioningAwareFileIndex(
         PartitionDirectory(InternalRow.empty, allFiles().toArray.filter(isNonEmptyFile))) :: Nil
     } else {
       if (recursiveFileLookup) {
-        throw new IllegalArgumentException(
-          "Datasource with partition do not allow recursive file loading.")
+        throw QueryCompilationErrors.recursiveFileLookupNotSupportedForPartitionedDataSourceError()
       }
       prunePartitions(partitionFilters, partitionSpec()).map {
         case PartitionPath(values, path) =>
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/FileBasedDataSourceSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/FileBasedDataSourceSuite.scala
index 8aa6f5a5d0e6..1fc45e9703f9 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/FileBasedDataSourceSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/FileBasedDataSourceSuite.scala
@@ -878,6 +878,29 @@ class FileBasedDataSourceSuite extends SharedSparkSession
     assert(fileList.toSet === expectedFileList.toSet)
   }
 
+  test("recursiveFileLookup with a partitioned catalog table is rejected") {
+    withTable("part_tbl") {
+      sql(
+        """
+          |CREATE TABLE part_tbl (id INT, value STRING)
+          |USING parquet
+          |PARTITIONED BY (year INT)
+          |""".stripMargin)
+      sql("INSERT INTO part_tbl PARTITION (year = 2024) VALUES (1, 'a')")
+      sql("INSERT INTO part_tbl PARTITION (year = 2025) VALUES (2, 'b')")
+      checkError(
+        exception = intercept[AnalysisException] {
+          spark.read
+            .option("recursiveFileLookup", "true")
+            .table("part_tbl")
+            .collect()
+        },
+        condition = "RECURSIVE_FILE_LOOKUP_NOT_SUPPORTED_FOR_PARTITIONED_DATA_SOURCE",
+        parameters = Map.empty[String, String]
+      )
+    }
+  }
+
   test("Return correct results when data columns overlap with partition columns") {
     Seq("parquet", "orc", "json").foreach { format =>
       withTempPath { path =>
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/FileIndexSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/FileIndexSuite.scala
index 1150f6163b97..f4de8a52810e 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/FileIndexSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/FileIndexSuite.scala
@@ -31,7 +31,8 @@ import org.mockito.Mockito.{mock, when}
 
 import org.apache.spark.{SparkException, SparkRuntimeException}
 import org.apache.spark.metrics.source.HiveCatalogMetrics
-import org.apache.spark.sql.{Row, SparkSession}
+import org.apache.spark.sql.{AnalysisException, Row, SparkSession}
+import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.catalyst.util._
 import org.apache.spark.sql.functions.col
 import org.apache.spark.sql.internal.{SQLConf, StaticSQLConf}
@@ -657,6 +658,28 @@ class FileIndexSuite extends SharedSparkSession {
     assert(FileIndexOptions.isValidOption("pathglobfilter"))
   }
 
+  test("recursiveFileLookup with a user-specified partition spec is rejected") {
+    withTempDir { dir =>
+      val partitionSchema = StructType(Seq(StructField("year", IntegerType, nullable = true)))
+      val partitionSpec = PartitionSpec(
+        partitionSchema,
+        Seq(PartitionPath(InternalRow(2024), new Path(dir.getCanonicalPath))))
+      val fileIndex = new InMemoryFileIndex(
+        spark,
+        rootPathsSpecified = Seq(new Path(dir.getCanonicalPath)),
+        parameters = Map("recursiveFileLookup" -> "true"),
+        userSpecifiedSchema = None,
+        userSpecifiedPartitionSpec = Some(partitionSpec))
+      checkError(
+        exception = intercept[AnalysisException] {
+          fileIndex.listFiles(Nil, Nil)
+        },
+        condition = "RECURSIVE_FILE_LOOKUP_NOT_SUPPORTED_FOR_PARTITIONED_DATA_SOURCE",
+        parameters = Map.empty[String, String]
+      )
+    }
+  }
+
   test("SPARK-52339: Correctly compare root paths") {
     withTempDir { dir =>
       val file1 = new File(dir, "text1.txt")


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org