[spark][hive] Spark & Hive support writing to postpone bucket

Author: JingsongLi

docs/content/primary-key-table/data-distribution.md

@@ -93,8 +93,6 @@ Postpone bucket mode is configured by `'bucket' = '-2'`.
 This mode aims to solve the difficulty to determine a fixed number of buckets
 and support different buckets for different partitions.
 
-Currently, only Flink supports this mode.
-
 When writing records into the table,
 all records will first be stored in the `bucket-postpone` directory of each partition
 and are not available to readers.

paimon-hive/paimon-hive-connector-common/src/test/java/org/apache/paimon/hive/HiveWriteITCase.java

@@ -19,6 +19,7 @@
 package org.apache.paimon.hive;
 
 import org.apache.paimon.CoreOptions;
+import org.apache.paimon.Snapshot;
 import org.apache.paimon.catalog.Identifier;
 import org.apache.paimon.data.BinaryString;
 import org.apache.paimon.data.Decimal;
@@ -28,8 +29,10 @@
 import org.apache.paimon.data.InternalRow;
 import org.apache.paimon.hive.mapred.PaimonOutputFormat;
 import org.apache.paimon.hive.objectinspector.PaimonObjectInspectorFactory;
+import org.apache.paimon.io.DataFileMeta;
 import org.apache.paimon.options.CatalogOptions;
 import org.apache.paimon.options.Options;
+import org.apache.paimon.table.FileStoreTable;
 import org.apache.paimon.table.Table;
 import org.apache.paimon.table.sink.StreamTableCommit;
 import org.apache.paimon.table.sink.StreamTableWrite;
@@ -1171,4 +1174,40 @@ public void testInsertMapOfRowType() throws Exception {
         List<Object[]> expect = hiveShell.executeStatement("select * from " + tableName);
         assertThat(select.toArray()).containsExactlyInAnyOrder(expect.toArray());
     }
+
+    @Test
+    public void testInsertIntoPostponeBucket() throws Exception {
+        Options options = new Options();
+        options.set(CatalogOptions.WAREHOUSE, folder.newFolder().toURI().toString());
+        options.set(CoreOptions.BUCKET, -2);
+        options.set(CoreOptions.FILE_FORMAT, "parquet");
+        Identifier identifier = Identifier.create(DATABASE_NAME, "postpone_bucket");
+        FileStoreTable table =
+                (FileStoreTable)
+                        FileStoreTestUtils.createFileStoreTable(
+                                options,
+                                RowType.of(DataTypes.INT(), DataTypes.INT(), DataTypes.INT()),
+                                Collections.emptyList(),
+                                Collections.singletonList("f0"),
+                                identifier);
+        String tableName =
+                writeData(
+                        table,
+                        table.location().toString(),
+                        Collections.singletonList(GenericRow.of(1, 2, 3)));
+        hiveShell.execute("insert into " + tableName + " values (1,2,3),(4,5,6)");
+
+        Snapshot snapshot = table.latestSnapshot().get();
+        DataFileMeta file =
+                table.manifestFileReader()
+                        .read(
+                                table.manifestListReader()
+                                        .read(snapshot.deltaManifestList())
+                                        .get(0)
+                                        .fileName())
+                        .get(0)
+                        .file();
+        // default format for postpone bucket is avro
+        assertThat(file.fileName()).endsWith(".avro");
+    }
 }

paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonSparkWriter.scala

@@ -233,7 +233,9 @@ case class PaimonSparkWriter(table: FileStoreTable) {
         writeWithoutBucket(data)
 
       case HASH_FIXED =>
-        if (!paimonExtensionEnabled) {
+        if (table.bucketSpec().getNumBuckets == -2) {
+          writeWithoutBucket(data)
+        } else if (!paimonExtensionEnabled) {
           // Topology: input -> bucket-assigner -> shuffle by partition & bucket
           writeWithBucketProcessor(
             withInitBucketCol,

paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/SparkWriteITCase.scala

@@ -18,10 +18,13 @@
 
 package org.apache.paimon.spark.sql
 
+import org.apache.paimon.Snapshot
+import org.apache.paimon.io.DataFileMeta
 import org.apache.paimon.spark.PaimonSparkTestBase
 
 import org.apache.spark.SparkConf
 import org.apache.spark.sql.Row
+import org.assertj.core.api.Assertions.assertThat
 import org.junit.jupiter.api.Assertions
 
 import java.sql.Timestamp
@@ -38,6 +41,33 @@ class SparkWriteWithNoExtensionITCase extends SparkWriteITCase {
 
 class SparkWriteITCase extends PaimonSparkTestBase {
 
+  test("Paimon Write : Postpone Bucket") {
+    withTable("PostponeTable") {
+      spark.sql("""
+                  |CREATE TABLE PostponeTable (
+                  |  id INT,
+                  |  v1 INT,
+                  |  v2 INT
+                  |) TBLPROPERTIES (
+                  | 'bucket' = '-2',
+                  | 'primary-key' = 'id',
+                  | 'file.format' = 'parquet'
+                  |)
+                  |""".stripMargin)
+
+      spark.sql("INSERT INTO PostponeTable VALUES (1, 1, 1)")
+
+      val table = loadTable("PostponeTable")
+      val snapshot = table.latestSnapshot.get
+      val file = table.manifestFileReader
+        .read(table.manifestListReader.read(snapshot.deltaManifestList).get(0).fileName)
+        .get(0)
+        .file
+      // default format for postpone bucket is avro
+      assertThat(file.fileName).endsWith(".avro")
+    }
+  }
+
   test("Paimon Write: AllTypes") {
     withTable("AllTypesTable") {
       val createTableSQL =