[GitHub] [hudi] codope commented on a diff in pull request #4958: [HUDI-3558] Consistent bucket index: bucket resizing (split&merge) & concurrent write during resizing

[GitBox <gi...@apache.org>]

codope commented on code in PR #4958:
URL: https://github.com/apache/hudi/pull/4958#discussion_r917350371


##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieIndexConfig.java:
##########
@@ -252,12 +257,37 @@ public class HoodieIndexConfig extends HoodieConfig {
       .withDocumentation("Only applies if index type is BUCKET. Determine the number of buckets in the hudi table, "
           + "and each partition is divided to N buckets.");
 
+  public static final ConfigProperty<String> BUCKET_INDEX_MAX_NUM_BUCKETS = ConfigProperty
+      .key("hoodie.bucket.index.max.num.buckets")
+      .noDefaultValue()
+      .withDocumentation("Only applies if bucket index engine is consistent hashing. Determine the upper bound of "
+          + "the num of buckets in the hudi table. Resizing are not allowed to split bucket if the number is reached.");

Review Comment:
   ```suggestion
             + "the number of buckets in the hudi table. Bucket resizing cannot be done higher than this max limit.");
   ```



##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/index/bucket/HoodieSparkConsistentBucketIndex.java:
##########
@@ -67,6 +77,61 @@ public HoodieData<WriteStatus> updateLocation(HoodieData<WriteStatus> writeStatu
                                                 HoodieEngineContext context,
                                                 HoodieTable hoodieTable)
       throws HoodieIndexException {
+    throw new HoodieIndexException("Consistent hashing index does not support update location without the instant parameter");
+  }
+
+  /**
+   * Persist hashing metadata to storage. Only clustering operations will modify the metadata.
+   * For example, splitting & merging buckets, or just sorting and producing a new bucket.
+   */
+  @Override
+  public HoodieData<WriteStatus> updateLocation(HoodieData<WriteStatus> writeStatuses,
+                                                HoodieEngineContext context,
+                                                HoodieTable hoodieTable,
+                                                String instantTime)
+      throws HoodieIndexException {
+    HoodieInstant instant = hoodieTable.getMetaClient().getActiveTimeline().findInstantsAfterOrEquals(instantTime, 1).firstInstant().get();
+    ValidationUtils.checkState(instant.getTimestamp().equals(instantTime), "Cannot get the same instant, instantTime: " + instantTime);
+    if (!instant.getAction().equals(HoodieTimeline.REPLACE_COMMIT_ACTION)) {
+      return writeStatuses;
+    }
+
+    // Double-check if it is a clustering operation by trying to obtain the clustering plan
+    Option<Pair<HoodieInstant, HoodieClusteringPlan>> instantPlanPair =
+        ClusteringUtils.getClusteringPlan(hoodieTable.getMetaClient(), HoodieTimeline.getReplaceCommitRequestedInstant(instantTime));
+    if (!instantPlanPair.isPresent()) {

Review Comment:
   Why is this double-checking necessary? I think we can avoid the first though it does not cost much. 



##########
hudi-client/hudi-client-common/src/test/java/org/apache/hudi/index/bucket/TestConsistentBucketIdIdentifier.java:
##########
@@ -76,4 +96,90 @@ public void testGetBucket() {
     Assertions.assertEquals(nodes.get(1), identifier.getBucketByFileId(FSUtils.createNewFileId("1", 0)));
     Assertions.assertEquals(nodes.get(2), identifier.getBucketByFileId(FSUtils.createNewFileId("2", 0)));
   }
+
+  /**
+   * @param v0  first node hash value
+   * @param v1  second node hash value
+   * @param mid mid node hash value generated by split the first bucket v0
+   */
+  @ParameterizedTest
+  @MethodSource("splitBucketParams")
+  public void testSplitBucket(int v0, int v1, int mid) {
+    // Hash range mapping:: [0, 0xf], (0xf, MAX]
+    List<ConsistentHashingNode> nodes = Arrays.asList(
+        new ConsistentHashingNode(v0, "0"),
+        new ConsistentHashingNode(v1, "1"));
+    HoodieConsistentHashingMetadata meta = new HoodieConsistentHashingMetadata((short) 0, "", "", 4, 0, nodes);
+    Option<List<ConsistentHashingNode>> res = new ConsistentBucketIdentifier(meta).splitBucket(nodes.get(0));
+    if (mid < 0) {
+      Assertions.assertTrue(!res.isPresent());
+      return;
+    }
+
+    List<ConsistentHashingNode> childNodes = res.get();
+    Assertions.assertEquals(2, childNodes.size());
+    Assertions.assertTrue(childNodes.stream().allMatch(c -> c.getTag() == ConsistentHashingNode.NodeTag.REPLACE));
+    Assertions.assertEquals(mid, childNodes.get(0).getValue());
+    Assertions.assertEquals(nodes.get(0).getValue(), childNodes.get(1).getValue());
+  }
+
+  @Test
+  public void testMerge() {
+    HoodieConsistentHashingMetadata meta = new HoodieConsistentHashingMetadata("partition", 8);
+    List<ConsistentHashingNode> nodes = meta.getNodes();
+
+    List<String> fileIds = IntStream.range(0, 3).mapToObj(i -> FSUtils.createNewFileId(nodes.get(i).getFileIdPrefix(), 0)).collect(Collectors.toList());
+    List<ConsistentHashingNode> childNodes = new ConsistentBucketIdentifier(meta).mergeBucket(fileIds);
+    Assertions.assertEquals(ConsistentHashingNode.NodeTag.DELETE, childNodes.get(0).getTag());
+    Assertions.assertEquals(ConsistentHashingNode.NodeTag.DELETE, childNodes.get(1).getTag());
+    Assertions.assertEquals(ConsistentHashingNode.NodeTag.REPLACE, childNodes.get(2).getTag());
+    Assertions.assertEquals(nodes.get(2).getValue(), childNodes.get(2).getValue());
+    Assertions.assertNotEquals(nodes.get(2).getFileIdPrefix(), childNodes.get(2).getFileIdPrefix());
+
+    fileIds = Arrays.asList(nodes.get(7), nodes.get(0), nodes.get(1)).stream()
+        .map(ConsistentHashingNode::getFileIdPrefix).map(f -> FSUtils.createNewFileId(f, 0)).collect(Collectors.toList());
+    childNodes = new ConsistentBucketIdentifier(meta).mergeBucket(fileIds);
+    Assertions.assertEquals(ConsistentHashingNode.NodeTag.DELETE, childNodes.get(0).getTag());
+    Assertions.assertEquals(ConsistentHashingNode.NodeTag.DELETE, childNodes.get(1).getTag());
+    Assertions.assertEquals(ConsistentHashingNode.NodeTag.REPLACE, childNodes.get(2).getTag());
+    Assertions.assertEquals(nodes.get(1).getValue(), childNodes.get(2).getValue());
+    Assertions.assertNotEquals(nodes.get(1).getFileIdPrefix(), childNodes.get(2).getFileIdPrefix());
+
+    boolean exception = false;
+    try {
+      fileIds = IntStream.range(0, 2).mapToObj(i -> FSUtils.createNewFileId(nodes.get(i * 2).getFileIdPrefix(), 0)).collect(Collectors.toList());
+      new ConsistentBucketIdentifier(meta).mergeBucket(fileIds);
+    } catch (Exception e) {
+      exception = true;
+    }
+    Assertions.assertTrue(exception);
+  }
+
+  @Test
+  public void testChildrenNodesInitialization() {
+    HoodieConsistentHashingMetadata metadata = new HoodieConsistentHashingMetadata("partition", 8);
+    List<ConsistentHashingNode> childrenNodes = new ArrayList<>();
+    childrenNodes.add(new ConsistentHashingNode(metadata.getNodes().get(0).getValue(), "d1", ConsistentHashingNode.NodeTag.DELETE));
+    childrenNodes.add(new ConsistentHashingNode(1024, "a1", ConsistentHashingNode.NodeTag.REPLACE));
+    childrenNodes.add(new ConsistentHashingNode(metadata.getNodes().get(1).getValue(), "a2", ConsistentHashingNode.NodeTag.REPLACE));
+    metadata.setChildrenNodes(childrenNodes);
+
+    ConsistentBucketIdentifier identifier = new ConsistentBucketIdentifier(metadata);
+    List<ConsistentHashingNode> nodes = new ArrayList<>(identifier.getNodes());
+    Assertions.assertEquals(1024, nodes.get(0).getValue());
+    Assertions.assertEquals("a1", nodes.get(0).getFileIdPrefix());
+    Assertions.assertEquals(metadata.getNodes().get(1).getValue(), nodes.get(1).getValue());
+    Assertions.assertEquals("a2", nodes.get(1).getFileIdPrefix());
+
+    childrenNodes = new ArrayList<>();
+    childrenNodes.add(new ConsistentHashingNode(metadata.getNodes().get(0).getValue(), "d1", ConsistentHashingNode.NodeTag.NORMAL));
+    metadata.setChildrenNodes(childrenNodes);
+    boolean isException = false;
+    try {
+      identifier = new ConsistentBucketIdentifier(metadata);
+    } catch (Exception e) {
+      isException = true;
+    }
+    Assertions.assertEquals(true, isException);

Review Comment:
   same here. probably a separate test for failure scenarios would be better.



##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieIndexConfig.java:
##########
@@ -252,12 +257,37 @@ public class HoodieIndexConfig extends HoodieConfig {
       .withDocumentation("Only applies if index type is BUCKET. Determine the number of buckets in the hudi table, "
           + "and each partition is divided to N buckets.");
 
+  public static final ConfigProperty<String> BUCKET_INDEX_MAX_NUM_BUCKETS = ConfigProperty
+      .key("hoodie.bucket.index.max.num.buckets")
+      .noDefaultValue()
+      .withDocumentation("Only applies if bucket index engine is consistent hashing. Determine the upper bound of "
+          + "the num of buckets in the hudi table. Resizing are not allowed to split bucket if the number is reached.");
+
+  public static final ConfigProperty<String> BUCKET_INDEX_MIN_NUM_BUCKETS = ConfigProperty
+      .key("hoodie.bucket.index.min.num.buckets")
+      .noDefaultValue()
+      .withDocumentation("Only applies if bucket index engine is consistent hashing. Determine the lower bound of "
+          + "the num of buckets in the hudi table. Resizing are not allowed to merge buckets if the number is reached.");
+
   public static final ConfigProperty<String> BUCKET_INDEX_HASH_FIELD = ConfigProperty
       .key("hoodie.bucket.index.hash.field")
       .noDefaultValue()
       .withDocumentation("Index key. It is used to index the record and find its file group. "
           + "If not set, use record key field as default");
 
+  public static final ConfigProperty<Double> BUCKET_SPLIT_THRESHOLD = ConfigProperty
+      .key("hoodie.bucket.index.split.threshold")
+      .defaultValue(2.0)
+      .withDocumentation("Control if the bucket should be split when using consistent hashing bucket index."
+          + "Specifically, if a file slice size reaches `hoodie.xxxx.max.file.size` * threshold, then split will be carried out.");
+
+  public static final ConfigProperty<Double> BUCKET_MERGE_THRESHOLD = ConfigProperty
+      .key("hoodie.bucket.index.merge.threshold")
+      .defaultValue(0.2)
+      .withDocumentation("Control if buckets should be merged when using consistent hashing bucket index"
+          + "Specifically, if a file slice size is smaller than `hoodie.xxxx.max.file.size` * threshold, then it will be considered"
+          + "as a merge condidate.");

Review Comment:
   ```suggestion
             + "as a merge candidate.");
   ```



##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/execution/bulkinsert/RDDConsistentBucketPartitioner.java:
##########
@@ -0,0 +1,280 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.execution.bulkinsert;
+
+import org.apache.hudi.avro.HoodieAvroUtils;
+import org.apache.hudi.common.config.SerializableSchema;
+import org.apache.hudi.common.fs.FSUtils;
+import org.apache.hudi.common.model.ConsistentHashingNode;
+import org.apache.hudi.common.model.HoodieConsistentHashingMetadata;
+import org.apache.hudi.common.model.HoodieKey;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.model.HoodieRecordPayload;
+import org.apache.hudi.common.model.HoodieTableType;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.ValidationUtils;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.index.bucket.ConsistentBucketIdentifier;
+import org.apache.hudi.index.bucket.HoodieSparkConsistentBucketIndex;
+import org.apache.hudi.io.AppendHandleFactory;
+import org.apache.hudi.io.SingleFileHandleCreateFactory;
+import org.apache.hudi.io.WriteHandleFactory;
+import org.apache.hudi.table.HoodieTable;
+
+import org.apache.avro.Schema;
+import org.apache.log4j.LogManager;
+import org.apache.log4j.Logger;
+import org.apache.spark.Partitioner;
+import org.apache.spark.api.java.JavaRDD;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import scala.Tuple2;
+
+import static org.apache.hudi.config.HoodieClusteringConfig.PLAN_STRATEGY_SORT_COLUMNS;
+
+/**
+ * A partitioner for (consistent hashing) bucket index used in bulk_insert
+ */
+public class RDDConsistentBucketPartitioner<T extends HoodieRecordPayload> extends RDDBucketIndexPartitioner<T> {
+
+  private static final Logger LOG = LogManager.getLogger(RDDConsistentBucketPartitioner.class);
+
+  private final HoodieTable table;
+  private final HoodieWriteConfig config;
+  private final List<String> indexKeyFields;
+  private final Map<String, List<ConsistentHashingNode>> hashingChildrenNodes;
+  private final String[] sortColumnNames;
+  private final boolean preserveHoodieMetadata;
+  private final boolean consistentLogicalTimestampEnabled;
+
+  private List<Boolean> doAppend;
+  private List<String> fileIdPfxList;
+
+  public RDDConsistentBucketPartitioner(HoodieTable table, HoodieWriteConfig config, Map<String, String> strategyParams, boolean preserveHoodieMetadata) {
+    this.table = table;
+    this.config = config;
+    this.indexKeyFields = Arrays.asList(config.getBucketIndexHashField().split(","));
+    this.hashingChildrenNodes = new HashMap<>();
+    this.consistentLogicalTimestampEnabled = config.isConsistentLogicalTimestampEnabled();
+    this.preserveHoodieMetadata = preserveHoodieMetadata;
+
+    if (strategyParams.containsKey(PLAN_STRATEGY_SORT_COLUMNS.key())) {
+      sortColumnNames = strategyParams.get(PLAN_STRATEGY_SORT_COLUMNS.key()).split(",");
+    } else {
+      sortColumnNames = null;
+    }
+  }
+
+  public RDDConsistentBucketPartitioner(HoodieTable table, HoodieWriteConfig config) {
+    this(table, config, Collections.emptyMap(), false);
+    ValidationUtils.checkArgument(table.getIndex() instanceof HoodieSparkConsistentBucketIndex,
+        "RDDConsistentBucketPartitioner can only be used together with consistent hashing bucket index");
+    ValidationUtils.checkArgument(table.getMetaClient().getTableType().equals(HoodieTableType.MERGE_ON_READ),
+        "CoW table with bucket index doesn't support bulk_insert");
+  }
+
+  /**
+   * Repartition the records to conform the bucket index storage layout constraints.
+   * Specifically, partition the records based on consistent bucket index, which is computed
+   * using hashing metadata and records' key.
+   *
+   * @param records               Input Hoodie records
+   * @param outputSparkPartitions Not used
+   * @return partitioned records, each partition of data corresponds to a bucket (i.e., file group)
+   */
+  @Override
+  public JavaRDD<HoodieRecord<T>> repartitionRecords(JavaRDD<HoodieRecord<T>> records, int outputSparkPartitions) {
+    Map<String, ConsistentBucketIdentifier> partitionToIdentifier = initializeBucketIdentifier(records);
+    Map<String, Map<String, Integer>> partitionToFileIdPfxIdxMap = generateFileIdPfx(partitionToIdentifier, outputSparkPartitions);
+    return doPartition(records, new Partitioner() {
+      @Override
+      public int numPartitions() {
+        return fileIdPfxList.size();
+      }
+
+      @Override
+      public int getPartition(Object key) {
+        HoodieKey hoodieKey = (HoodieKey) key;
+        String partition = hoodieKey.getPartitionPath();
+        ConsistentHashingNode node = partitionToIdentifier.get(partition).getBucket(hoodieKey, indexKeyFields);
+        return partitionToFileIdPfxIdxMap.get(partition).get(node.getFileIdPrefix());
+      }
+    });
+  }
+
+  @Override
+  public boolean arePartitionRecordsSorted() {
+    return (sortColumnNames != null && sortColumnNames.length > 0)
+        || table.requireSortedRecords() || config.getBulkInsertSortMode() != BulkInsertSortMode.NONE;
+  }
+
+  @Override
+  public Option<WriteHandleFactory> getWriteHandleFactory(int idx) {
+    return doAppend.get(idx) ? Option.of(new AppendHandleFactory()) :
+        Option.of(new SingleFileHandleCreateFactory(FSUtils.createNewFileId(getFileIdPfx(idx), 0), this.preserveHoodieMetadata));
+  }
+
+  @Override
+  public String getFileIdPfx(int partitionId) {
+    return fileIdPfxList.get(partitionId);
+  }
+
+  public void addHashingChildrenNodes(String partition, List<ConsistentHashingNode> nodes) {
+    ValidationUtils.checkState(nodes.stream().noneMatch(n -> n.getTag() == ConsistentHashingNode.NodeTag.NORMAL), "children nodes should not be tagged as NORMAL");
+    hashingChildrenNodes.put(partition, nodes);
+  }
+
+  /**
+   * Get (construct) the bucket identifier of the given partition
+   */
+  private ConsistentBucketIdentifier getBucketIdentifier(String partition) {
+    HoodieSparkConsistentBucketIndex index = (HoodieSparkConsistentBucketIndex) table.getIndex();
+    HoodieConsistentHashingMetadata metadata = index.loadOrCreateMetadata(this.table, partition);
+    if (hashingChildrenNodes.containsKey(partition)) {
+      metadata.setChildrenNodes(hashingChildrenNodes.get(partition));
+    }
+    return new ConsistentBucketIdentifier(metadata);
+  }
+
+  /**
+   * Initialize hashing metadata of input records. The metadata of all related partitions will be loaded, and
+   * the mapping from partition to its bucket identifier is constructed.
+   */
+  private Map<String, ConsistentBucketIdentifier> initializeBucketIdentifier(JavaRDD<HoodieRecord<T>> records) {
+    return records.map(HoodieRecord::getPartitionPath).distinct().collect().stream()
+        .collect(Collectors.toMap(p -> p, p -> getBucketIdentifier(p)));
+  }
+
+  /**
+   * Initialize fileIdPfx for each data partition. Specifically, the following fields is constructed:
+   * - fileIdPfx: the Nth element corresponds to the Nth data partition, indicating its fileIdPfx
+   * - doAppend: represents if the Nth data partition should use AppendHandler
+   * - partitionToFileIdPfxIdxMap (return value): (table partition) -> (fileIdPfx -> idx) mapping
+   *
+   * @param parallelism Not used, the actual parallelism is determined by the bucket number

Review Comment:
   let's remove it if not used



##########
hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/clustering/plan/strategy/TestSparkConsistentBucketClusteringPlanStrategy.java:
##########
@@ -0,0 +1,190 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.client.clustering.plan.strategy;
+
+import org.apache.hudi.avro.model.HoodieClusteringGroup;
+import org.apache.hudi.common.fs.FSUtils;
+import org.apache.hudi.common.model.ConsistentHashingNode;
+import org.apache.hudi.common.model.FileSlice;
+import org.apache.hudi.common.model.HoodieBaseFile;
+import org.apache.hudi.common.model.HoodieConsistentHashingMetadata;
+import org.apache.hudi.common.model.HoodieLogFile;
+import org.apache.hudi.common.model.HoodieTableType;
+import org.apache.hudi.common.testutils.HoodieTestUtils;
+import org.apache.hudi.common.util.collection.Triple;
+import org.apache.hudi.config.HoodieIndexConfig;
+import org.apache.hudi.config.HoodieStorageConfig;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.index.HoodieIndex;
+import org.apache.hudi.index.bucket.ConsistentBucketIdentifier;
+import org.apache.hudi.table.HoodieSparkTable;
+import org.apache.hudi.table.HoodieTable;
+import org.apache.hudi.testutils.HoodieClientTestHarness;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Random;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import static org.apache.hudi.config.HoodieIndexConfig.BUCKET_MERGE_THRESHOLD;
+import static org.apache.hudi.config.HoodieIndexConfig.BUCKET_SPLIT_THRESHOLD;
+
+public class TestSparkConsistentBucketClusteringPlanStrategy extends HoodieClientTestHarness {
+
+  private final Random random = new Random();
+
+  private void setup() throws IOException {
+    initPath();
+    initSparkContexts();
+    initFileSystem();
+    metaClient = HoodieTestUtils.init(hadoopConf, basePath, HoodieTableType.MERGE_ON_READ);
+  }
+
+  @AfterEach
+  public void tearDown() throws IOException {
+    cleanupResources();
+  }
+
+  @Test
+  public void testBuildSplitClusteringGroup() throws IOException {
+    setup();
+    int maxFileSize = 5120;
+    HoodieWriteConfig config = HoodieWriteConfig.newBuilder().withPath(basePath)
+        .withIndexConfig(HoodieIndexConfig.newBuilder().withIndexType(HoodieIndex.IndexType.BUCKET)
+            .withBucketIndexEngineType(HoodieIndex.BucketIndexEngineType.CONSISTENT_HASHING)
+            .withBucketMaxNum(6)
+            .withBucketNum("4").build())
+        .withStorageConfig(HoodieStorageConfig.newBuilder()
+            .parquetMaxFileSize(maxFileSize).build())
+        .build();
+
+    HoodieTable hoodieTable = HoodieSparkTable.create(config, context, metaClient);
+    SparkConsistentBucketClusteringPlanStrategy planStrategy = new SparkConsistentBucketClusteringPlanStrategy(hoodieTable, context, config);
+
+    HoodieConsistentHashingMetadata metadata = new HoodieConsistentHashingMetadata("partition", config.getBucketIndexNumBuckets());
+    ConsistentBucketIdentifier identifier = new ConsistentBucketIdentifier(metadata);
+
+    int[] fsSize = {maxFileSize * 5, (int) (maxFileSize * BUCKET_SPLIT_THRESHOLD.defaultValue() + 1), maxFileSize, maxFileSize * 5};
+    List<FileSlice> fileSlices = IntStream.range(0, metadata.getNodes().size()).mapToObj(
+        i -> createFileSliceWithSize(metadata.getNodes().get(i).getFileIdPrefix(), 1024, fsSize[i] - 1024)
+    ).collect(Collectors.toList());
+
+    /**
+     * 1. Test split candidate selection based on file size
+     * 2. Test the effectiveness of split slot
+     */
+    Triple res = planStrategy.buildSplitClusteringGroups(identifier, fileSlices, 2);
+    Assertions.assertEquals(2, res.getMiddle());
+    List<HoodieClusteringGroup> groups = (List<HoodieClusteringGroup>) res.getLeft();
+    Assertions.assertEquals(2, groups.size());
+    Assertions.assertEquals(fileSlices.get(0).getFileId(), groups.get(0).getSlices().get(0).getFileId());
+    Assertions.assertEquals(fileSlices.get(1).getFileId(), groups.get(1).getSlices().get(0).getFileId());
+    List<FileSlice> fsUntouched = (List<FileSlice>) res.getRight();
+    Assertions.assertEquals(2, fsUntouched.size());
+    Assertions.assertEquals(fileSlices.get(2), fsUntouched.get(0));
+    Assertions.assertEquals(fileSlices.get(3), fsUntouched.get(1));
+  }
+
+  @Test
+  public void testBuildMergeClusteringGroup() throws Exception {

Review Comment:
   This test flakes in my local runs. Can you please check?
   <img width="1722" alt="Screenshot 2022-07-10 at 5 16 15 PM" src="https://user-images.githubusercontent.com/16440354/178143560-41b23226-64b5-40ee-b821-4a875b4815ca.png">
   



##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieIndexConfig.java:
##########
@@ -252,12 +257,37 @@ public class HoodieIndexConfig extends HoodieConfig {
       .withDocumentation("Only applies if index type is BUCKET. Determine the number of buckets in the hudi table, "
           + "and each partition is divided to N buckets.");
 
+  public static final ConfigProperty<String> BUCKET_INDEX_MAX_NUM_BUCKETS = ConfigProperty
+      .key("hoodie.bucket.index.max.num.buckets")
+      .noDefaultValue()
+      .withDocumentation("Only applies if bucket index engine is consistent hashing. Determine the upper bound of "
+          + "the num of buckets in the hudi table. Resizing are not allowed to split bucket if the number is reached.");
+
+  public static final ConfigProperty<String> BUCKET_INDEX_MIN_NUM_BUCKETS = ConfigProperty
+      .key("hoodie.bucket.index.min.num.buckets")
+      .noDefaultValue()
+      .withDocumentation("Only applies if bucket index engine is consistent hashing. Determine the lower bound of "
+          + "the num of buckets in the hudi table. Resizing are not allowed to merge buckets if the number is reached.");

Review Comment:
   ```suggestion
             + "the number of buckets in the hudi table. Bucket resizing cannot be done lower than this min limit.");
   ```



##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java:
##########
@@ -2600,6 +2629,18 @@ private void validate() {
         ValidationUtils.checkArgument(!writeConfig.getString(HoodieCompactionConfig.FAILED_WRITES_CLEANER_POLICY)
             .equals(HoodieFailedWritesCleaningPolicy.EAGER.name()), "To enable optimistic concurrency control, set hoodie.cleaner.policy.failed.writes=LAZY");
       }
+
+      if (writeConfig.getIndexType() == HoodieIndex.IndexType.BUCKET && writeConfig.getBucketIndexEngineType() == HoodieIndex.BucketIndexEngineType.CONSISTENT_HASHING) {

Review Comment:
   This is not validation per se. Let's extract this whole logic to a separate method and call that method in `setDefaults()`. Also, I think a better place to do this is in `HoodieIndexConfig`. We can keep the getters in `HoodieWriteConfig`.



##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/clustering/plan/strategy/SparkConsistentBucketClusteringPlanStrategy.java:
##########
@@ -0,0 +1,332 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.client.clustering.plan.strategy;
+
+import org.apache.hudi.avro.model.HoodieClusteringGroup;
+import org.apache.hudi.client.WriteStatus;
+import org.apache.hudi.common.engine.HoodieEngineContext;
+import org.apache.hudi.common.fs.FSUtils;
+import org.apache.hudi.common.model.ConsistentHashingNode;
+import org.apache.hudi.common.model.FileSlice;
+import org.apache.hudi.common.model.HoodieConsistentHashingMetadata;
+import org.apache.hudi.common.model.HoodieFileFormat;
+import org.apache.hudi.common.model.HoodieKey;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.model.HoodieRecordPayload;
+import org.apache.hudi.common.table.timeline.HoodieTimeline;
+import org.apache.hudi.common.table.view.SyncableFileSystemView;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.StringUtils;
+import org.apache.hudi.common.util.ValidationUtils;
+import org.apache.hudi.common.util.collection.Triple;
+import org.apache.hudi.config.HoodieClusteringConfig;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.exception.HoodieClusteringException;
+import org.apache.hudi.index.bucket.ConsistentBucketIdentifier;
+import org.apache.hudi.index.bucket.HoodieSparkConsistentBucketIndex;
+import org.apache.hudi.table.HoodieTable;
+import org.apache.hudi.table.action.cluster.strategy.PartitionAwareClusteringPlanStrategy;
+
+import org.apache.log4j.LogManager;
+import org.apache.log4j.Logger;
+import org.apache.spark.api.java.JavaRDD;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import java.util.stream.Stream;
+
+/**
+ * Clustering plan strategy specifically for consistent bucket index
+ */
+public class SparkConsistentBucketClusteringPlanStrategy<T extends HoodieRecordPayload<T>>
+    extends PartitionAwareClusteringPlanStrategy<T, JavaRDD<HoodieRecord<T>>, JavaRDD<HoodieKey>, JavaRDD<WriteStatus>> {
+
+  private static final Logger LOG = LogManager.getLogger(SparkConsistentBucketClusteringPlanStrategy.class);
+
+  public static final String METADATA_PARTITION_KEY = "clustering.group.partition";
+  public static final String METADATA_CHILD_NODE_KEY = "clustering.group.child.node";
+  public static final String METADATA_SEQUENCE_NUMBER_KEY = "clustering.group.sequence.no";
+
+  public SparkConsistentBucketClusteringPlanStrategy(HoodieTable table, HoodieEngineContext engineContext, HoodieWriteConfig writeConfig) {
+    super(table, engineContext, writeConfig);
+    validate();
+  }
+
+  /**
+   * TODO maybe add force config to schedule the clustering. It could allow clustering on partitions that are not doing write operation.
+   * Block clustering if there is any ongoing concurrent writers
+   *
+   * @return true if the schedule can proceed
+   */
+  @Override
+  public boolean checkPrecondition() {
+    HoodieTimeline timeline = getHoodieTable().getActiveTimeline().getDeltaCommitTimeline().filterInflightsAndRequested();

Review Comment:
   Why only considering delta commit timeline? Why not write commit timeline?



##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/clustering/plan/strategy/SparkConsistentBucketClusteringPlanStrategy.java:
##########
@@ -0,0 +1,332 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.client.clustering.plan.strategy;
+
+import org.apache.hudi.avro.model.HoodieClusteringGroup;
+import org.apache.hudi.client.WriteStatus;
+import org.apache.hudi.common.engine.HoodieEngineContext;
+import org.apache.hudi.common.fs.FSUtils;
+import org.apache.hudi.common.model.ConsistentHashingNode;
+import org.apache.hudi.common.model.FileSlice;
+import org.apache.hudi.common.model.HoodieConsistentHashingMetadata;
+import org.apache.hudi.common.model.HoodieFileFormat;
+import org.apache.hudi.common.model.HoodieKey;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.model.HoodieRecordPayload;
+import org.apache.hudi.common.table.timeline.HoodieTimeline;
+import org.apache.hudi.common.table.view.SyncableFileSystemView;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.StringUtils;
+import org.apache.hudi.common.util.ValidationUtils;
+import org.apache.hudi.common.util.collection.Triple;
+import org.apache.hudi.config.HoodieClusteringConfig;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.exception.HoodieClusteringException;
+import org.apache.hudi.index.bucket.ConsistentBucketIdentifier;
+import org.apache.hudi.index.bucket.HoodieSparkConsistentBucketIndex;
+import org.apache.hudi.table.HoodieTable;
+import org.apache.hudi.table.action.cluster.strategy.PartitionAwareClusteringPlanStrategy;
+
+import org.apache.log4j.LogManager;
+import org.apache.log4j.Logger;
+import org.apache.spark.api.java.JavaRDD;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import java.util.stream.Stream;
+
+/**
+ * Clustering plan strategy specifically for consistent bucket index
+ */
+public class SparkConsistentBucketClusteringPlanStrategy<T extends HoodieRecordPayload<T>>
+    extends PartitionAwareClusteringPlanStrategy<T, JavaRDD<HoodieRecord<T>>, JavaRDD<HoodieKey>, JavaRDD<WriteStatus>> {
+
+  private static final Logger LOG = LogManager.getLogger(SparkConsistentBucketClusteringPlanStrategy.class);
+
+  public static final String METADATA_PARTITION_KEY = "clustering.group.partition";
+  public static final String METADATA_CHILD_NODE_KEY = "clustering.group.child.node";
+  public static final String METADATA_SEQUENCE_NUMBER_KEY = "clustering.group.sequence.no";
+
+  public SparkConsistentBucketClusteringPlanStrategy(HoodieTable table, HoodieEngineContext engineContext, HoodieWriteConfig writeConfig) {
+    super(table, engineContext, writeConfig);
+    validate();
+  }
+
+  /**
+   * TODO maybe add force config to schedule the clustering. It could allow clustering on partitions that are not doing write operation.
+   * Block clustering if there is any ongoing concurrent writers
+   *
+   * @return true if the schedule can proceed
+   */
+  @Override
+  public boolean checkPrecondition() {
+    HoodieTimeline timeline = getHoodieTable().getActiveTimeline().getDeltaCommitTimeline().filterInflightsAndRequested();
+    if (!timeline.empty()) {
+      LOG.warn("When using consistent bucket, clustering cannot be scheduled async if there are concurrent writers. "
+          + "Writer instant: " + timeline.getInstants().collect(Collectors.toList()));
+      return false;
+    }
+    return true;
+  }
+
+  /**
+   * Generate candidate clustering file slices of the given partition.
+   * If there is inflight / requested clustering working on the partition, then return empty list
+   * to ensure serialized update to the hashing metadata.
+   *
+   * @return candidate file slices to be clustered (i.e., sort, bucket split or merge)
+   */
+  @Override
+  protected Stream<FileSlice> getFileSlicesEligibleForClustering(String partition) {
+    SyncableFileSystemView fileSystemView = (SyncableFileSystemView) getHoodieTable().getSliceView();
+    boolean isPartitionInClustering = fileSystemView.getFileGroupsInPendingClustering().anyMatch(p -> p.getLeft().getPartitionPath().equals(partition));
+    if (isPartitionInClustering) {
+      LOG.info("Partition: " + partition + " is already in clustering, skip");
+      return Stream.empty();
+    }
+
+    return super.getFileSlicesEligibleForClustering(partition);
+  }
+
+  @Override
+  protected Map<String, String> getStrategyParams() {
+    Map<String, String> params = new HashMap<>();
+    if (!StringUtils.isNullOrEmpty(getWriteConfig().getClusteringSortColumns())) {
+      params.put(HoodieClusteringConfig.PLAN_STRATEGY_SORT_COLUMNS.key(), getWriteConfig().getClusteringSortColumns());
+    }
+    return params;
+  }
+
+  /**
+   * Generate cluster group based on split, merge and sort rules
+   */
+  @Override
+  protected Stream<HoodieClusteringGroup> buildClusteringGroupsForPartition(String partitionPath, List<FileSlice> fileSlices) {
+    ValidationUtils.checkArgument(getHoodieTable().getIndex() instanceof HoodieSparkConsistentBucketIndex,
+        "Mismatch of index type and the clustering strategy, index: " + getHoodieTable().getIndex().getClass().getSimpleName());
+    HoodieConsistentHashingMetadata metadata = HoodieSparkConsistentBucketIndex.loadMetadata(getHoodieTable(), partitionPath);
+    ValidationUtils.checkArgument(metadata != null, "Metadata is null for partition: " + partitionPath);
+    ConsistentBucketIdentifier identifier = new ConsistentBucketIdentifier(metadata);
+
+    // Apply split rule
+    int splitSlot = getWriteConfig().getBucketIndexMaxNumBuckets() - identifier.getNumBuckets();
+    Triple<List<HoodieClusteringGroup>, Integer, List<FileSlice>> splitResult =
+        buildSplitClusteringGroups(identifier, fileSlices, splitSlot);
+    List<HoodieClusteringGroup> ret = new ArrayList<>(splitResult.getLeft());
+
+    // Apply merge rule
+    int mergeSlot = identifier.getNumBuckets() - getWriteConfig().getBucketIndexMinNumBuckets() + splitResult.getMiddle();
+    Triple<List<HoodieClusteringGroup>, Integer, List<FileSlice>> mergeResult =
+        buildMergeClusteringGroup(identifier, splitResult.getRight(), mergeSlot);
+    ret.addAll(mergeResult.getLeft());
+
+    // Apply sort only to the remaining file groups
+    ret.addAll(mergeResult.getRight().stream().map(fs -> {
+      ConsistentHashingNode oldNode = identifier.getBucketByFileId(fs.getFileId());
+      ConsistentHashingNode newNode = new ConsistentHashingNode(oldNode.getValue(), FSUtils.createNewFileIdPfx(), ConsistentHashingNode.NodeTag.REPLACE);
+      return HoodieClusteringGroup.newBuilder()
+          .setSlices(getFileSliceInfo(Collections.singletonList(fs)))
+          .setNumOutputFileGroups(1)
+          .setMetrics(buildMetrics(Collections.singletonList(fs)))
+          .setExtraMetadata(constructExtraMetadata(fs.getPartitionPath(), Collections.singletonList(newNode), identifier.getMetadata().getSeqNo()))
+          .build();
+    }).collect(Collectors.toList()));
+
+    return ret.stream();
+  }
+
+  /**
+   * Generate clustering groups according to split rules。

Review Comment:
   nit: looks like some special character instead of `.` after `... split rules`



##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/clustering/plan/strategy/SparkConsistentBucketClusteringPlanStrategy.java:
##########
@@ -0,0 +1,332 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.client.clustering.plan.strategy;
+
+import org.apache.hudi.avro.model.HoodieClusteringGroup;
+import org.apache.hudi.client.WriteStatus;
+import org.apache.hudi.common.engine.HoodieEngineContext;
+import org.apache.hudi.common.fs.FSUtils;
+import org.apache.hudi.common.model.ConsistentHashingNode;
+import org.apache.hudi.common.model.FileSlice;
+import org.apache.hudi.common.model.HoodieConsistentHashingMetadata;
+import org.apache.hudi.common.model.HoodieFileFormat;
+import org.apache.hudi.common.model.HoodieKey;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.model.HoodieRecordPayload;
+import org.apache.hudi.common.table.timeline.HoodieTimeline;
+import org.apache.hudi.common.table.view.SyncableFileSystemView;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.StringUtils;
+import org.apache.hudi.common.util.ValidationUtils;
+import org.apache.hudi.common.util.collection.Triple;
+import org.apache.hudi.config.HoodieClusteringConfig;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.exception.HoodieClusteringException;
+import org.apache.hudi.index.bucket.ConsistentBucketIdentifier;
+import org.apache.hudi.index.bucket.HoodieSparkConsistentBucketIndex;
+import org.apache.hudi.table.HoodieTable;
+import org.apache.hudi.table.action.cluster.strategy.PartitionAwareClusteringPlanStrategy;
+
+import org.apache.log4j.LogManager;
+import org.apache.log4j.Logger;
+import org.apache.spark.api.java.JavaRDD;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import java.util.stream.Stream;
+
+/**
+ * Clustering plan strategy specifically for consistent bucket index
+ */
+public class SparkConsistentBucketClusteringPlanStrategy<T extends HoodieRecordPayload<T>>
+    extends PartitionAwareClusteringPlanStrategy<T, JavaRDD<HoodieRecord<T>>, JavaRDD<HoodieKey>, JavaRDD<WriteStatus>> {
+
+  private static final Logger LOG = LogManager.getLogger(SparkConsistentBucketClusteringPlanStrategy.class);
+
+  public static final String METADATA_PARTITION_KEY = "clustering.group.partition";
+  public static final String METADATA_CHILD_NODE_KEY = "clustering.group.child.node";
+  public static final String METADATA_SEQUENCE_NUMBER_KEY = "clustering.group.sequence.no";
+
+  public SparkConsistentBucketClusteringPlanStrategy(HoodieTable table, HoodieEngineContext engineContext, HoodieWriteConfig writeConfig) {
+    super(table, engineContext, writeConfig);
+    validate();
+  }
+
+  /**
+   * TODO maybe add force config to schedule the clustering. It could allow clustering on partitions that are not doing write operation.
+   * Block clustering if there is any ongoing concurrent writers
+   *
+   * @return true if the schedule can proceed
+   */
+  @Override
+  public boolean checkPrecondition() {
+    HoodieTimeline timeline = getHoodieTable().getActiveTimeline().getDeltaCommitTimeline().filterInflightsAndRequested();
+    if (!timeline.empty()) {
+      LOG.warn("When using consistent bucket, clustering cannot be scheduled async if there are concurrent writers. "
+          + "Writer instant: " + timeline.getInstants().collect(Collectors.toList()));
+      return false;
+    }
+    return true;
+  }
+
+  /**
+   * Generate candidate clustering file slices of the given partition.
+   * If there is inflight / requested clustering working on the partition, then return empty list
+   * to ensure serialized update to the hashing metadata.
+   *
+   * @return candidate file slices to be clustered (i.e., sort, bucket split or merge)
+   */
+  @Override
+  protected Stream<FileSlice> getFileSlicesEligibleForClustering(String partition) {
+    SyncableFileSystemView fileSystemView = (SyncableFileSystemView) getHoodieTable().getSliceView();
+    boolean isPartitionInClustering = fileSystemView.getFileGroupsInPendingClustering().anyMatch(p -> p.getLeft().getPartitionPath().equals(partition));
+    if (isPartitionInClustering) {
+      LOG.info("Partition: " + partition + " is already in clustering, skip");
+      return Stream.empty();
+    }
+
+    return super.getFileSlicesEligibleForClustering(partition);
+  }
+
+  @Override
+  protected Map<String, String> getStrategyParams() {
+    Map<String, String> params = new HashMap<>();
+    if (!StringUtils.isNullOrEmpty(getWriteConfig().getClusteringSortColumns())) {
+      params.put(HoodieClusteringConfig.PLAN_STRATEGY_SORT_COLUMNS.key(), getWriteConfig().getClusteringSortColumns());
+    }
+    return params;
+  }
+
+  /**
+   * Generate cluster group based on split, merge and sort rules
+   */
+  @Override
+  protected Stream<HoodieClusteringGroup> buildClusteringGroupsForPartition(String partitionPath, List<FileSlice> fileSlices) {
+    ValidationUtils.checkArgument(getHoodieTable().getIndex() instanceof HoodieSparkConsistentBucketIndex,
+        "Mismatch of index type and the clustering strategy, index: " + getHoodieTable().getIndex().getClass().getSimpleName());
+    HoodieConsistentHashingMetadata metadata = HoodieSparkConsistentBucketIndex.loadMetadata(getHoodieTable(), partitionPath);
+    ValidationUtils.checkArgument(metadata != null, "Metadata is null for partition: " + partitionPath);
+    ConsistentBucketIdentifier identifier = new ConsistentBucketIdentifier(metadata);
+
+    // Apply split rule
+    int splitSlot = getWriteConfig().getBucketIndexMaxNumBuckets() - identifier.getNumBuckets();
+    Triple<List<HoodieClusteringGroup>, Integer, List<FileSlice>> splitResult =
+        buildSplitClusteringGroups(identifier, fileSlices, splitSlot);
+    List<HoodieClusteringGroup> ret = new ArrayList<>(splitResult.getLeft());
+
+    // Apply merge rule
+    int mergeSlot = identifier.getNumBuckets() - getWriteConfig().getBucketIndexMinNumBuckets() + splitResult.getMiddle();
+    Triple<List<HoodieClusteringGroup>, Integer, List<FileSlice>> mergeResult =
+        buildMergeClusteringGroup(identifier, splitResult.getRight(), mergeSlot);
+    ret.addAll(mergeResult.getLeft());
+
+    // Apply sort only to the remaining file groups
+    ret.addAll(mergeResult.getRight().stream().map(fs -> {
+      ConsistentHashingNode oldNode = identifier.getBucketByFileId(fs.getFileId());
+      ConsistentHashingNode newNode = new ConsistentHashingNode(oldNode.getValue(), FSUtils.createNewFileIdPfx(), ConsistentHashingNode.NodeTag.REPLACE);
+      return HoodieClusteringGroup.newBuilder()
+          .setSlices(getFileSliceInfo(Collections.singletonList(fs)))
+          .setNumOutputFileGroups(1)
+          .setMetrics(buildMetrics(Collections.singletonList(fs)))
+          .setExtraMetadata(constructExtraMetadata(fs.getPartitionPath(), Collections.singletonList(newNode), identifier.getMetadata().getSeqNo()))
+          .build();
+    }).collect(Collectors.toList()));
+
+    return ret.stream();
+  }
+
+  /**
+   * Generate clustering groups according to split rules。
+   * Currently, we always split bucket into two sub-buckets.
+   *
+   * @param identifier bucket identifier
+   * @param fileSlices file slice candidate to be built as split clustering groups
+   * @param splitSlot  number of new bucket allowed to produce, in order to constrain the upper bound of the total number of bucket
+   * @return list of clustering group, number of new buckets generated, remaining file slice (that does not split)
+   */
+  protected Triple<List<HoodieClusteringGroup>, Integer, List<FileSlice>> buildSplitClusteringGroups(
+      ConsistentBucketIdentifier identifier, List<FileSlice> fileSlices, int splitSlot) {
+    List<HoodieClusteringGroup> retGroup = new ArrayList<>();
+    List<FileSlice> fsUntouched = new ArrayList<>();
+    long splitSize = getSplitSize();
+    int remainingSplitSlot = splitSlot;
+    for (FileSlice fs : fileSlices) {
+      boolean needSplit = fs.getTotalFileSize() > splitSize;
+      if (!needSplit || remainingSplitSlot == 0) {
+        fsUntouched.add(fs);
+        continue;
+      }
+
+      Option<List<ConsistentHashingNode>> nodes = identifier.splitBucket(fs.getFileId());
+
+      // Bucket cannot be split
+      if (!nodes.isPresent()) {
+        fsUntouched.add(fs);
+        continue;
+      }
+
+      remainingSplitSlot--;
+      List<FileSlice> fsList = Collections.singletonList(fs);
+      retGroup.add(HoodieClusteringGroup.newBuilder()
+          .setSlices(getFileSliceInfo(fsList))
+          .setNumOutputFileGroups(2)
+          .setMetrics(buildMetrics(fsList))
+          .setExtraMetadata(constructExtraMetadata(fs.getPartitionPath(), nodes.get(), identifier.getMetadata().getSeqNo()))
+          .build());
+    }
+    return Triple.of(retGroup, splitSlot - remainingSplitSlot, fsUntouched);
+  }
+
+  /**
+   * Generate clustering group according to merge rules
+   *
+   * @param identifier bucket identifier
+   * @param fileSlices file slice candidates to be built as merge clustering groups
+   * @param mergeSlot  number of bucket allowed to be merged, in order to guarantee the lower bound of the total number of bucket
+   * @return list of clustering group, number of buckets merged (removed), remaining file slice (that does not be merged)
+   */
+  protected Triple<List<HoodieClusteringGroup>, Integer, List<FileSlice>> buildMergeClusteringGroup(
+      ConsistentBucketIdentifier identifier, List<FileSlice> fileSlices, int mergeSlot) {
+    if (fileSlices.size() <= 1) {
+      return Triple.of(Collections.emptyList(), 0, fileSlices);
+    }
+
+    long mergeSize = getMergeSize();
+    int remainingMergeSlot = mergeSlot;
+    List<HoodieClusteringGroup> groups = new ArrayList<>();
+    boolean[] added = new boolean[fileSlices.size()];
+
+    fileSlices.sort(Comparator.comparingInt(a -> identifier.getBucketByFileId(a.getFileId()).getValue()));
+    // In each round, we check if the ith file slice can be merged with its predecessors and successors
+    for (int i = 0; i < fileSlices.size(); ++i) {
+      if (added[i] || fileSlices.get(i).getTotalFileSize() > mergeSize) {
+        continue;
+      }
+
+      int startIdx = i;
+      int endIdx = i;
+      // Backward check
+      long totalSize = fileSlices.get(i).getTotalFileSize();
+      do {
+        int preIdx = startIdx >= 1 ? startIdx - 1 : fileSlices.size() - 1;
+        boolean isNeighbour = identifier.getBucketByFileId(fileSlices.get(preIdx).getFileId()) == identifier.getFormerBucket(fileSlices.get(startIdx).getFileId());
+        /**
+         * Merge condition:
+         * 1. there is still slot to merge bucket
+         * 2. the previous file slices is not merged
+         * 3. the previous file slice and current file slice are neighbour in the hash ring
+         * 4. Both the total file size up to now and the previous file slice size are smaller than merge size threshold
+         */
+        if (remainingMergeSlot == 0 || added[preIdx] || !isNeighbour || totalSize > mergeSize || fileSlices.get(preIdx).getTotalFileSize() > mergeSize) {
+          break;
+        }
+
+        // Mark preIdx as merge candidate
+        totalSize += fileSlices.get(preIdx).getTotalFileSize();
+        startIdx = preIdx;
+        remainingMergeSlot--;
+      } while (startIdx != i);
+
+      // Forward check
+      do {
+        int nextIdx = endIdx + 1 < fileSlices.size() ? endIdx + 1 : 0;
+        boolean isNeighbour = identifier.getBucketByFileId(fileSlices.get(endIdx).getFileId()) == identifier.getFormerBucket(fileSlices.get(nextIdx).getFileId());
+
+        if (remainingMergeSlot == 0 || added[nextIdx] || !isNeighbour || totalSize > mergeSize || fileSlices.get(nextIdx).getTotalFileSize() > mergeSize) {
+          break;
+        }
+
+        // Mark nextIdx as merge candidate
+        totalSize += fileSlices.get(nextIdx).getTotalFileSize();
+        endIdx = nextIdx;
+        remainingMergeSlot--;

Review Comment:
   This manipulation looks similar to backward check. Can we extract to a method and simplify? 



##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/index/bucket/HoodieSparkConsistentBucketIndex.java:
##########
@@ -67,6 +77,61 @@ public HoodieData<WriteStatus> updateLocation(HoodieData<WriteStatus> writeStatu
                                                 HoodieEngineContext context,
                                                 HoodieTable hoodieTable)
       throws HoodieIndexException {
+    throw new HoodieIndexException("Consistent hashing index does not support update location without the instant parameter");
+  }
+
+  /**
+   * Persist hashing metadata to storage. Only clustering operations will modify the metadata.
+   * For example, splitting & merging buckets, or just sorting and producing a new bucket.
+   */
+  @Override
+  public HoodieData<WriteStatus> updateLocation(HoodieData<WriteStatus> writeStatuses,
+                                                HoodieEngineContext context,
+                                                HoodieTable hoodieTable,
+                                                String instantTime)
+      throws HoodieIndexException {
+    HoodieInstant instant = hoodieTable.getMetaClient().getActiveTimeline().findInstantsAfterOrEquals(instantTime, 1).firstInstant().get();
+    ValidationUtils.checkState(instant.getTimestamp().equals(instantTime), "Cannot get the same instant, instantTime: " + instantTime);
+    if (!instant.getAction().equals(HoodieTimeline.REPLACE_COMMIT_ACTION)) {
+      return writeStatuses;
+    }
+
+    // Double-check if it is a clustering operation by trying to obtain the clustering plan
+    Option<Pair<HoodieInstant, HoodieClusteringPlan>> instantPlanPair =
+        ClusteringUtils.getClusteringPlan(hoodieTable.getMetaClient(), HoodieTimeline.getReplaceCommitRequestedInstant(instantTime));
+    if (!instantPlanPair.isPresent()) {
+      return writeStatuses;
+    }
+
+    HoodieClusteringPlan plan = instantPlanPair.get().getRight();
+    HoodieJavaRDD.getJavaRDD(context.parallelize(plan.getInputGroups().stream().map(HoodieClusteringGroup::getExtraMetadata).collect(Collectors.toList())))
+        .mapToPair(m -> new Tuple2<>(m.get(SparkConsistentBucketClusteringPlanStrategy.METADATA_PARTITION_KEY), m)
+    ).groupByKey().foreach((input) -> {
+      // Process each partition
+      String partition = input._1();
+      List<ConsistentHashingNode> childNodes = new ArrayList<>();
+      int seqNo = 0;

Review Comment:
   Does this method need to be synchronized to handle multi-writers? Or maybe extract `seqNo` as a class member variable that is `AtomicInteger`?



##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bucket/ConsistentBucketIdentifier.java:
##########
@@ -88,17 +96,106 @@ protected ConsistentHashingNode getBucket(int hashValue) {
     return tailMap.isEmpty() ? ring.firstEntry().getValue() : tailMap.get(tailMap.firstKey());
   }
 
+  /**
+   * Get the former node of the given node (inferred from file id).
+   */
+  public ConsistentHashingNode getFormerBucket(String fileId) {
+    return getFormerBucket(getBucketByFileId(fileId).getValue());
+  }
+
+  /**
+   * Get the former node of the given node (inferred from hash value).
+   */
+  public ConsistentHashingNode getFormerBucket(int hashValue) {
+    SortedMap<Integer, ConsistentHashingNode> headMap = ring.headMap(hashValue);
+    return headMap.isEmpty() ? ring.lastEntry().getValue() : headMap.get(headMap.lastKey());
+  }
+
+  public List<ConsistentHashingNode> mergeBucket(List<String> fileIds) {
+    // Get nodes using fileIds
+    List<ConsistentHashingNode> nodes = fileIds.stream().map(this::getBucketByFileId).collect(Collectors.toList());

Review Comment:
   Should we add some validation like there should be at least two nodes while merging, and that merging  does not result in total number of buckets being less than min buckets?



##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bucket/ConsistentBucketIdentifier.java:
##########
@@ -88,17 +96,106 @@ protected ConsistentHashingNode getBucket(int hashValue) {
     return tailMap.isEmpty() ? ring.firstEntry().getValue() : tailMap.get(tailMap.firstKey());
   }
 
+  /**
+   * Get the former node of the given node (inferred from file id).
+   */
+  public ConsistentHashingNode getFormerBucket(String fileId) {
+    return getFormerBucket(getBucketByFileId(fileId).getValue());
+  }
+
+  /**
+   * Get the former node of the given node (inferred from hash value).
+   */
+  public ConsistentHashingNode getFormerBucket(int hashValue) {
+    SortedMap<Integer, ConsistentHashingNode> headMap = ring.headMap(hashValue);
+    return headMap.isEmpty() ? ring.lastEntry().getValue() : headMap.get(headMap.lastKey());
+  }
+
+  public List<ConsistentHashingNode> mergeBucket(List<String> fileIds) {
+    // Get nodes using fileIds
+    List<ConsistentHashingNode> nodes = fileIds.stream().map(this::getBucketByFileId).collect(Collectors.toList());
+
+    // Validate the input
+    for (int i = 0; i < nodes.size() - 1; ++i) {
+      ValidationUtils.checkState(getFormerBucket(nodes.get(i + 1).getValue()).getValue() == nodes.get(i).getValue(), "Cannot merge discontinuous hash range");
+    }
+
+    // Create child nodes with proper tag (keep the last one and delete other nodes)
+    List<ConsistentHashingNode> childNodes = new ArrayList<>(nodes.size());
+    for (int i = 0; i < nodes.size() - 1; ++i) {
+      childNodes.add(new ConsistentHashingNode(nodes.get(i).getValue(), null, ConsistentHashingNode.NodeTag.DELETE));
+    }
+    childNodes.add(new ConsistentHashingNode(nodes.get(nodes.size() - 1).getValue(), FSUtils.createNewFileIdPfx(), ConsistentHashingNode.NodeTag.REPLACE));
+    return childNodes;
+  }
+
+  public Option<List<ConsistentHashingNode>> splitBucket(String fileId) {
+    ConsistentHashingNode bucket = getBucketByFileId(fileId);
+    ValidationUtils.checkState(bucket != null, "FileId has no corresponding bucket");
+    return splitBucket(bucket);
+  }
+
+  /**
+   * Split bucket in the range middle, also generate the corresponding file ids
+   *
+   * TODO support different split criteria, e.g., distribute records evenly using statistics

Review Comment:
   Let's add JIRA ticket for this TODO



##########
hudi-client/hudi-client-common/src/test/java/org/apache/hudi/index/bucket/TestConsistentBucketIdIdentifier.java:
##########
@@ -76,4 +96,90 @@ public void testGetBucket() {
     Assertions.assertEquals(nodes.get(1), identifier.getBucketByFileId(FSUtils.createNewFileId("1", 0)));
     Assertions.assertEquals(nodes.get(2), identifier.getBucketByFileId(FSUtils.createNewFileId("2", 0)));
   }
+
+  /**
+   * @param v0  first node hash value
+   * @param v1  second node hash value
+   * @param mid mid node hash value generated by split the first bucket v0
+   */
+  @ParameterizedTest
+  @MethodSource("splitBucketParams")
+  public void testSplitBucket(int v0, int v1, int mid) {
+    // Hash range mapping:: [0, 0xf], (0xf, MAX]
+    List<ConsistentHashingNode> nodes = Arrays.asList(
+        new ConsistentHashingNode(v0, "0"),
+        new ConsistentHashingNode(v1, "1"));
+    HoodieConsistentHashingMetadata meta = new HoodieConsistentHashingMetadata((short) 0, "", "", 4, 0, nodes);
+    Option<List<ConsistentHashingNode>> res = new ConsistentBucketIdentifier(meta).splitBucket(nodes.get(0));
+    if (mid < 0) {
+      Assertions.assertTrue(!res.isPresent());
+      return;
+    }
+
+    List<ConsistentHashingNode> childNodes = res.get();
+    Assertions.assertEquals(2, childNodes.size());

Review Comment:
   static import `assertEquals` and other `Assertions#` methods.



##########
hudi-client/hudi-client-common/src/test/java/org/apache/hudi/index/bucket/TestConsistentBucketIdIdentifier.java:
##########
@@ -76,4 +96,90 @@ public void testGetBucket() {
     Assertions.assertEquals(nodes.get(1), identifier.getBucketByFileId(FSUtils.createNewFileId("1", 0)));
     Assertions.assertEquals(nodes.get(2), identifier.getBucketByFileId(FSUtils.createNewFileId("2", 0)));
   }
+
+  /**
+   * @param v0  first node hash value
+   * @param v1  second node hash value
+   * @param mid mid node hash value generated by split the first bucket v0
+   */
+  @ParameterizedTest
+  @MethodSource("splitBucketParams")
+  public void testSplitBucket(int v0, int v1, int mid) {
+    // Hash range mapping:: [0, 0xf], (0xf, MAX]
+    List<ConsistentHashingNode> nodes = Arrays.asList(
+        new ConsistentHashingNode(v0, "0"),
+        new ConsistentHashingNode(v1, "1"));
+    HoodieConsistentHashingMetadata meta = new HoodieConsistentHashingMetadata((short) 0, "", "", 4, 0, nodes);
+    Option<List<ConsistentHashingNode>> res = new ConsistentBucketIdentifier(meta).splitBucket(nodes.get(0));
+    if (mid < 0) {
+      Assertions.assertTrue(!res.isPresent());
+      return;
+    }
+
+    List<ConsistentHashingNode> childNodes = res.get();
+    Assertions.assertEquals(2, childNodes.size());
+    Assertions.assertTrue(childNodes.stream().allMatch(c -> c.getTag() == ConsistentHashingNode.NodeTag.REPLACE));
+    Assertions.assertEquals(mid, childNodes.get(0).getValue());
+    Assertions.assertEquals(nodes.get(0).getValue(), childNodes.get(1).getValue());
+  }
+
+  @Test
+  public void testMerge() {
+    HoodieConsistentHashingMetadata meta = new HoodieConsistentHashingMetadata("partition", 8);
+    List<ConsistentHashingNode> nodes = meta.getNodes();
+
+    List<String> fileIds = IntStream.range(0, 3).mapToObj(i -> FSUtils.createNewFileId(nodes.get(i).getFileIdPrefix(), 0)).collect(Collectors.toList());
+    List<ConsistentHashingNode> childNodes = new ConsistentBucketIdentifier(meta).mergeBucket(fileIds);
+    Assertions.assertEquals(ConsistentHashingNode.NodeTag.DELETE, childNodes.get(0).getTag());
+    Assertions.assertEquals(ConsistentHashingNode.NodeTag.DELETE, childNodes.get(1).getTag());
+    Assertions.assertEquals(ConsistentHashingNode.NodeTag.REPLACE, childNodes.get(2).getTag());
+    Assertions.assertEquals(nodes.get(2).getValue(), childNodes.get(2).getValue());
+    Assertions.assertNotEquals(nodes.get(2).getFileIdPrefix(), childNodes.get(2).getFileIdPrefix());
+
+    fileIds = Arrays.asList(nodes.get(7), nodes.get(0), nodes.get(1)).stream()
+        .map(ConsistentHashingNode::getFileIdPrefix).map(f -> FSUtils.createNewFileId(f, 0)).collect(Collectors.toList());
+    childNodes = new ConsistentBucketIdentifier(meta).mergeBucket(fileIds);
+    Assertions.assertEquals(ConsistentHashingNode.NodeTag.DELETE, childNodes.get(0).getTag());
+    Assertions.assertEquals(ConsistentHashingNode.NodeTag.DELETE, childNodes.get(1).getTag());
+    Assertions.assertEquals(ConsistentHashingNode.NodeTag.REPLACE, childNodes.get(2).getTag());
+    Assertions.assertEquals(nodes.get(1).getValue(), childNodes.get(2).getValue());
+    Assertions.assertNotEquals(nodes.get(1).getFileIdPrefix(), childNodes.get(2).getFileIdPrefix());
+
+    boolean exception = false;
+    try {
+      fileIds = IntStream.range(0, 2).mapToObj(i -> FSUtils.createNewFileId(nodes.get(i * 2).getFileIdPrefix(), 0)).collect(Collectors.toList());
+      new ConsistentBucketIdentifier(meta).mergeBucket(fileIds);
+    } catch (Exception e) {
+      exception = true;
+    }
+    Assertions.assertTrue(exception);

Review Comment:
   how about extracting this to a separate test which asserts exception thrown and any err message? Better not to mix happy test cases with failure scenarios.



##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/clustering/run/strategy/SparkConsistentBucketClusteringExecutionStrategy.java:
##########
@@ -0,0 +1,79 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.client.clustering.run.strategy;
+
+import org.apache.hudi.client.WriteStatus;
+import org.apache.hudi.client.clustering.plan.strategy.SparkConsistentBucketClusteringPlanStrategy;
+import org.apache.hudi.common.data.HoodieData;
+import org.apache.hudi.common.engine.HoodieEngineContext;
+import org.apache.hudi.common.model.ConsistentHashingNode;
+import org.apache.hudi.common.model.HoodieFileGroupId;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.model.HoodieRecordPayload;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.exception.HoodieClusteringException;
+import org.apache.hudi.execution.bulkinsert.RDDConsistentBucketPartitioner;
+import org.apache.hudi.table.HoodieTable;
+import org.apache.hudi.table.action.commit.SparkBulkInsertHelper;
+
+import org.apache.avro.Schema;
+import org.apache.log4j.LogManager;
+import org.apache.log4j.Logger;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+
+/**
+ * Clustering execution strategy specifically for consistent hashing index
+ */
+public class SparkConsistentBucketClusteringExecutionStrategy<T extends HoodieRecordPayload<T>>
+    extends MultipleSparkJobExecutionStrategy<T> {
+
+  private static final Logger LOG = LogManager.getLogger(SparkSortAndSizeExecutionStrategy.class);

Review Comment:
   ```suggestion
     private static final Logger LOG = LogManager.getLogger(SparkConsistentBucketClusteringExecutionStrategy.class);
   ```



##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bucket/BucketIndexLocationMapper.java:
##########
@@ -28,8 +28,8 @@
 public interface BucketIndexLocationMapper extends Serializable {
 
   /**
-   * Get record location given hoodie key and partition path
+   * Get record location given hoodie key
    */
-  Option<HoodieRecordLocation> getRecordLocation(HoodieKey key, String partitionPath);
+  Option<HoodieRecordLocation> getRecordLocation(HoodieKey key);

Review Comment:
   +1
   `HoodiKey` already has `partitionPath`



##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/index/bucket/HoodieSparkConsistentBucketIndex.java:
##########
@@ -67,6 +77,61 @@ public HoodieData<WriteStatus> updateLocation(HoodieData<WriteStatus> writeStatu
                                                 HoodieEngineContext context,
                                                 HoodieTable hoodieTable)
       throws HoodieIndexException {
+    throw new HoodieIndexException("Consistent hashing index does not support update location without the instant parameter");
+  }
+
+  /**
+   * Persist hashing metadata to storage. Only clustering operations will modify the metadata.
+   * For example, splitting & merging buckets, or just sorting and producing a new bucket.
+   */
+  @Override
+  public HoodieData<WriteStatus> updateLocation(HoodieData<WriteStatus> writeStatuses,
+                                                HoodieEngineContext context,
+                                                HoodieTable hoodieTable,
+                                                String instantTime)
+      throws HoodieIndexException {
+    HoodieInstant instant = hoodieTable.getMetaClient().getActiveTimeline().findInstantsAfterOrEquals(instantTime, 1).firstInstant().get();
+    ValidationUtils.checkState(instant.getTimestamp().equals(instantTime), "Cannot get the same instant, instantTime: " + instantTime);
+    if (!instant.getAction().equals(HoodieTimeline.REPLACE_COMMIT_ACTION)) {
+      return writeStatuses;
+    }
+
+    // Double-check if it is a clustering operation by trying to obtain the clustering plan
+    Option<Pair<HoodieInstant, HoodieClusteringPlan>> instantPlanPair =
+        ClusteringUtils.getClusteringPlan(hoodieTable.getMetaClient(), HoodieTimeline.getReplaceCommitRequestedInstant(instantTime));
+    if (!instantPlanPair.isPresent()) {
+      return writeStatuses;
+    }
+
+    HoodieClusteringPlan plan = instantPlanPair.get().getRight();
+    HoodieJavaRDD.getJavaRDD(context.parallelize(plan.getInputGroups().stream().map(HoodieClusteringGroup::getExtraMetadata).collect(Collectors.toList())))
+        .mapToPair(m -> new Tuple2<>(m.get(SparkConsistentBucketClusteringPlanStrategy.METADATA_PARTITION_KEY), m)
+    ).groupByKey().foreach((input) -> {
+      // Process each partition
+      String partition = input._1();
+      List<ConsistentHashingNode> childNodes = new ArrayList<>();
+      int seqNo = 0;
+      for (Map<String, String> m: input._2()) {
+        String nodesJson = m.get(SparkConsistentBucketClusteringPlanStrategy.METADATA_CHILD_NODE_KEY);
+        childNodes.addAll(ConsistentHashingNode.fromJsonString(nodesJson));
+        seqNo = Integer.parseInt(m.get(SparkConsistentBucketClusteringPlanStrategy.METADATA_SEQUENCE_NUMBER_KEY));
+      }
+
+      HoodieConsistentHashingMetadata meta = loadMetadata(hoodieTable, partition);

Review Comment:
   `meta` can be null as well. We should handle that. I would suggest to change  `loadMetadata` in the following way:
   1. Make it return `Option<HoodieConsistentHashingMetadata>`
   2. Do not do `table.getMetaClient().getFs().exists(metadataPath)` check in `loadMetadata` as it s a costly operation. Let the call fail with IOException and we either handle that or pass it to the caller. 



##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/clustering/update/strategy/SparkConsistentHashingDuplicateUpdateStrategy.java:
##########
@@ -0,0 +1,146 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.client.clustering.update.strategy;
+
+import org.apache.hudi.avro.model.HoodieClusteringGroup;
+import org.apache.hudi.avro.model.HoodieClusteringPlan;
+import org.apache.hudi.client.clustering.plan.strategy.SparkConsistentBucketClusteringPlanStrategy;
+import org.apache.hudi.common.data.HoodieData;
+import org.apache.hudi.common.engine.HoodieEngineContext;
+import org.apache.hudi.common.fs.FSUtils;
+import org.apache.hudi.common.model.ConsistentHashingNode;
+import org.apache.hudi.common.model.HoodieAvroRecord;
+import org.apache.hudi.common.model.HoodieConsistentHashingMetadata;
+import org.apache.hudi.common.model.HoodieFileGroupId;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.model.HoodieRecordLocation;
+import org.apache.hudi.common.model.HoodieRecordPayload;
+import org.apache.hudi.common.table.timeline.HoodieInstant;
+import org.apache.hudi.common.table.timeline.HoodieTimeline;
+import org.apache.hudi.common.util.ClusteringUtils;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.ValidationUtils;
+import org.apache.hudi.common.util.collection.Pair;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.index.HoodieIndexUtils;
+import org.apache.hudi.index.bucket.ConsistentBucketIdentifier;
+import org.apache.hudi.index.bucket.HoodieSparkConsistentBucketIndex;
+import org.apache.hudi.table.HoodieTable;
+import org.apache.hudi.table.action.cluster.strategy.UpdateStrategy;
+
+import org.apache.log4j.LogManager;
+import org.apache.log4j.Logger;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+/**
+ * Update strategy for (consistent hashing) bucket index
+ * If updates to file groups that are under clustering are identified, then generate
+ * two same records for each update, routing to both old and new file groups
+ */
+public class SparkConsistentHashingDuplicateUpdateStrategy<T extends HoodieRecordPayload<T>> extends UpdateStrategy<T, HoodieData<HoodieRecord<T>>> {
+
+  private static final Logger LOG = LogManager.getLogger(SparkConsistentHashingDuplicateUpdateStrategy.class);
+
+  public SparkConsistentHashingDuplicateUpdateStrategy(HoodieEngineContext engineContext, HoodieTable table, Set<HoodieFileGroupId> fileGroupsInPendingClustering) {
+    super(engineContext, table, fileGroupsInPendingClustering);
+  }
+
+  @Override
+  public Pair<HoodieData<HoodieRecord<T>>, Set<HoodieFileGroupId>> handleUpdate(HoodieData<HoodieRecord<T>> taggedRecordsRDD) {
+    if (fileGroupsInPendingClustering.isEmpty()) {
+      return Pair.of(taggedRecordsRDD, Collections.emptySet());
+    }
+
+    HoodieData<HoodieRecord<T>> filteredRecordsRDD = taggedRecordsRDD.filter(r -> {
+      ValidationUtils.checkState(r.getCurrentLocation() != null);
+      return fileGroupsInPendingClustering.contains(new HoodieFileGroupId(r.getPartitionPath(), r.getCurrentLocation().getFileId()));
+    });
+
+    if (filteredRecordsRDD.count() == 0) {
+      return Pair.of(taggedRecordsRDD, Collections.emptySet());
+    }
+
+    // Read all pending/ongoing clustering plans
+    List<Pair<HoodieInstant, HoodieClusteringPlan>> instantPlanPairs =
+        table.getMetaClient().getActiveTimeline().filterInflightsAndRequested().filter(instant -> instant.getAction().equals(HoodieTimeline.REPLACE_COMMIT_ACTION)).getInstants()
+            .map(instant -> ClusteringUtils.getClusteringPlan(table.getMetaClient(), instant))
+            .flatMap(o -> o.isPresent() ? Stream.of(o.get()) : Stream.empty())
+            .collect(Collectors.toList());
+
+    // Construct child node for each partition & build the bucket identifier
+    final Set<String> partitions = new HashSet<>(filteredRecordsRDD.map(HoodieRecord::getPartitionPath).distinct().collectAsList());
+    Map<String, HoodieConsistentHashingMetadata> partitionToHashingMeta = new HashMap<>();
+    Map<String, String> partitionToInstant = new HashMap<>();
+    for (Pair<HoodieInstant, HoodieClusteringPlan> pair : instantPlanPairs) {
+      String instant = pair.getLeft().getTimestamp();
+      HoodieClusteringPlan plan = pair.getRight();
+      extractHashingMetadataFromClusteringPlan(instant, plan, partitions, partitionToHashingMeta, partitionToInstant);
+    }
+    Map<String, ConsistentBucketIdentifier> partitionToIdentifier = partitionToHashingMeta.entrySet().stream()
+        .collect(Collectors.toMap(Map.Entry::getKey, e -> new ConsistentBucketIdentifier(e.getValue())));
+
+    // Produce records tagged with new record location
+    List<String> indexKeyFields = Arrays.asList(table.getConfig().getBucketIndexHashField().split(","));
+    HoodieData<HoodieRecord<T>> redirectedRecordsRDD = filteredRecordsRDD.map(r -> {
+      ConsistentHashingNode node = partitionToIdentifier.get(r.getPartitionPath()).getBucket(r.getKey(), indexKeyFields);
+      return HoodieIndexUtils.getTaggedRecord(new HoodieAvroRecord(r.getKey(), r.getData(), r.getOperation()),

Review Comment:
   static import `getTaggedRecord`



##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/clustering/update/strategy/SparkConsistentHashingDuplicateUpdateStrategy.java:
##########
@@ -0,0 +1,146 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.client.clustering.update.strategy;
+
+import org.apache.hudi.avro.model.HoodieClusteringGroup;
+import org.apache.hudi.avro.model.HoodieClusteringPlan;
+import org.apache.hudi.client.clustering.plan.strategy.SparkConsistentBucketClusteringPlanStrategy;
+import org.apache.hudi.common.data.HoodieData;
+import org.apache.hudi.common.engine.HoodieEngineContext;
+import org.apache.hudi.common.fs.FSUtils;
+import org.apache.hudi.common.model.ConsistentHashingNode;
+import org.apache.hudi.common.model.HoodieAvroRecord;
+import org.apache.hudi.common.model.HoodieConsistentHashingMetadata;
+import org.apache.hudi.common.model.HoodieFileGroupId;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.model.HoodieRecordLocation;
+import org.apache.hudi.common.model.HoodieRecordPayload;
+import org.apache.hudi.common.table.timeline.HoodieInstant;
+import org.apache.hudi.common.table.timeline.HoodieTimeline;
+import org.apache.hudi.common.util.ClusteringUtils;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.ValidationUtils;
+import org.apache.hudi.common.util.collection.Pair;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.index.HoodieIndexUtils;
+import org.apache.hudi.index.bucket.ConsistentBucketIdentifier;
+import org.apache.hudi.index.bucket.HoodieSparkConsistentBucketIndex;
+import org.apache.hudi.table.HoodieTable;
+import org.apache.hudi.table.action.cluster.strategy.UpdateStrategy;
+
+import org.apache.log4j.LogManager;
+import org.apache.log4j.Logger;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+/**
+ * Update strategy for (consistent hashing) bucket index
+ * If updates to file groups that are under clustering are identified, then generate
+ * two same records for each update, routing to both old and new file groups
+ */
+public class SparkConsistentHashingDuplicateUpdateStrategy<T extends HoodieRecordPayload<T>> extends UpdateStrategy<T, HoodieData<HoodieRecord<T>>> {
+
+  private static final Logger LOG = LogManager.getLogger(SparkConsistentHashingDuplicateUpdateStrategy.class);
+
+  public SparkConsistentHashingDuplicateUpdateStrategy(HoodieEngineContext engineContext, HoodieTable table, Set<HoodieFileGroupId> fileGroupsInPendingClustering) {
+    super(engineContext, table, fileGroupsInPendingClustering);
+  }
+
+  @Override
+  public Pair<HoodieData<HoodieRecord<T>>, Set<HoodieFileGroupId>> handleUpdate(HoodieData<HoodieRecord<T>> taggedRecordsRDD) {
+    if (fileGroupsInPendingClustering.isEmpty()) {
+      return Pair.of(taggedRecordsRDD, Collections.emptySet());
+    }
+
+    HoodieData<HoodieRecord<T>> filteredRecordsRDD = taggedRecordsRDD.filter(r -> {
+      ValidationUtils.checkState(r.getCurrentLocation() != null);
+      return fileGroupsInPendingClustering.contains(new HoodieFileGroupId(r.getPartitionPath(), r.getCurrentLocation().getFileId()));
+    });
+
+    if (filteredRecordsRDD.count() == 0) {
+      return Pair.of(taggedRecordsRDD, Collections.emptySet());
+    }
+
+    // Read all pending/ongoing clustering plans
+    List<Pair<HoodieInstant, HoodieClusteringPlan>> instantPlanPairs =
+        table.getMetaClient().getActiveTimeline().filterInflightsAndRequested().filter(instant -> instant.getAction().equals(HoodieTimeline.REPLACE_COMMIT_ACTION)).getInstants()
+            .map(instant -> ClusteringUtils.getClusteringPlan(table.getMetaClient(), instant))
+            .flatMap(o -> o.isPresent() ? Stream.of(o.get()) : Stream.empty())
+            .collect(Collectors.toList());
+
+    // Construct child node for each partition & build the bucket identifier
+    final Set<String> partitions = new HashSet<>(filteredRecordsRDD.map(HoodieRecord::getPartitionPath).distinct().collectAsList());
+    Map<String, HoodieConsistentHashingMetadata> partitionToHashingMeta = new HashMap<>();
+    Map<String, String> partitionToInstant = new HashMap<>();
+    for (Pair<HoodieInstant, HoodieClusteringPlan> pair : instantPlanPairs) {
+      String instant = pair.getLeft().getTimestamp();
+      HoodieClusteringPlan plan = pair.getRight();
+      extractHashingMetadataFromClusteringPlan(instant, plan, partitions, partitionToHashingMeta, partitionToInstant);
+    }
+    Map<String, ConsistentBucketIdentifier> partitionToIdentifier = partitionToHashingMeta.entrySet().stream()
+        .collect(Collectors.toMap(Map.Entry::getKey, e -> new ConsistentBucketIdentifier(e.getValue())));
+
+    // Produce records tagged with new record location
+    List<String> indexKeyFields = Arrays.asList(table.getConfig().getBucketIndexHashField().split(","));
+    HoodieData<HoodieRecord<T>> redirectedRecordsRDD = filteredRecordsRDD.map(r -> {
+      ConsistentHashingNode node = partitionToIdentifier.get(r.getPartitionPath()).getBucket(r.getKey(), indexKeyFields);
+      return HoodieIndexUtils.getTaggedRecord(new HoodieAvroRecord(r.getKey(), r.getData(), r.getOperation()),
+          Option.ofNullable(new HoodieRecordLocation(partitionToInstant.get(r.getPartitionPath()), FSUtils.createNewFileId(node.getFileIdPrefix(), 0))));
+    });
+
+    // Return combined iterator (the original and records with new location)
+    return Pair.of(taggedRecordsRDD.union(redirectedRecordsRDD), Collections.emptySet());
+  }
+
+  private void extractHashingMetadataFromClusteringPlan(String instant, HoodieClusteringPlan plan, final Set<String> recordPartitions,
+                                                        Map<String, HoodieConsistentHashingMetadata> partitionToHashingMeta, Map<String, String> partitionToInstant) {
+    for (HoodieClusteringGroup group : plan.getInputGroups()) {
+      Map<String, String> groupMeta = group.getExtraMetadata();
+      String p = groupMeta.get(SparkConsistentBucketClusteringPlanStrategy.METADATA_PARTITION_KEY);

Review Comment:
   Let's guard this whle logic with `groupMeta.containsKey(METADATA_PARTITION_KEY)` check.
   Also, static import the constant.



##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/clustering/update/strategy/SparkConsistentHashingDuplicateUpdateStrategy.java:
##########
@@ -0,0 +1,146 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.client.clustering.update.strategy;
+
+import org.apache.hudi.avro.model.HoodieClusteringGroup;
+import org.apache.hudi.avro.model.HoodieClusteringPlan;
+import org.apache.hudi.client.clustering.plan.strategy.SparkConsistentBucketClusteringPlanStrategy;
+import org.apache.hudi.common.data.HoodieData;
+import org.apache.hudi.common.engine.HoodieEngineContext;
+import org.apache.hudi.common.fs.FSUtils;
+import org.apache.hudi.common.model.ConsistentHashingNode;
+import org.apache.hudi.common.model.HoodieAvroRecord;
+import org.apache.hudi.common.model.HoodieConsistentHashingMetadata;
+import org.apache.hudi.common.model.HoodieFileGroupId;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.model.HoodieRecordLocation;
+import org.apache.hudi.common.model.HoodieRecordPayload;
+import org.apache.hudi.common.table.timeline.HoodieInstant;
+import org.apache.hudi.common.table.timeline.HoodieTimeline;
+import org.apache.hudi.common.util.ClusteringUtils;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.ValidationUtils;
+import org.apache.hudi.common.util.collection.Pair;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.index.HoodieIndexUtils;
+import org.apache.hudi.index.bucket.ConsistentBucketIdentifier;
+import org.apache.hudi.index.bucket.HoodieSparkConsistentBucketIndex;
+import org.apache.hudi.table.HoodieTable;
+import org.apache.hudi.table.action.cluster.strategy.UpdateStrategy;
+
+import org.apache.log4j.LogManager;
+import org.apache.log4j.Logger;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+/**
+ * Update strategy for (consistent hashing) bucket index
+ * If updates to file groups that are under clustering are identified, then generate
+ * two same records for each update, routing to both old and new file groups
+ */
+public class SparkConsistentHashingDuplicateUpdateStrategy<T extends HoodieRecordPayload<T>> extends UpdateStrategy<T, HoodieData<HoodieRecord<T>>> {

Review Comment:
   Consider renaming to `SparkConsistentBucketDuplicateUpdateStrategy` for the sake of consistent naming everywhere. 



##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/clustering/update/strategy/SparkConsistentHashingDuplicateUpdateStrategy.java:
##########
@@ -0,0 +1,146 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.client.clustering.update.strategy;
+
+import org.apache.hudi.avro.model.HoodieClusteringGroup;
+import org.apache.hudi.avro.model.HoodieClusteringPlan;
+import org.apache.hudi.client.clustering.plan.strategy.SparkConsistentBucketClusteringPlanStrategy;
+import org.apache.hudi.common.data.HoodieData;
+import org.apache.hudi.common.engine.HoodieEngineContext;
+import org.apache.hudi.common.fs.FSUtils;
+import org.apache.hudi.common.model.ConsistentHashingNode;
+import org.apache.hudi.common.model.HoodieAvroRecord;
+import org.apache.hudi.common.model.HoodieConsistentHashingMetadata;
+import org.apache.hudi.common.model.HoodieFileGroupId;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.model.HoodieRecordLocation;
+import org.apache.hudi.common.model.HoodieRecordPayload;
+import org.apache.hudi.common.table.timeline.HoodieInstant;
+import org.apache.hudi.common.table.timeline.HoodieTimeline;
+import org.apache.hudi.common.util.ClusteringUtils;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.ValidationUtils;
+import org.apache.hudi.common.util.collection.Pair;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.index.HoodieIndexUtils;
+import org.apache.hudi.index.bucket.ConsistentBucketIdentifier;
+import org.apache.hudi.index.bucket.HoodieSparkConsistentBucketIndex;
+import org.apache.hudi.table.HoodieTable;
+import org.apache.hudi.table.action.cluster.strategy.UpdateStrategy;
+
+import org.apache.log4j.LogManager;
+import org.apache.log4j.Logger;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+/**
+ * Update strategy for (consistent hashing) bucket index
+ * If updates to file groups that are under clustering are identified, then generate
+ * two same records for each update, routing to both old and new file groups
+ */
+public class SparkConsistentHashingDuplicateUpdateStrategy<T extends HoodieRecordPayload<T>> extends UpdateStrategy<T, HoodieData<HoodieRecord<T>>> {
+
+  private static final Logger LOG = LogManager.getLogger(SparkConsistentHashingDuplicateUpdateStrategy.class);
+
+  public SparkConsistentHashingDuplicateUpdateStrategy(HoodieEngineContext engineContext, HoodieTable table, Set<HoodieFileGroupId> fileGroupsInPendingClustering) {
+    super(engineContext, table, fileGroupsInPendingClustering);
+  }
+
+  @Override
+  public Pair<HoodieData<HoodieRecord<T>>, Set<HoodieFileGroupId>> handleUpdate(HoodieData<HoodieRecord<T>> taggedRecordsRDD) {
+    if (fileGroupsInPendingClustering.isEmpty()) {
+      return Pair.of(taggedRecordsRDD, Collections.emptySet());
+    }
+
+    HoodieData<HoodieRecord<T>> filteredRecordsRDD = taggedRecordsRDD.filter(r -> {
+      ValidationUtils.checkState(r.getCurrentLocation() != null);
+      return fileGroupsInPendingClustering.contains(new HoodieFileGroupId(r.getPartitionPath(), r.getCurrentLocation().getFileId()));
+    });
+
+    if (filteredRecordsRDD.count() == 0) {
+      return Pair.of(taggedRecordsRDD, Collections.emptySet());
+    }
+
+    // Read all pending/ongoing clustering plans
+    List<Pair<HoodieInstant, HoodieClusteringPlan>> instantPlanPairs =
+        table.getMetaClient().getActiveTimeline().filterInflightsAndRequested().filter(instant -> instant.getAction().equals(HoodieTimeline.REPLACE_COMMIT_ACTION)).getInstants()
+            .map(instant -> ClusteringUtils.getClusteringPlan(table.getMetaClient(), instant))
+            .flatMap(o -> o.isPresent() ? Stream.of(o.get()) : Stream.empty())
+            .collect(Collectors.toList());
+
+    // Construct child node for each partition & build the bucket identifier
+    final Set<String> partitions = new HashSet<>(filteredRecordsRDD.map(HoodieRecord::getPartitionPath).distinct().collectAsList());
+    Map<String, HoodieConsistentHashingMetadata> partitionToHashingMeta = new HashMap<>();
+    Map<String, String> partitionToInstant = new HashMap<>();
+    for (Pair<HoodieInstant, HoodieClusteringPlan> pair : instantPlanPairs) {
+      String instant = pair.getLeft().getTimestamp();
+      HoodieClusteringPlan plan = pair.getRight();
+      extractHashingMetadataFromClusteringPlan(instant, plan, partitions, partitionToHashingMeta, partitionToInstant);
+    }
+    Map<String, ConsistentBucketIdentifier> partitionToIdentifier = partitionToHashingMeta.entrySet().stream()
+        .collect(Collectors.toMap(Map.Entry::getKey, e -> new ConsistentBucketIdentifier(e.getValue())));
+
+    // Produce records tagged with new record location
+    List<String> indexKeyFields = Arrays.asList(table.getConfig().getBucketIndexHashField().split(","));
+    HoodieData<HoodieRecord<T>> redirectedRecordsRDD = filteredRecordsRDD.map(r -> {
+      ConsistentHashingNode node = partitionToIdentifier.get(r.getPartitionPath()).getBucket(r.getKey(), indexKeyFields);
+      return HoodieIndexUtils.getTaggedRecord(new HoodieAvroRecord(r.getKey(), r.getData(), r.getOperation()),

Review Comment:
   we should handle the case when `partitionToIdentifier` does not contain the key.



##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/clustering/run/strategy/SparkConsistentBucketClusteringExecutionStrategy.java:
##########
@@ -0,0 +1,79 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.client.clustering.run.strategy;
+
+import org.apache.hudi.client.WriteStatus;
+import org.apache.hudi.client.clustering.plan.strategy.SparkConsistentBucketClusteringPlanStrategy;
+import org.apache.hudi.common.data.HoodieData;
+import org.apache.hudi.common.engine.HoodieEngineContext;
+import org.apache.hudi.common.model.ConsistentHashingNode;
+import org.apache.hudi.common.model.HoodieFileGroupId;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.model.HoodieRecordPayload;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.exception.HoodieClusteringException;
+import org.apache.hudi.execution.bulkinsert.RDDConsistentBucketPartitioner;
+import org.apache.hudi.table.HoodieTable;
+import org.apache.hudi.table.action.commit.SparkBulkInsertHelper;
+
+import org.apache.avro.Schema;
+import org.apache.log4j.LogManager;
+import org.apache.log4j.Logger;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+
+/**
+ * Clustering execution strategy specifically for consistent hashing index
+ */
+public class SparkConsistentBucketClusteringExecutionStrategy<T extends HoodieRecordPayload<T>>
+    extends MultipleSparkJobExecutionStrategy<T> {
+
+  private static final Logger LOG = LogManager.getLogger(SparkSortAndSizeExecutionStrategy.class);
+
+  public SparkConsistentBucketClusteringExecutionStrategy(HoodieTable table, HoodieEngineContext engineContext,
+                                                          HoodieWriteConfig writeConfig) {
+    super(table, engineContext, writeConfig);
+  }
+
+  @Override
+  public HoodieData<WriteStatus> performClusteringWithRecordsRDD(HoodieData<HoodieRecord<T>> inputRecords, int numOutputGroups, String instantTime,
+                                                                 Map<String, String> strategyParams, Schema schema, List<HoodieFileGroupId> fileGroupIdList,
+                                                                 boolean preserveHoodieMetadata, Map<String, String> extraMetadata) {
+
+    LOG.info("Starting clustering for a group, parallelism:" + numOutputGroups + " commit:" + instantTime);
+    Properties props = getWriteConfig().getProps();
+    // We are calling another action executor - disable auto commit. Strategy is only expected to write data in new files.
+    props.put(HoodieWriteConfig.AUTO_COMMIT_ENABLE.key(), Boolean.FALSE.toString());
+    HoodieWriteConfig newConfig = HoodieWriteConfig.newBuilder().withProps(props).build();
+
+    RDDConsistentBucketPartitioner<T> partitioner = new RDDConsistentBucketPartitioner<>(getHoodieTable(), getWriteConfig(), strategyParams, preserveHoodieMetadata);
+    try {
+      List<ConsistentHashingNode> nodes = ConsistentHashingNode.fromJsonString(extraMetadata.get(SparkConsistentBucketClusteringPlanStrategy.METADATA_CHILD_NODE_KEY));
+      partitioner.addHashingChildrenNodes(extraMetadata.get(SparkConsistentBucketClusteringPlanStrategy.METADATA_PARTITION_KEY), nodes);
+    } catch (Exception e) {
+      LOG.error("Failed to add hashing children nodes", e);
+      throw new HoodieClusteringException("Failed to add hashing children nodes", e);
+    }
+
+    return (HoodieData<WriteStatus>) SparkBulkInsertHelper.newInstance()
+        .bulkInsert(inputRecords, instantTime, getHoodieTable(), newConfig, false, partitioner, true, numOutputGroups, null);

Review Comment:
   Why is writeHandleFactory (last argument of `bulkInsert()` call) passed as null? If it's not needed, I would rather change the `bulkInsert()` method argument to `Option<WriteHandleFactory>` and handle empty handle in the callee.



##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/clustering/update/strategy/SparkConsistentHashingDuplicateUpdateStrategy.java:
##########
@@ -0,0 +1,146 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.client.clustering.update.strategy;
+
+import org.apache.hudi.avro.model.HoodieClusteringGroup;
+import org.apache.hudi.avro.model.HoodieClusteringPlan;
+import org.apache.hudi.client.clustering.plan.strategy.SparkConsistentBucketClusteringPlanStrategy;
+import org.apache.hudi.common.data.HoodieData;
+import org.apache.hudi.common.engine.HoodieEngineContext;
+import org.apache.hudi.common.fs.FSUtils;
+import org.apache.hudi.common.model.ConsistentHashingNode;
+import org.apache.hudi.common.model.HoodieAvroRecord;
+import org.apache.hudi.common.model.HoodieConsistentHashingMetadata;
+import org.apache.hudi.common.model.HoodieFileGroupId;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.model.HoodieRecordLocation;
+import org.apache.hudi.common.model.HoodieRecordPayload;
+import org.apache.hudi.common.table.timeline.HoodieInstant;
+import org.apache.hudi.common.table.timeline.HoodieTimeline;
+import org.apache.hudi.common.util.ClusteringUtils;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.ValidationUtils;
+import org.apache.hudi.common.util.collection.Pair;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.index.HoodieIndexUtils;
+import org.apache.hudi.index.bucket.ConsistentBucketIdentifier;
+import org.apache.hudi.index.bucket.HoodieSparkConsistentBucketIndex;
+import org.apache.hudi.table.HoodieTable;
+import org.apache.hudi.table.action.cluster.strategy.UpdateStrategy;
+
+import org.apache.log4j.LogManager;
+import org.apache.log4j.Logger;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+/**
+ * Update strategy for (consistent hashing) bucket index
+ * If updates to file groups that are under clustering are identified, then generate
+ * two same records for each update, routing to both old and new file groups
+ */
+public class SparkConsistentHashingDuplicateUpdateStrategy<T extends HoodieRecordPayload<T>> extends UpdateStrategy<T, HoodieData<HoodieRecord<T>>> {
+
+  private static final Logger LOG = LogManager.getLogger(SparkConsistentHashingDuplicateUpdateStrategy.class);
+
+  public SparkConsistentHashingDuplicateUpdateStrategy(HoodieEngineContext engineContext, HoodieTable table, Set<HoodieFileGroupId> fileGroupsInPendingClustering) {
+    super(engineContext, table, fileGroupsInPendingClustering);
+  }
+
+  @Override
+  public Pair<HoodieData<HoodieRecord<T>>, Set<HoodieFileGroupId>> handleUpdate(HoodieData<HoodieRecord<T>> taggedRecordsRDD) {
+    if (fileGroupsInPendingClustering.isEmpty()) {
+      return Pair.of(taggedRecordsRDD, Collections.emptySet());
+    }
+
+    HoodieData<HoodieRecord<T>> filteredRecordsRDD = taggedRecordsRDD.filter(r -> {
+      ValidationUtils.checkState(r.getCurrentLocation() != null);
+      return fileGroupsInPendingClustering.contains(new HoodieFileGroupId(r.getPartitionPath(), r.getCurrentLocation().getFileId()));
+    });
+
+    if (filteredRecordsRDD.count() == 0) {
+      return Pair.of(taggedRecordsRDD, Collections.emptySet());
+    }
+
+    // Read all pending/ongoing clustering plans
+    List<Pair<HoodieInstant, HoodieClusteringPlan>> instantPlanPairs =
+        table.getMetaClient().getActiveTimeline().filterInflightsAndRequested().filter(instant -> instant.getAction().equals(HoodieTimeline.REPLACE_COMMIT_ACTION)).getInstants()
+            .map(instant -> ClusteringUtils.getClusteringPlan(table.getMetaClient(), instant))
+            .flatMap(o -> o.isPresent() ? Stream.of(o.get()) : Stream.empty())
+            .collect(Collectors.toList());
+
+    // Construct child node for each partition & build the bucket identifier
+    final Set<String> partitions = new HashSet<>(filteredRecordsRDD.map(HoodieRecord::getPartitionPath).distinct().collectAsList());
+    Map<String, HoodieConsistentHashingMetadata> partitionToHashingMeta = new HashMap<>();
+    Map<String, String> partitionToInstant = new HashMap<>();
+    for (Pair<HoodieInstant, HoodieClusteringPlan> pair : instantPlanPairs) {
+      String instant = pair.getLeft().getTimestamp();
+      HoodieClusteringPlan plan = pair.getRight();
+      extractHashingMetadataFromClusteringPlan(instant, plan, partitions, partitionToHashingMeta, partitionToInstant);
+    }
+    Map<String, ConsistentBucketIdentifier> partitionToIdentifier = partitionToHashingMeta.entrySet().stream()
+        .collect(Collectors.toMap(Map.Entry::getKey, e -> new ConsistentBucketIdentifier(e.getValue())));
+
+    // Produce records tagged with new record location
+    List<String> indexKeyFields = Arrays.asList(table.getConfig().getBucketIndexHashField().split(","));
+    HoodieData<HoodieRecord<T>> redirectedRecordsRDD = filteredRecordsRDD.map(r -> {
+      ConsistentHashingNode node = partitionToIdentifier.get(r.getPartitionPath()).getBucket(r.getKey(), indexKeyFields);
+      return HoodieIndexUtils.getTaggedRecord(new HoodieAvroRecord(r.getKey(), r.getData(), r.getOperation()),
+          Option.ofNullable(new HoodieRecordLocation(partitionToInstant.get(r.getPartitionPath()), FSUtils.createNewFileId(node.getFileIdPrefix(), 0))));
+    });
+
+    // Return combined iterator (the original and records with new location)
+    return Pair.of(taggedRecordsRDD.union(redirectedRecordsRDD), Collections.emptySet());
+  }
+
+  private void extractHashingMetadataFromClusteringPlan(String instant, HoodieClusteringPlan plan, final Set<String> recordPartitions,
+                                                        Map<String, HoodieConsistentHashingMetadata> partitionToHashingMeta, Map<String, String> partitionToInstant) {
+    for (HoodieClusteringGroup group : plan.getInputGroups()) {
+      Map<String, String> groupMeta = group.getExtraMetadata();
+      String p = groupMeta.get(SparkConsistentBucketClusteringPlanStrategy.METADATA_PARTITION_KEY);
+      // Skip unrelated clustering group
+      if (!recordPartitions.contains(p)) {
+        return;
+      }
+
+      String preInstant = partitionToInstant.putIfAbsent(p, instant);
+      ValidationUtils.checkState(preInstant == null || preInstant.equals(instant), "Find a partition: " + p + " with two clustering instants");
+      if (!partitionToHashingMeta.containsKey(p)) {
+        partitionToHashingMeta.put(p, HoodieSparkConsistentBucketIndex.loadMetadata(table, p));

Review Comment:
   `loadMetadata` could return null. Preferably, handle that case.



##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/execution/bulkinsert/RDDBucketIndexPartitioner.java:
##########
@@ -0,0 +1,33 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.execution.bulkinsert;
+
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.model.HoodieRecordPayload;
+import org.apache.hudi.table.BulkInsertPartitioner;
+
+import org.apache.spark.api.java.JavaRDD;
+
+/**
+ * Abstract of bucket index bulk_insert partitioner
+ * TODO implement partitioner for SIMPLE BUCKET INDEX

Review Comment:
   Let's add JIRA ticket tracking tihs TODO.



##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/execution/bulkinsert/BulkInsertInternalPartitionerFactory.java:
##########
@@ -18,15 +18,26 @@
 
 package org.apache.hudi.execution.bulkinsert;
 
+import org.apache.hudi.config.HoodieWriteConfig;
 import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.index.HoodieIndex;
 import org.apache.hudi.table.BulkInsertPartitioner;
+import org.apache.hudi.table.HoodieTable;
 
 /**
  * A factory to generate built-in partitioner to repartition input records into at least
  * expected number of output spark partitions for bulk insert operation.
  */
 public abstract class BulkInsertInternalPartitionerFactory {
 
+  public static BulkInsertPartitioner get(HoodieTable table, HoodieWriteConfig config) {

Review Comment:
   This will get called in executor right? If yes, then i'm a bit worried that `RDDConsistentBucketPartitioner` has `HoodieTable` and `HoodieWriteConfig` as constructor arguments. There could be significant serialization cost.



##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/execution/bulkinsert/RDDConsistentBucketPartitioner.java:
##########
@@ -0,0 +1,280 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.execution.bulkinsert;
+
+import org.apache.hudi.avro.HoodieAvroUtils;
+import org.apache.hudi.common.config.SerializableSchema;
+import org.apache.hudi.common.fs.FSUtils;
+import org.apache.hudi.common.model.ConsistentHashingNode;
+import org.apache.hudi.common.model.HoodieConsistentHashingMetadata;
+import org.apache.hudi.common.model.HoodieKey;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.model.HoodieRecordPayload;
+import org.apache.hudi.common.model.HoodieTableType;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.ValidationUtils;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.index.bucket.ConsistentBucketIdentifier;
+import org.apache.hudi.index.bucket.HoodieSparkConsistentBucketIndex;
+import org.apache.hudi.io.AppendHandleFactory;
+import org.apache.hudi.io.SingleFileHandleCreateFactory;
+import org.apache.hudi.io.WriteHandleFactory;
+import org.apache.hudi.table.HoodieTable;
+
+import org.apache.avro.Schema;
+import org.apache.log4j.LogManager;
+import org.apache.log4j.Logger;
+import org.apache.spark.Partitioner;
+import org.apache.spark.api.java.JavaRDD;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import scala.Tuple2;
+
+import static org.apache.hudi.config.HoodieClusteringConfig.PLAN_STRATEGY_SORT_COLUMNS;
+
+/**
+ * A partitioner for (consistent hashing) bucket index used in bulk_insert
+ */
+public class RDDConsistentBucketPartitioner<T extends HoodieRecordPayload> extends RDDBucketIndexPartitioner<T> {
+
+  private static final Logger LOG = LogManager.getLogger(RDDConsistentBucketPartitioner.class);
+
+  private final HoodieTable table;
+  private final HoodieWriteConfig config;
+  private final List<String> indexKeyFields;
+  private final Map<String, List<ConsistentHashingNode>> hashingChildrenNodes;
+  private final String[] sortColumnNames;
+  private final boolean preserveHoodieMetadata;
+  private final boolean consistentLogicalTimestampEnabled;
+
+  private List<Boolean> doAppend;
+  private List<String> fileIdPfxList;
+
+  public RDDConsistentBucketPartitioner(HoodieTable table, HoodieWriteConfig config, Map<String, String> strategyParams, boolean preserveHoodieMetadata) {
+    this.table = table;
+    this.config = config;
+    this.indexKeyFields = Arrays.asList(config.getBucketIndexHashField().split(","));
+    this.hashingChildrenNodes = new HashMap<>();
+    this.consistentLogicalTimestampEnabled = config.isConsistentLogicalTimestampEnabled();
+    this.preserveHoodieMetadata = preserveHoodieMetadata;
+
+    if (strategyParams.containsKey(PLAN_STRATEGY_SORT_COLUMNS.key())) {
+      sortColumnNames = strategyParams.get(PLAN_STRATEGY_SORT_COLUMNS.key()).split(",");
+    } else {
+      sortColumnNames = null;
+    }
+  }
+
+  public RDDConsistentBucketPartitioner(HoodieTable table, HoodieWriteConfig config) {
+    this(table, config, Collections.emptyMap(), false);
+    ValidationUtils.checkArgument(table.getIndex() instanceof HoodieSparkConsistentBucketIndex,
+        "RDDConsistentBucketPartitioner can only be used together with consistent hashing bucket index");
+    ValidationUtils.checkArgument(table.getMetaClient().getTableType().equals(HoodieTableType.MERGE_ON_READ),
+        "CoW table with bucket index doesn't support bulk_insert");
+  }
+
+  /**
+   * Repartition the records to conform the bucket index storage layout constraints.
+   * Specifically, partition the records based on consistent bucket index, which is computed
+   * using hashing metadata and records' key.
+   *
+   * @param records               Input Hoodie records
+   * @param outputSparkPartitions Not used
+   * @return partitioned records, each partition of data corresponds to a bucket (i.e., file group)
+   */
+  @Override
+  public JavaRDD<HoodieRecord<T>> repartitionRecords(JavaRDD<HoodieRecord<T>> records, int outputSparkPartitions) {
+    Map<String, ConsistentBucketIdentifier> partitionToIdentifier = initializeBucketIdentifier(records);
+    Map<String, Map<String, Integer>> partitionToFileIdPfxIdxMap = generateFileIdPfx(partitionToIdentifier, outputSparkPartitions);
+    return doPartition(records, new Partitioner() {
+      @Override
+      public int numPartitions() {
+        return fileIdPfxList.size();
+      }
+
+      @Override
+      public int getPartition(Object key) {
+        HoodieKey hoodieKey = (HoodieKey) key;
+        String partition = hoodieKey.getPartitionPath();
+        ConsistentHashingNode node = partitionToIdentifier.get(partition).getBucket(hoodieKey, indexKeyFields);
+        return partitionToFileIdPfxIdxMap.get(partition).get(node.getFileIdPrefix());
+      }
+    });
+  }
+
+  @Override
+  public boolean arePartitionRecordsSorted() {
+    return (sortColumnNames != null && sortColumnNames.length > 0)
+        || table.requireSortedRecords() || config.getBulkInsertSortMode() != BulkInsertSortMode.NONE;
+  }
+
+  @Override
+  public Option<WriteHandleFactory> getWriteHandleFactory(int idx) {
+    return doAppend.get(idx) ? Option.of(new AppendHandleFactory()) :
+        Option.of(new SingleFileHandleCreateFactory(FSUtils.createNewFileId(getFileIdPfx(idx), 0), this.preserveHoodieMetadata));
+  }
+
+  @Override
+  public String getFileIdPfx(int partitionId) {
+    return fileIdPfxList.get(partitionId);
+  }
+
+  public void addHashingChildrenNodes(String partition, List<ConsistentHashingNode> nodes) {
+    ValidationUtils.checkState(nodes.stream().noneMatch(n -> n.getTag() == ConsistentHashingNode.NodeTag.NORMAL), "children nodes should not be tagged as NORMAL");
+    hashingChildrenNodes.put(partition, nodes);
+  }
+
+  /**
+   * Get (construct) the bucket identifier of the given partition
+   */
+  private ConsistentBucketIdentifier getBucketIdentifier(String partition) {
+    HoodieSparkConsistentBucketIndex index = (HoodieSparkConsistentBucketIndex) table.getIndex();
+    HoodieConsistentHashingMetadata metadata = index.loadOrCreateMetadata(this.table, partition);
+    if (hashingChildrenNodes.containsKey(partition)) {
+      metadata.setChildrenNodes(hashingChildrenNodes.get(partition));
+    }
+    return new ConsistentBucketIdentifier(metadata);
+  }
+
+  /**
+   * Initialize hashing metadata of input records. The metadata of all related partitions will be loaded, and
+   * the mapping from partition to its bucket identifier is constructed.
+   */
+  private Map<String, ConsistentBucketIdentifier> initializeBucketIdentifier(JavaRDD<HoodieRecord<T>> records) {
+    return records.map(HoodieRecord::getPartitionPath).distinct().collect().stream()
+        .collect(Collectors.toMap(p -> p, p -> getBucketIdentifier(p)));
+  }
+
+  /**
+   * Initialize fileIdPfx for each data partition. Specifically, the following fields is constructed:

Review Comment:
   javadoc does not match the arguments. 



##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/execution/bulkinsert/RDDConsistentBucketPartitioner.java:
##########
@@ -0,0 +1,280 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.execution.bulkinsert;
+
+import org.apache.hudi.avro.HoodieAvroUtils;
+import org.apache.hudi.common.config.SerializableSchema;
+import org.apache.hudi.common.fs.FSUtils;
+import org.apache.hudi.common.model.ConsistentHashingNode;
+import org.apache.hudi.common.model.HoodieConsistentHashingMetadata;
+import org.apache.hudi.common.model.HoodieKey;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.model.HoodieRecordPayload;
+import org.apache.hudi.common.model.HoodieTableType;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.ValidationUtils;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.index.bucket.ConsistentBucketIdentifier;
+import org.apache.hudi.index.bucket.HoodieSparkConsistentBucketIndex;
+import org.apache.hudi.io.AppendHandleFactory;
+import org.apache.hudi.io.SingleFileHandleCreateFactory;
+import org.apache.hudi.io.WriteHandleFactory;
+import org.apache.hudi.table.HoodieTable;
+
+import org.apache.avro.Schema;
+import org.apache.log4j.LogManager;
+import org.apache.log4j.Logger;
+import org.apache.spark.Partitioner;
+import org.apache.spark.api.java.JavaRDD;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import scala.Tuple2;
+
+import static org.apache.hudi.config.HoodieClusteringConfig.PLAN_STRATEGY_SORT_COLUMNS;
+
+/**
+ * A partitioner for (consistent hashing) bucket index used in bulk_insert
+ */
+public class RDDConsistentBucketPartitioner<T extends HoodieRecordPayload> extends RDDBucketIndexPartitioner<T> {
+
+  private static final Logger LOG = LogManager.getLogger(RDDConsistentBucketPartitioner.class);
+
+  private final HoodieTable table;
+  private final HoodieWriteConfig config;
+  private final List<String> indexKeyFields;
+  private final Map<String, List<ConsistentHashingNode>> hashingChildrenNodes;
+  private final String[] sortColumnNames;
+  private final boolean preserveHoodieMetadata;
+  private final boolean consistentLogicalTimestampEnabled;
+
+  private List<Boolean> doAppend;
+  private List<String> fileIdPfxList;
+
+  public RDDConsistentBucketPartitioner(HoodieTable table, HoodieWriteConfig config, Map<String, String> strategyParams, boolean preserveHoodieMetadata) {
+    this.table = table;
+    this.config = config;
+    this.indexKeyFields = Arrays.asList(config.getBucketIndexHashField().split(","));
+    this.hashingChildrenNodes = new HashMap<>();
+    this.consistentLogicalTimestampEnabled = config.isConsistentLogicalTimestampEnabled();
+    this.preserveHoodieMetadata = preserveHoodieMetadata;
+
+    if (strategyParams.containsKey(PLAN_STRATEGY_SORT_COLUMNS.key())) {
+      sortColumnNames = strategyParams.get(PLAN_STRATEGY_SORT_COLUMNS.key()).split(",");
+    } else {
+      sortColumnNames = null;
+    }
+  }
+
+  public RDDConsistentBucketPartitioner(HoodieTable table, HoodieWriteConfig config) {
+    this(table, config, Collections.emptyMap(), false);
+    ValidationUtils.checkArgument(table.getIndex() instanceof HoodieSparkConsistentBucketIndex,
+        "RDDConsistentBucketPartitioner can only be used together with consistent hashing bucket index");
+    ValidationUtils.checkArgument(table.getMetaClient().getTableType().equals(HoodieTableType.MERGE_ON_READ),
+        "CoW table with bucket index doesn't support bulk_insert");
+  }
+
+  /**
+   * Repartition the records to conform the bucket index storage layout constraints.
+   * Specifically, partition the records based on consistent bucket index, which is computed
+   * using hashing metadata and records' key.
+   *
+   * @param records               Input Hoodie records
+   * @param outputSparkPartitions Not used
+   * @return partitioned records, each partition of data corresponds to a bucket (i.e., file group)
+   */
+  @Override
+  public JavaRDD<HoodieRecord<T>> repartitionRecords(JavaRDD<HoodieRecord<T>> records, int outputSparkPartitions) {
+    Map<String, ConsistentBucketIdentifier> partitionToIdentifier = initializeBucketIdentifier(records);
+    Map<String, Map<String, Integer>> partitionToFileIdPfxIdxMap = generateFileIdPfx(partitionToIdentifier, outputSparkPartitions);
+    return doPartition(records, new Partitioner() {
+      @Override
+      public int numPartitions() {
+        return fileIdPfxList.size();
+      }
+
+      @Override
+      public int getPartition(Object key) {
+        HoodieKey hoodieKey = (HoodieKey) key;
+        String partition = hoodieKey.getPartitionPath();
+        ConsistentHashingNode node = partitionToIdentifier.get(partition).getBucket(hoodieKey, indexKeyFields);
+        return partitionToFileIdPfxIdxMap.get(partition).get(node.getFileIdPrefix());
+      }
+    });
+  }
+
+  @Override
+  public boolean arePartitionRecordsSorted() {
+    return (sortColumnNames != null && sortColumnNames.length > 0)
+        || table.requireSortedRecords() || config.getBulkInsertSortMode() != BulkInsertSortMode.NONE;
+  }
+
+  @Override
+  public Option<WriteHandleFactory> getWriteHandleFactory(int idx) {
+    return doAppend.get(idx) ? Option.of(new AppendHandleFactory()) :

Review Comment:
   Probably, we may never hit null `this.writeHandleFactory` in `BulkInsertMapFunction` due to this overridden method, but it's still better to handle null there using Option<WriteHandleFactory> as suggested.



##########
hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/testutils/HoodieMergeOnReadTestUtils.java:
##########
@@ -51,6 +51,45 @@
  */
 public class HoodieMergeOnReadTestUtils {
 
+  public static List<RecordReader> getRecordReadersUsingInputFormat(Configuration conf, List<String> inputPaths,
+                                                                    String basePath, JobConf jobConf, boolean realtime, boolean populateMetaField) {
+    Schema schema = new Schema.Parser().parse(HoodieTestDataGenerator.TRIP_EXAMPLE_SCHEMA);
+    return getRecordReadersUsingInputFormat(conf, inputPaths, basePath, jobConf, realtime, schema,
+        HoodieTestDataGenerator.TRIP_HIVE_COLUMN_TYPES, false, new ArrayList<>(), populateMetaField);
+  }
+
+  public static List<RecordReader> getRecordReadersUsingInputFormat(Configuration conf, List<String> inputPaths, String basePath, JobConf jobConf, boolean realtime, Schema rawSchema,
+                                                                    String rawHiveColumnTypes, boolean projectCols, List<String> projectedColumns, boolean populateMetaFields) {
+    HoodieTableMetaClient metaClient = HoodieTableMetaClient.builder().setConf(conf).setBasePath(basePath).build();
+    FileInputFormat inputFormat = HoodieInputFormatUtils.getInputFormat(metaClient.getTableConfig().getBaseFileFormat(), realtime, jobConf);
+    Schema schema;
+    String hiveColumnTypes;
+
+    if (populateMetaFields) {
+      schema = HoodieAvroUtils.addMetadataFields(rawSchema);
+      hiveColumnTypes = HoodieAvroUtils.addMetadataColumnTypes(rawHiveColumnTypes);
+    } else {
+      schema = rawSchema;
+      hiveColumnTypes = rawHiveColumnTypes;
+    }
+
+    setPropsForInputFormat(inputFormat, jobConf, schema, hiveColumnTypes, projectCols, projectedColumns, populateMetaFields);
+
+    try {
+      FileInputFormat.setInputPaths(jobConf, String.join(",", inputPaths));
+      InputSplit[] splits = inputFormat.getSplits(jobConf, inputPaths.size());
+
+      List<RecordReader> recordReaders = new ArrayList<>();
+      for (InputSplit split : splits) {
+        recordReaders.add(inputFormat.getRecordReader(split, jobConf, null));
+      }
+      return recordReaders;
+    } catch (IOException e) {
+      e.printStackTrace();

Review Comment:
   let's just LOG.error



##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/commit/BaseSparkCommitActionExecutor.java:
##########
@@ -344,6 +353,12 @@ public Iterator<List<WriteStatus>> handleUpdate(String partitionPath, String fil
       LOG.info("Empty partition with fileId => " + fileId);
       return Collections.emptyIterator();
     }
+
+    // Pre-check: if the old file does not exist (which may happen in bucket index case), fallback to insert
+    if (!table.getBaseFileOnlyView().getLatestBaseFile(partitionPath, fileId).isPresent()) {

Review Comment:
   ```suggestion
       if (!table.getBaseFileOnlyView().getLatestBaseFile(partitionPath, fileId).isPresent() && HoodieIndex.IndexType.BUCKET.equals(config.getIndexType())) {
   ```



##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/execution/bulkinsert/RDDConsistentBucketPartitioner.java:
##########
@@ -0,0 +1,280 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.execution.bulkinsert;
+
+import org.apache.hudi.avro.HoodieAvroUtils;
+import org.apache.hudi.common.config.SerializableSchema;
+import org.apache.hudi.common.fs.FSUtils;
+import org.apache.hudi.common.model.ConsistentHashingNode;
+import org.apache.hudi.common.model.HoodieConsistentHashingMetadata;
+import org.apache.hudi.common.model.HoodieKey;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.model.HoodieRecordPayload;
+import org.apache.hudi.common.model.HoodieTableType;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.ValidationUtils;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.index.bucket.ConsistentBucketIdentifier;
+import org.apache.hudi.index.bucket.HoodieSparkConsistentBucketIndex;
+import org.apache.hudi.io.AppendHandleFactory;
+import org.apache.hudi.io.SingleFileHandleCreateFactory;
+import org.apache.hudi.io.WriteHandleFactory;
+import org.apache.hudi.table.HoodieTable;
+
+import org.apache.avro.Schema;
+import org.apache.log4j.LogManager;
+import org.apache.log4j.Logger;
+import org.apache.spark.Partitioner;
+import org.apache.spark.api.java.JavaRDD;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import scala.Tuple2;
+
+import static org.apache.hudi.config.HoodieClusteringConfig.PLAN_STRATEGY_SORT_COLUMNS;
+
+/**
+ * A partitioner for (consistent hashing) bucket index used in bulk_insert
+ */
+public class RDDConsistentBucketPartitioner<T extends HoodieRecordPayload> extends RDDBucketIndexPartitioner<T> {
+
+  private static final Logger LOG = LogManager.getLogger(RDDConsistentBucketPartitioner.class);
+
+  private final HoodieTable table;
+  private final HoodieWriteConfig config;
+  private final List<String> indexKeyFields;
+  private final Map<String, List<ConsistentHashingNode>> hashingChildrenNodes;
+  private final String[] sortColumnNames;
+  private final boolean preserveHoodieMetadata;
+  private final boolean consistentLogicalTimestampEnabled;
+
+  private List<Boolean> doAppend;
+  private List<String> fileIdPfxList;
+
+  public RDDConsistentBucketPartitioner(HoodieTable table, HoodieWriteConfig config, Map<String, String> strategyParams, boolean preserveHoodieMetadata) {
+    this.table = table;
+    this.config = config;
+    this.indexKeyFields = Arrays.asList(config.getBucketIndexHashField().split(","));
+    this.hashingChildrenNodes = new HashMap<>();
+    this.consistentLogicalTimestampEnabled = config.isConsistentLogicalTimestampEnabled();
+    this.preserveHoodieMetadata = preserveHoodieMetadata;
+
+    if (strategyParams.containsKey(PLAN_STRATEGY_SORT_COLUMNS.key())) {
+      sortColumnNames = strategyParams.get(PLAN_STRATEGY_SORT_COLUMNS.key()).split(",");
+    } else {
+      sortColumnNames = null;
+    }
+  }
+
+  public RDDConsistentBucketPartitioner(HoodieTable table, HoodieWriteConfig config) {
+    this(table, config, Collections.emptyMap(), false);
+    ValidationUtils.checkArgument(table.getIndex() instanceof HoodieSparkConsistentBucketIndex,
+        "RDDConsistentBucketPartitioner can only be used together with consistent hashing bucket index");
+    ValidationUtils.checkArgument(table.getMetaClient().getTableType().equals(HoodieTableType.MERGE_ON_READ),
+        "CoW table with bucket index doesn't support bulk_insert");
+  }
+
+  /**
+   * Repartition the records to conform the bucket index storage layout constraints.
+   * Specifically, partition the records based on consistent bucket index, which is computed
+   * using hashing metadata and records' key.
+   *
+   * @param records               Input Hoodie records
+   * @param outputSparkPartitions Not used
+   * @return partitioned records, each partition of data corresponds to a bucket (i.e., file group)
+   */
+  @Override
+  public JavaRDD<HoodieRecord<T>> repartitionRecords(JavaRDD<HoodieRecord<T>> records, int outputSparkPartitions) {
+    Map<String, ConsistentBucketIdentifier> partitionToIdentifier = initializeBucketIdentifier(records);
+    Map<String, Map<String, Integer>> partitionToFileIdPfxIdxMap = generateFileIdPfx(partitionToIdentifier, outputSparkPartitions);
+    return doPartition(records, new Partitioner() {
+      @Override
+      public int numPartitions() {
+        return fileIdPfxList.size();
+      }
+
+      @Override
+      public int getPartition(Object key) {
+        HoodieKey hoodieKey = (HoodieKey) key;
+        String partition = hoodieKey.getPartitionPath();
+        ConsistentHashingNode node = partitionToIdentifier.get(partition).getBucket(hoodieKey, indexKeyFields);
+        return partitionToFileIdPfxIdxMap.get(partition).get(node.getFileIdPrefix());
+      }
+    });
+  }
+
+  @Override
+  public boolean arePartitionRecordsSorted() {
+    return (sortColumnNames != null && sortColumnNames.length > 0)
+        || table.requireSortedRecords() || config.getBulkInsertSortMode() != BulkInsertSortMode.NONE;
+  }
+
+  @Override
+  public Option<WriteHandleFactory> getWriteHandleFactory(int idx) {
+    return doAppend.get(idx) ? Option.of(new AppendHandleFactory()) :
+        Option.of(new SingleFileHandleCreateFactory(FSUtils.createNewFileId(getFileIdPfx(idx), 0), this.preserveHoodieMetadata));
+  }
+
+  @Override
+  public String getFileIdPfx(int partitionId) {
+    return fileIdPfxList.get(partitionId);
+  }
+
+  public void addHashingChildrenNodes(String partition, List<ConsistentHashingNode> nodes) {
+    ValidationUtils.checkState(nodes.stream().noneMatch(n -> n.getTag() == ConsistentHashingNode.NodeTag.NORMAL), "children nodes should not be tagged as NORMAL");
+    hashingChildrenNodes.put(partition, nodes);
+  }
+
+  /**
+   * Get (construct) the bucket identifier of the given partition
+   */
+  private ConsistentBucketIdentifier getBucketIdentifier(String partition) {
+    HoodieSparkConsistentBucketIndex index = (HoodieSparkConsistentBucketIndex) table.getIndex();
+    HoodieConsistentHashingMetadata metadata = index.loadOrCreateMetadata(this.table, partition);
+    if (hashingChildrenNodes.containsKey(partition)) {
+      metadata.setChildrenNodes(hashingChildrenNodes.get(partition));
+    }
+    return new ConsistentBucketIdentifier(metadata);
+  }
+
+  /**
+   * Initialize hashing metadata of input records. The metadata of all related partitions will be loaded, and
+   * the mapping from partition to its bucket identifier is constructed.
+   */
+  private Map<String, ConsistentBucketIdentifier> initializeBucketIdentifier(JavaRDD<HoodieRecord<T>> records) {
+    return records.map(HoodieRecord::getPartitionPath).distinct().collect().stream()
+        .collect(Collectors.toMap(p -> p, p -> getBucketIdentifier(p)));

Review Comment:
   Note that Collectors.toMap is not concurrent. Would there be a need for `mergeFunction` to break ties?



##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/execution/bulkinsert/RDDConsistentBucketPartitioner.java:
##########
@@ -0,0 +1,280 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.execution.bulkinsert;
+
+import org.apache.hudi.avro.HoodieAvroUtils;
+import org.apache.hudi.common.config.SerializableSchema;
+import org.apache.hudi.common.fs.FSUtils;
+import org.apache.hudi.common.model.ConsistentHashingNode;
+import org.apache.hudi.common.model.HoodieConsistentHashingMetadata;
+import org.apache.hudi.common.model.HoodieKey;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.model.HoodieRecordPayload;
+import org.apache.hudi.common.model.HoodieTableType;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.ValidationUtils;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.index.bucket.ConsistentBucketIdentifier;
+import org.apache.hudi.index.bucket.HoodieSparkConsistentBucketIndex;
+import org.apache.hudi.io.AppendHandleFactory;
+import org.apache.hudi.io.SingleFileHandleCreateFactory;
+import org.apache.hudi.io.WriteHandleFactory;
+import org.apache.hudi.table.HoodieTable;
+
+import org.apache.avro.Schema;
+import org.apache.log4j.LogManager;
+import org.apache.log4j.Logger;
+import org.apache.spark.Partitioner;
+import org.apache.spark.api.java.JavaRDD;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import scala.Tuple2;
+
+import static org.apache.hudi.config.HoodieClusteringConfig.PLAN_STRATEGY_SORT_COLUMNS;
+
+/**
+ * A partitioner for (consistent hashing) bucket index used in bulk_insert
+ */
+public class RDDConsistentBucketPartitioner<T extends HoodieRecordPayload> extends RDDBucketIndexPartitioner<T> {
+
+  private static final Logger LOG = LogManager.getLogger(RDDConsistentBucketPartitioner.class);
+
+  private final HoodieTable table;
+  private final HoodieWriteConfig config;
+  private final List<String> indexKeyFields;
+  private final Map<String, List<ConsistentHashingNode>> hashingChildrenNodes;
+  private final String[] sortColumnNames;
+  private final boolean preserveHoodieMetadata;
+  private final boolean consistentLogicalTimestampEnabled;
+
+  private List<Boolean> doAppend;
+  private List<String> fileIdPfxList;
+
+  public RDDConsistentBucketPartitioner(HoodieTable table, HoodieWriteConfig config, Map<String, String> strategyParams, boolean preserveHoodieMetadata) {
+    this.table = table;
+    this.config = config;

Review Comment:
   `HoodieTable` should already have the config. Probably, we can reduce one argument.



##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/execution/bulkinsert/RDDConsistentBucketPartitioner.java:
##########
@@ -0,0 +1,280 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.execution.bulkinsert;
+
+import org.apache.hudi.avro.HoodieAvroUtils;
+import org.apache.hudi.common.config.SerializableSchema;
+import org.apache.hudi.common.fs.FSUtils;
+import org.apache.hudi.common.model.ConsistentHashingNode;
+import org.apache.hudi.common.model.HoodieConsistentHashingMetadata;
+import org.apache.hudi.common.model.HoodieKey;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.model.HoodieRecordPayload;
+import org.apache.hudi.common.model.HoodieTableType;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.ValidationUtils;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.index.bucket.ConsistentBucketIdentifier;
+import org.apache.hudi.index.bucket.HoodieSparkConsistentBucketIndex;
+import org.apache.hudi.io.AppendHandleFactory;
+import org.apache.hudi.io.SingleFileHandleCreateFactory;
+import org.apache.hudi.io.WriteHandleFactory;
+import org.apache.hudi.table.HoodieTable;
+
+import org.apache.avro.Schema;
+import org.apache.log4j.LogManager;
+import org.apache.log4j.Logger;
+import org.apache.spark.Partitioner;
+import org.apache.spark.api.java.JavaRDD;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import scala.Tuple2;
+
+import static org.apache.hudi.config.HoodieClusteringConfig.PLAN_STRATEGY_SORT_COLUMNS;
+
+/**
+ * A partitioner for (consistent hashing) bucket index used in bulk_insert
+ */
+public class RDDConsistentBucketPartitioner<T extends HoodieRecordPayload> extends RDDBucketIndexPartitioner<T> {
+
+  private static final Logger LOG = LogManager.getLogger(RDDConsistentBucketPartitioner.class);
+
+  private final HoodieTable table;
+  private final HoodieWriteConfig config;
+  private final List<String> indexKeyFields;
+  private final Map<String, List<ConsistentHashingNode>> hashingChildrenNodes;
+  private final String[] sortColumnNames;
+  private final boolean preserveHoodieMetadata;
+  private final boolean consistentLogicalTimestampEnabled;
+
+  private List<Boolean> doAppend;
+  private List<String> fileIdPfxList;
+
+  public RDDConsistentBucketPartitioner(HoodieTable table, HoodieWriteConfig config, Map<String, String> strategyParams, boolean preserveHoodieMetadata) {
+    this.table = table;
+    this.config = config;
+    this.indexKeyFields = Arrays.asList(config.getBucketIndexHashField().split(","));
+    this.hashingChildrenNodes = new HashMap<>();
+    this.consistentLogicalTimestampEnabled = config.isConsistentLogicalTimestampEnabled();
+    this.preserveHoodieMetadata = preserveHoodieMetadata;
+
+    if (strategyParams.containsKey(PLAN_STRATEGY_SORT_COLUMNS.key())) {
+      sortColumnNames = strategyParams.get(PLAN_STRATEGY_SORT_COLUMNS.key()).split(",");
+    } else {
+      sortColumnNames = null;
+    }
+  }
+
+  public RDDConsistentBucketPartitioner(HoodieTable table, HoodieWriteConfig config) {
+    this(table, config, Collections.emptyMap(), false);
+    ValidationUtils.checkArgument(table.getIndex() instanceof HoodieSparkConsistentBucketIndex,
+        "RDDConsistentBucketPartitioner can only be used together with consistent hashing bucket index");
+    ValidationUtils.checkArgument(table.getMetaClient().getTableType().equals(HoodieTableType.MERGE_ON_READ),
+        "CoW table with bucket index doesn't support bulk_insert");
+  }
+
+  /**
+   * Repartition the records to conform the bucket index storage layout constraints.
+   * Specifically, partition the records based on consistent bucket index, which is computed
+   * using hashing metadata and records' key.
+   *
+   * @param records               Input Hoodie records
+   * @param outputSparkPartitions Not used
+   * @return partitioned records, each partition of data corresponds to a bucket (i.e., file group)
+   */
+  @Override
+  public JavaRDD<HoodieRecord<T>> repartitionRecords(JavaRDD<HoodieRecord<T>> records, int outputSparkPartitions) {
+    Map<String, ConsistentBucketIdentifier> partitionToIdentifier = initializeBucketIdentifier(records);
+    Map<String, Map<String, Integer>> partitionToFileIdPfxIdxMap = generateFileIdPfx(partitionToIdentifier, outputSparkPartitions);
+    return doPartition(records, new Partitioner() {
+      @Override
+      public int numPartitions() {
+        return fileIdPfxList.size();
+      }
+
+      @Override
+      public int getPartition(Object key) {
+        HoodieKey hoodieKey = (HoodieKey) key;
+        String partition = hoodieKey.getPartitionPath();
+        ConsistentHashingNode node = partitionToIdentifier.get(partition).getBucket(hoodieKey, indexKeyFields);
+        return partitionToFileIdPfxIdxMap.get(partition).get(node.getFileIdPrefix());
+      }
+    });
+  }
+
+  @Override
+  public boolean arePartitionRecordsSorted() {
+    return (sortColumnNames != null && sortColumnNames.length > 0)
+        || table.requireSortedRecords() || config.getBulkInsertSortMode() != BulkInsertSortMode.NONE;
+  }
+
+  @Override
+  public Option<WriteHandleFactory> getWriteHandleFactory(int idx) {
+    return doAppend.get(idx) ? Option.of(new AppendHandleFactory()) :
+        Option.of(new SingleFileHandleCreateFactory(FSUtils.createNewFileId(getFileIdPfx(idx), 0), this.preserveHoodieMetadata));
+  }
+
+  @Override
+  public String getFileIdPfx(int partitionId) {
+    return fileIdPfxList.get(partitionId);
+  }
+
+  public void addHashingChildrenNodes(String partition, List<ConsistentHashingNode> nodes) {
+    ValidationUtils.checkState(nodes.stream().noneMatch(n -> n.getTag() == ConsistentHashingNode.NodeTag.NORMAL), "children nodes should not be tagged as NORMAL");
+    hashingChildrenNodes.put(partition, nodes);
+  }
+
+  /**
+   * Get (construct) the bucket identifier of the given partition
+   */
+  private ConsistentBucketIdentifier getBucketIdentifier(String partition) {
+    HoodieSparkConsistentBucketIndex index = (HoodieSparkConsistentBucketIndex) table.getIndex();
+    HoodieConsistentHashingMetadata metadata = index.loadOrCreateMetadata(this.table, partition);
+    if (hashingChildrenNodes.containsKey(partition)) {
+      metadata.setChildrenNodes(hashingChildrenNodes.get(partition));
+    }
+    return new ConsistentBucketIdentifier(metadata);
+  }
+
+  /**
+   * Initialize hashing metadata of input records. The metadata of all related partitions will be loaded, and
+   * the mapping from partition to its bucket identifier is constructed.
+   */
+  private Map<String, ConsistentBucketIdentifier> initializeBucketIdentifier(JavaRDD<HoodieRecord<T>> records) {
+    return records.map(HoodieRecord::getPartitionPath).distinct().collect().stream()
+        .collect(Collectors.toMap(p -> p, p -> getBucketIdentifier(p)));
+  }
+
+  /**
+   * Initialize fileIdPfx for each data partition. Specifically, the following fields is constructed:
+   * - fileIdPfx: the Nth element corresponds to the Nth data partition, indicating its fileIdPfx
+   * - doAppend: represents if the Nth data partition should use AppendHandler
+   * - partitionToFileIdPfxIdxMap (return value): (table partition) -> (fileIdPfx -> idx) mapping
+   *
+   * @param parallelism Not used, the actual parallelism is determined by the bucket number
+   */
+  private Map<String, Map<String, Integer>> generateFileIdPfx(Map<String, ConsistentBucketIdentifier> partitionToIdentifier, int parallelism) {
+    Map<String, Map<String, Integer>> partitionToFileIdPfxIdxMap = new HashMap(partitionToIdentifier.size() * 2);
+    doAppend = new ArrayList<>();
+    fileIdPfxList = new ArrayList<>();
+    int count = 0;
+    for (ConsistentBucketIdentifier identifier : partitionToIdentifier.values()) {
+      Map<String, Integer> fileIdPfxToIdx = new HashMap();
+      for (ConsistentHashingNode node : identifier.getNodes()) {
+        fileIdPfxToIdx.put(node.getFileIdPrefix(), count++);
+      }
+      fileIdPfxList.addAll(identifier.getNodes().stream().map(ConsistentHashingNode::getFileIdPrefix).collect(Collectors.toList()));
+      if (identifier.getMetadata().isFirstCreated()) {
+        // Create new file group when the hashing metadata is new (i.e., first write to the partition)
+        doAppend.addAll(Collections.nCopies(identifier.getNodes().size(), false));
+      } else {
+        // Child node requires generating a fresh new base file, rather than log file
+        doAppend.addAll(identifier.getNodes().stream().map(n -> n.getTag() == ConsistentHashingNode.NodeTag.NORMAL).collect(Collectors.toList()));
+      }
+      partitionToFileIdPfxIdxMap.put(identifier.getMetadata().getPartitionPath(), fileIdPfxToIdx);
+    }
+
+    ValidationUtils.checkState(fileIdPfxList.size() == partitionToIdentifier.values().stream().mapToInt(ConsistentBucketIdentifier::getNumBuckets).sum(),
+        "Error state after constructing fileId & idx mapping");
+    return partitionToFileIdPfxIdxMap;
+  }
+
+  /**
+   * Execute partition using the given partitioner.
+   * If sorting is required, will do it within each data partition:
+   * - if sortColumnNames is specified, apply sort to the column (the behaviour is the same as `RDDCustomColumnsSortPartitioner`
+   * - if table requires sort or BulkInsertSortMode is not None, then sort by record key within partition.
+   * By default, do partition only.
+   *
+   * @param records
+   * @param partitioner a default partition that accepts `HoodieKey` as the partition key
+   * @return
+   */
+  private JavaRDD<HoodieRecord<T>> doPartition(JavaRDD<HoodieRecord<T>> records, Partitioner partitioner) {
+    if (sortColumnNames != null && sortColumnNames.length > 0) {
+      return doPartitionAndCustomColumnSort(records, partitioner);
+    } else if (table.requireSortedRecords() || config.getBulkInsertSortMode() != BulkInsertSortMode.NONE) {
+      return doPartitionAndSortByRecordKey(records, partitioner);
+    } else {
+      // By default, do partition only
+      return records.mapToPair(record -> new Tuple2<>(record.getKey(), record))
+          .partitionBy(partitioner).map(Tuple2::_2);
+    }
+  }
+
+  /**
+   * Sort by specified column value. The behaviour is the same as `RDDCustomColumnsSortPartitioner`
+   *
+   * @param records
+   * @param partitioner
+   * @return
+   */
+  private JavaRDD<HoodieRecord<T>> doPartitionAndCustomColumnSort(JavaRDD<HoodieRecord<T>> records, Partitioner partitioner) {
+    final String[] sortColumns = sortColumnNames;
+    final SerializableSchema schema = new SerializableSchema(HoodieAvroUtils.addMetadataFields((new Schema.Parser().parse(config.getSchema()))));
+    Comparator<HoodieRecord<T>> comparator = (Comparator<HoodieRecord<T>> & Serializable) (t1, t2) -> {
+      Object obj1 = HoodieAvroUtils.getRecordColumnValues(t1, sortColumns, schema, consistentLogicalTimestampEnabled);
+      Object obj2 = HoodieAvroUtils.getRecordColumnValues(t2, sortColumns, schema, consistentLogicalTimestampEnabled);
+      return ((Comparable) obj1).compareTo(obj2);
+    };
+
+    return records.mapToPair(record -> new Tuple2<>(record, record))
+        .repartitionAndSortWithinPartitions(new Partitioner() {
+          @Override
+          public int numPartitions() {
+            return partitioner.numPartitions();
+          }
+
+          @Override
+          public int getPartition(Object key) {
+            return partitioner.getPartition(((HoodieRecord) key).getKey());
+          }
+        }, comparator).map(Tuple2::_2);
+  }
+
+  /**
+   * Sort by record key within each partition. The behaviour is the same as BulkInsertSortMode.PARTITION_SORT.
+   *
+   * @param records
+   * @param partitioner
+   * @return
+   */
+  private JavaRDD<HoodieRecord<T>> doPartitionAndSortByRecordKey(JavaRDD<HoodieRecord<T>> records, Partitioner partitioner) {
+    if (config.getBulkInsertSortMode() == BulkInsertSortMode.GLOBAL_SORT) {
+      LOG.warn("Consistent bucket does not support global sort mode, the sort will only be done within each data partition");
+    }
+
+    Comparator<HoodieKey> comparator = (Comparator<HoodieKey> & Serializable) (t1, t2) -> {
+      return t1.getRecordKey().compareTo(t2.getRecordKey());
+    };

Review Comment:
   ```suggestion
       Comparator<HoodieKey> comparator = (Comparator<HoodieKey> & Serializable) (t1, t2) -> t1.getRecordKey().compareTo(t2.getRecordKey());
   ```



##########
hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestSparkConsistentBucketClustering.java:
##########
@@ -0,0 +1,296 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.client.functional;
+
+import org.apache.hudi.client.WriteStatus;
+import org.apache.hudi.client.clustering.plan.strategy.SparkConsistentBucketClusteringPlanStrategy;
+import org.apache.hudi.client.clustering.run.strategy.SparkConsistentBucketClusteringExecutionStrategy;
+import org.apache.hudi.client.clustering.update.strategy.SparkConsistentHashingDuplicateUpdateStrategy;
+import org.apache.hudi.common.fs.ConsistencyGuardConfig;
+import org.apache.hudi.common.model.HoodieConsistentHashingMetadata;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.model.HoodieTableType;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.timeline.HoodieActiveTimeline;
+import org.apache.hudi.common.table.view.FileSystemViewStorageConfig;
+import org.apache.hudi.common.table.view.FileSystemViewStorageType;
+import org.apache.hudi.common.testutils.HoodieTestDataGenerator;
+import org.apache.hudi.common.testutils.HoodieTestUtils;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.config.HoodieClusteringConfig;
+import org.apache.hudi.config.HoodieCompactionConfig;
+import org.apache.hudi.config.HoodieIndexConfig;
+import org.apache.hudi.config.HoodieStorageConfig;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.execution.bulkinsert.BulkInsertSortMode;
+import org.apache.hudi.index.HoodieIndex;
+import org.apache.hudi.index.bucket.HoodieSparkConsistentBucketIndex;
+import org.apache.hudi.keygen.constant.KeyGeneratorOptions;
+import org.apache.hudi.table.HoodieSparkTable;
+import org.apache.hudi.table.HoodieTable;
+import org.apache.hudi.table.action.cluster.ClusteringPlanPartitionFilterMode;
+import org.apache.hudi.testutils.HoodieClientTestHarness;
+import org.apache.hudi.testutils.HoodieMergeOnReadTestUtils;
+import org.apache.hudi.testutils.MetadataMergeWriteStatus;
+
+import org.apache.avro.Schema;
+import org.apache.avro.generic.GenericRecord;
+import org.apache.hadoop.hive.serde2.io.DoubleWritable;
+import org.apache.hadoop.io.ArrayWritable;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.mapred.RecordReader;
+import org.apache.spark.api.java.JavaRDD;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.io.IOException;
+import java.nio.file.Paths;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Random;
+import java.util.stream.Collectors;
+
+import static org.apache.hudi.config.HoodieClusteringConfig.DAYBASED_LOOKBACK_PARTITIONS;
+import static org.apache.hudi.config.HoodieClusteringConfig.PLAN_PARTITION_FILTER_MODE;
+import static org.apache.hudi.config.HoodieClusteringConfig.PLAN_STRATEGY_SKIP_PARTITIONS_FROM_LATEST;
+
+@Tag("functional")
+public class TestSparkConsistentBucketClustering extends HoodieClientTestHarness {
+
+  private final Random random = new Random();
+  private HoodieIndex index;

Review Comment:
   not used anywhere. let's remove it?



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org