[HADP-56688] Add configuration for hive bucket definition support (apache#701)

Author: colinmjj

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

@@ -1522,6 +1522,13 @@ object SQLConf {
       .booleanConf
       .createWithDefault(true)
 
+  val HIVE_BUCKET_WRITE_COMPATIBLE =
+    buildConf("spark.sql.hive.bucketWriteCompatible")
+      .doc("When true, bucket number for hive table will be checked during the writing process")
+      .version("3.5.0")
+      .booleanConf
+      .createWithDefault(false)
+
   val HIVE_FILESOURCE_PARTITION_FILE_CACHE_SIZE =
     buildConf("spark.sql.hive.filesourcePartitionFileCacheSize")
       .doc("When nonzero, enable caching of partition file metadata in memory. All tables share " +
@@ -5761,6 +5768,8 @@ class SQLConf extends Serializable with Logging with SqlApiConf {
 
   def manageFilesourcePartitions: Boolean = getConf(HIVE_MANAGE_FILESOURCE_PARTITIONS)
 
+  def bucketWriteCompatible: Boolean = getConf(HIVE_BUCKET_WRITE_COMPATIBLE)
+
   def filesourcePartitionFileCacheSize: Long = getConf(HIVE_FILESOURCE_PARTITION_FILE_CACHE_SIZE)
 
   def caseSensitiveInferenceMode: HiveCaseSensitiveInferenceMode.Value =

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

@@ -196,11 +196,12 @@ private[hive] class HiveMetastoreCatalog(sparkSession: SparkSession) extends Log
       QualifiedTableName(relation.tableMeta.database, relation.tableMeta.identifier.table)
 
     val lazyPruningEnabled = sparkSession.sqlContext.conf.manageFilesourcePartitions
+    val bucketWriteCompatible = sparkSession.sqlContext.conf.bucketWriteCompatible
     val tablePath = new Path(relation.tableMeta.location)
     val fileFormat = fileFormatClass.getConstructor().newInstance()
     val bucketSpec = relation.tableMeta.bucketSpec
     val (hiveOptions, hiveBucketSpec) =
-      if (isWrite) {
+      if (isWrite && bucketWriteCompatible) {
         (options.updated(BucketingUtils.optionForHiveCompatibleBucketWrite, "true"),
           bucketSpec)
       } else {

sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/InsertIntoHiveTable.scala

@@ -338,7 +338,12 @@ object InsertIntoHiveTable extends V1WritesHiveUtils {
 
     val partitionColumns = getDynamicPartitionColumns(table, partition, query)
     val bucketSpec = table.bucketSpec
-    val options = getOptionsWithHiveBucketWrite(bucketSpec)
+    val bucketWriteCompatible = sparkSession.sqlContext.conf.bucketWriteCompatible
+    val options = if (bucketWriteCompatible) {
+      getOptionsWithHiveBucketWrite(bucketSpec)
+    } else {
+      Map.empty[String, String]
+    }
 
     new InsertIntoHiveTable(table, partition, query, overwrite, ifPartitionNotExists,
       outputColumnNames, partitionColumns, bucketSpec, options, fileFormat, hiveTempPath)

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

@@ -999,7 +999,8 @@ class InsertSuite extends QueryTest with TestHiveSingleton with BeforeAndAfter
            dataSource <- Seq("USING PARQUET", "STORED AS PARQUET")) {
         withSQLConf(SQLConf.COALESCE_PARTITIONS_ENABLED.key -> "false",
           SQLConf.COALESCE_PARTITIONS_MIN_PARTITION_NUM.key -> "2",
-          SQLConf.AUTO_REPARTITION_BEFORE_WRITING_ENABLED.key -> enabled) {
+          SQLConf.AUTO_REPARTITION_BEFORE_WRITING_ENABLED.key -> enabled,
+          SQLConf.HIVE_BUCKET_WRITE_COMPATIBLE.key -> "true") {
           withTable("part_tab") {
             sql(s"""CREATE TABLE part_tab(a int, b int, c int)
                  |$dataSource

sql/hive/src/test/scala/org/apache/spark/sql/sources/BucketedWriteWithHiveSupportSuite.scala

@@ -53,7 +53,8 @@ class BucketedWriteWithHiveSupportSuite extends BucketedWriteSuite with TestHive
       Seq("true", "false").foreach { enableConvertMetastore =>
         withSQLConf(HiveUtils.CONVERT_METASTORE_PARQUET.key -> enableConvertMetastore,
           HiveUtils.CONVERT_METASTORE_ORC.key -> enableConvertMetastore,
-          SQLConf.DYNAMIC_PARTITION_OVERWRITE_LOCK_ENABLED.key -> "true") {
+          SQLConf.DYNAMIC_PARTITION_OVERWRITE_LOCK_ENABLED.key -> "true",
+          SQLConf.HIVE_BUCKET_WRITE_COMPATIBLE.key -> "true") {
           withTable(table) {
             sql(
               s"""