add new config to enable the function

Author: brandonlin

sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala

@@ -3142,6 +3142,14 @@ object SQLConf {
     .booleanConf
     .createWithDefault(false)
 
+  val ENABLE_SUBDIRECTORY_SUPPORT_WITH_NON_PARTITIONED_TABLE =
+    buildConf("spark.sql.nonPartitionedTable.subdirectory.read.enabled")
+      .doc("When set to true, Spark SQL could read the files of " +
+        " Non-partitioned hive table from subdirectories under root path of table")
+      .version("3.2.0")
+      .booleanConf
+      .createWithDefault(false)
+
   /**
    * Holds information about keys that have been deprecated.
    *
@@ -3829,6 +3837,9 @@ class SQLConf extends Serializable with Logging {
 
   def legacyIntervalEnabled: Boolean = getConf(LEGACY_INTERVAL_ENABLED)
 
+  def nonPartitionedTableSubDirectoryReadSupport: Boolean =
+    getConf(ENABLE_SUBDIRECTORY_SUPPORT_WITH_NON_PARTITIONED_TABLE)
+
   /** ********************** SQLConf functionality methods ************ */
 
   /** Set Spark SQL configuration properties. */

sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveMetastoreCatalog.scala

@@ -280,10 +280,7 @@ private[hive] class HiveMetastoreCatalog(sparkSession: SparkSession) extends Log
 
   private def getDirectoryPathSeq(rootPath: Path): Seq[String] = {
     val enableSupportSubDirectories =
-      sparkSession.sparkContext.
-        hadoopConfiguration.getBoolean("hive.mapred.supports.subdirectories", false) &&
-        sparkSession.sparkContext.
-          hadoopConfiguration.getBoolean("mapred.input.dir.recursive", false)
+      sparkSession.sessionState.conf.nonPartitionedTableSubDirectoryReadSupport
 
     if (enableSupportSubDirectories) {
       val fs = rootPath.getFileSystem(sparkSession.sessionState.newHadoopConf())

sql/hive/src/test/scala/org/apache/spark/sql/hive/HiveMetastoreCatalogSuite.scala

@@ -367,28 +367,25 @@ class DataSourceWithHiveMetastoreCatalogSuite
   test("SPARK-28098: allow reader could read files from subdirectories") {
     withTempPath(dir => {
       withTable("dirTest") {
-        val testData = java.util.Arrays.asList(Row(1), Row(2), Row(3), Row(4), Row(5))
+        withSQLConf(
+          "spark.sql.nonPartitionedTable.subdirectory.read.enabled" -> "true") {
+          val testData = java.util.Arrays.asList(Row(1), Row(2), Row(3), Row(4), Row(5))
 
-        spark.sparkContext
-          .hadoopConfiguration.set("hive.mapred.supports.subdirectories", "true")
+          val dataFrame = spark.sqlContext
+            .createDataFrame(testData, StructType(Seq(StructField("val", IntegerType))))
 
-        spark.sparkContext
-          .hadoopConfiguration.set("mapred.input.dir.recursive", "true")
-
-        val dataFrame = spark.sqlContext
-          .createDataFrame(testData, StructType(Seq(StructField("val", IntegerType))))
-
-        dataFrame
-          .coalesce(1)
-          .write
-          .mode(SaveMode.Overwrite)
-          .format("orc")
-          .save(s"${dir.getCanonicalPath}/sub1/sub2")
+          dataFrame
+            .coalesce(1)
+            .write
+            .mode(SaveMode.Overwrite)
+            .format("orc")
+            .save(s"${dir.getCanonicalPath}/sub1/sub2")
 
-        spark.sql("CREATE EXTERNAL TABLE dirTest (val INT)" +
-          s" STORED AS ORC LOCATION '${dir.toURI}'")
+          spark.sql("CREATE EXTERNAL TABLE dirTest (val INT)" +
+            s" STORED AS ORC LOCATION '${dir.toURI}'")
 
-        checkAnswer(spark.sql("select * from dirTest"), dataFrame)
+          checkAnswer(spark.sql("select * from dirTest"), dataFrame)
+        }
       }
     })
   }