[SPARK-32516][SQL] 'path' option cannot coexist with load()'s path parameters

sql/core/src/main/scala/org/apache/spark/sql/DataFrameReader.scala

@@ -229,6 +229,7 @@ class DataFrameReader private[sql](sparkSession: SparkSession) extends Logging {
    */
   def load(path: String): DataFrame = {
     // force invocation of `load(...varargs...)`
+    verifyPathOptionDoesNotExist()
     option("path", path).load(Seq.empty: _*)
   }
 
@@ -245,6 +246,10 @@ class DataFrameReader private[sql](sparkSession: SparkSession) extends Logging {
         "read files of Hive data source directly.")
     }
 
+    if (paths.nonEmpty) {
+      verifyPathOptionDoesNotExist()
+    }
+
     DataSource.lookupDataSourceV2(source, sparkSession.sessionState.conf).map { provider =>
       val catalogManager = sparkSession.sessionState.catalogManager
       val sessionOptions = DataSourceV2Utils.extractSessionConfigs(
@@ -297,6 +302,16 @@ class DataFrameReader private[sql](sparkSession: SparkSession) extends Logging {
         options = extraOptions.originalMap).resolveRelation())
   }
 
+  /**
+   * Called when load() takes non-empty path parameters.
+   */
+  private def verifyPathOptionDoesNotExist() = {
+    if (extraOptions.contains("path")) {
+      throw new AnalysisException("There is a path option set and load() is called with path " +
+        "parameters. Either remove the path option or put it into the load() parameters.")
+    }
+  }
+
   /**
    * Construct a `DataFrame` representing the database table accessible via JDBC URL
    * url named table and connection properties.

sql/core/src/test/scala/org/apache/spark/sql/FileBasedDataSourceSuite.scala

@@ -826,27 +826,6 @@ class FileBasedDataSourceSuite extends QueryTest
     }
   }
 
-  test("File table location should include both values of option `path` and `paths`") {
-    withSQLConf(SQLConf.USE_V1_SOURCE_LIST.key -> "") {
-      withTempPaths(3) { paths =>
-        paths.zipWithIndex.foreach { case (path, index) =>
-          Seq(index).toDF("a").write.mode("overwrite").parquet(path.getCanonicalPath)
-        }
-        val df = spark
-          .read
-          .option("path", paths.head.getCanonicalPath)
-          .parquet(paths(1).getCanonicalPath, paths(2).getCanonicalPath)
-        df.queryExecution.optimizedPlan match {
-          case PhysicalOperation(_, _, DataSourceV2ScanRelation(table: ParquetTable, _, _)) =>
-            assert(table.paths.toSet == paths.map(_.getCanonicalPath).toSet)
-          case _ =>
-            throw new AnalysisException("Can not match ParquetTable in the query.")
-        }
-        checkAnswer(df, Seq(0, 1, 2).map(Row(_)))
-      }
-    }
-  }
-
   test("SPARK-31935: Hadoop file system config should be effective in data source options") {
     Seq("parquet", "").foreach { format =>
       withSQLConf(

sql/core/src/test/scala/org/apache/spark/sql/test/DataFrameReaderWriterSuite.scala

@@ -224,15 +224,15 @@ class DataFrameReaderWriterSuite extends QueryTest with SharedSparkSession with
     assert(LastOptions.parameters("opt3") == "3")
   }
 
-  test("SPARK-32364: path argument of load function should override all existing options") {
+  test("SPARK-32364: option key should be case insensitive") {
     spark.read
       .format("org.apache.spark.sql.test")
       .option("paTh", "1")
       .option("PATH", "2")
       .option("Path", "3")
       .option("patH", "4")
-      .load("5")
-    assert(LastOptions.parameters("path") == "5")
+      .load()
+    assert(LastOptions.parameters("path") == "4")
   }
 
   test("SPARK-32364: path argument of save function should override all existing options") {
@@ -1105,4 +1105,20 @@ class DataFrameReaderWriterSuite extends QueryTest with SharedSparkSession with
       }
     }
   }
+
+  test("SPARK-32516: 'path' option cannot coexist with load()'s path parameters") {
+    withTempPath { dir =>
+      val path = dir.getAbsolutePath
+      Seq(1).toDF.write.mode("overwrite").parquet(path)
+
+      def verifyLoadFails(f: () => DataFrame): Unit = {
+        val e = intercept[AnalysisException](f())
+        assert(e.getMessage.contains(
+          "Either remove the path option or put it into the load() parameters"))
+      }
+
+      verifyLoadFails(() => spark.read.option("path", path).parquet(path))
+      verifyLoadFails(() => spark.read.option("path", path).format("parquet").load(path))
+    }
+  }
 }