[HUDI-4902] Set default partitioner for SIMPLE BUCKET index (#6759)

Author: xicm

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieLayoutConfig.java

@@ -48,6 +48,9 @@ public class HoodieLayoutConfig extends HoodieConfig {
       .noDefaultValue()
       .withDocumentation("Partitioner class, it is used to distribute data in a specific way.");
 
+  public static final String SIMPLE_BUCKET_LAYOUT_PARTITIONER_CLASS_NAME =
+      "org.apache.hudi.table.action.commit.SparkBucketIndexPartitioner";
+
   private HoodieLayoutConfig() {
     super();
   }
@@ -80,8 +83,14 @@ public HoodieLayoutConfig build() {
     }
 
     private void setDefault() {
-      if (layoutConfig.contains(HoodieIndexConfig.INDEX_TYPE.key()) && layoutConfig.getString(HoodieIndexConfig.INDEX_TYPE.key()).equals(HoodieIndex.IndexType.BUCKET.name())) {
+      if (layoutConfig.contains(HoodieIndexConfig.INDEX_TYPE.key())
+          && layoutConfig.getString(HoodieIndexConfig.INDEX_TYPE.key()).equals(HoodieIndex.IndexType.BUCKET.name())) {
         layoutConfig.setDefaultValue(LAYOUT_TYPE, HoodieStorageLayout.LayoutType.BUCKET.name());
+
+        // Currently, the partitioner of the SIMPLE bucket index is supported by SparkBucketIndexPartitioner only.
+        if ("SIMPLE".equals(layoutConfig.getString(HoodieIndexConfig.BUCKET_INDEX_ENGINE_TYPE))) {
+          layoutConfig.setDefaultValue(LAYOUT_PARTITIONER_CLASS_NAME, SIMPLE_BUCKET_LAYOUT_PARTITIONER_CLASS_NAME);
+        }
       }
       layoutConfig.setDefaultValue(LAYOUT_TYPE, LAYOUT_TYPE.defaultValue());
     }

hudi-client/hudi-client-common/src/test/java/org/apache/hudi/config/TestHoodieWriteConfig.java

@@ -371,6 +371,23 @@ public void testAutoConcurrencyConfigAdjustmentWithMetadataTableDisabled(HoodieT
         HoodieFailedWritesCleaningPolicy.LAZY, FileSystemBasedLockProviderTestClass.class.getName());
   }
 
+  @Test
+  public void testSimpleBucketIndexPartitionerConfig() {
+    HoodieWriteConfig writeConfig = HoodieWriteConfig.newBuilder().withPath("/tmp")
+        .withIndexConfig(HoodieIndexConfig.newBuilder().withIndexType(HoodieIndex.IndexType.BUCKET)
+            .withBucketIndexEngineType(HoodieIndex.BucketIndexEngineType.SIMPLE).build())
+        .build();
+    assertEquals(HoodieLayoutConfig.SIMPLE_BUCKET_LAYOUT_PARTITIONER_CLASS_NAME, writeConfig.getString(HoodieLayoutConfig.LAYOUT_PARTITIONER_CLASS_NAME));
+
+    HoodieWriteConfig overwritePartitioner = HoodieWriteConfig.newBuilder().withPath("/tmp")
+        .withIndexConfig(HoodieIndexConfig.newBuilder().withIndexType(HoodieIndex.IndexType.BUCKET)
+            .withBucketIndexEngineType(HoodieIndex.BucketIndexEngineType.SIMPLE)
+            .build())
+        .withLayoutConfig(HoodieLayoutConfig.newBuilder().withLayoutPartitioner("org.apache.hudi.table.action.commit.UpsertPartitioner").build())
+        .build();
+    assertEquals("org.apache.hudi.table.action.commit.UpsertPartitioner", overwritePartitioner.getString(HoodieLayoutConfig.LAYOUT_PARTITIONER_CLASS_NAME));
+  }
+
   private HoodieWriteConfig createWriteConfig(Map<String, String> configs) {
     final Properties properties = new Properties();
     configs.forEach(properties::setProperty);