[HUDI-3123] review fix

Author: YuweiXiao

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/utils/LazyIterableIterator.java

@@ -45,7 +45,9 @@ public LazyIterableIterator(Iterator<I> in) {
   /**
    * Called once, before any elements are processed.
    */
-  protected abstract void start();
+  protected void start() {
+
+  }
 
   /**
    * Block computation to be overwritten by sub classes.
@@ -55,7 +57,9 @@ public LazyIterableIterator(Iterator<I> in) {
   /**
    * Called once, after all elements are processed.
    */
-  protected abstract void end();
+  protected void end() {
+
+  }
 
   //////////////////
   // iterable implementation

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bucket/BucketIdentifier.java

@@ -31,7 +31,7 @@
 import java.util.stream.Collectors;
 
 public class BucketIdentifier implements Serializable {
-  // compatible with the spark bucket name
+  // Compatible with the spark bucket name
   private static final Pattern BUCKET_NAME = Pattern.compile(".*_(\\d+)(?:\\..*)?$");
 
   public static int getBucketId(HoodieRecord record, String indexKeyFields, int numBuckets) {
@@ -45,7 +45,7 @@ public static int getBucketId(HoodieKey hoodieKey, String indexKeyFields, int nu
   protected static List<String> getHashKeys(HoodieKey hoodieKey, String indexKeyFields) {
     List<String> hashKeys;
     if (!hoodieKey.getRecordKey().contains(":")) {
-      hashKeyFields = Collections.singletonList(hoodieKey.getRecordKey());
+      hashKeys = Collections.singletonList(hoodieKey.getRecordKey());
     } else {
       Map<String, String> recordKeyPairs = Arrays.stream(hoodieKey.getRecordKey().split(","))
           .map(p -> p.split(":"))
@@ -57,7 +57,7 @@ protected static List<String> getHashKeys(HoodieKey hoodieKey, String indexKeyFi
     return hashKeys;
   }
 
-  // only for test
+  // Only for test
   public  static int getBucketId(List<String> hashKeyFields, int numBuckets) {
     return hashKeyFields.hashCode() % numBuckets;
   }

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bucket/ConsistentBucketIdentifier.java

@@ -38,16 +38,18 @@ public class ConsistentBucketIdentifier extends BucketIdentifier {
    */
   private final HoodieConsistentHashingMetadata metadata;
   /**
-   * in-memory structure to speed up rang mapping (hashing value -> hashing node)
+   * In-memory structure to speed up ring mapping (hashing value -> hashing node)
    */
-  private TreeMap<Integer, ConsistentHashingNode> ring;
+  private final TreeMap<Integer, ConsistentHashingNode> ring;
   /**
-   * mapping from fileId -> hashing node
+   * Mapping from fileId -> hashing node
    */
-  private Map<String, ConsistentHashingNode> fileIdToBucket;
+  private final Map<String, ConsistentHashingNode> fileIdToBucket;
 
   public ConsistentBucketIdentifier(HoodieConsistentHashingMetadata metadata) {
     this.metadata = metadata;
+    this.fileIdToBucket = new HashMap<>();
+    this.ring = new TreeMap<>();
     initialize();
   }
 
@@ -82,7 +84,7 @@ protected ConsistentHashingNode getBucket(List<String> hashKeys) {
     for (int i = 0; i < hashKeys.size(); ++i) {
       hashValue = HashID.getXXHash32(hashKeys.get(i), hashValue);
     }
-    return getBucket(hashValue & HoodieConsistentHashingMetadata.MAX_HASH_VALUE);
+    return getBucket(hashValue & HoodieConsistentHashingMetadata.HASH_VALUE_MASK);
   }
 
   protected ConsistentHashingNode getBucket(int hashValue) {
@@ -93,15 +95,13 @@ protected ConsistentHashingNode getBucket(int hashValue) {
   /**
    * Initialize necessary data structure to facilitate bucket identifying.
    * Specifically, we construct:
-   * - a in-memory tree (ring) to speed up range mapping searching.
-   * - a hash table (fileIdToBucket) to allow lookup of bucket using fileId.
+   * - An in-memory tree (ring) to speed up range mapping searching.
+   * - A hash table (fileIdToBucket) to allow lookup of bucket using fileId.
    */
   private void initialize() {
-    this.fileIdToBucket = new HashMap<>();
-    this.ring = new TreeMap<>();
     for (ConsistentHashingNode p : metadata.getNodes()) {
       ring.put(p.getValue(), p);
-      // one bucket has only one file group, so append 0 directly
+      // One bucket has only one file group, so append 0 directly
       fileIdToBucket.put(FSUtils.createNewFileId(p.getFileIdPfx(), 0), p);
     }
   }

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bucket/HoodieBucketIndex.java

@@ -45,15 +45,15 @@ public abstract class HoodieBucketIndex extends HoodieIndex<Object, Object> {
 
   private static final Logger LOG = LogManager.getLogger(HoodieBucketIndex.class);
 
-  protected int numBuckets;
-  protected String indexKeyFields;
+  protected final int numBuckets;
+  protected final String indexKeyFields;
 
   public HoodieBucketIndex(HoodieWriteConfig config) {
     super(config);
 
     this.numBuckets = config.getBucketIndexNumBuckets();
     this.indexKeyFields = config.getBucketIndexHashField();
-    LOG.info("use bucket index, numBuckets = " + numBuckets + ", indexFields: " + indexKeyFields);
+    LOG.info("Use bucket index, numBuckets = " + numBuckets + ", indexFields: " + indexKeyFields);
   }
 
   @Override
@@ -69,31 +69,20 @@ public <R> HoodieData<HoodieRecord<R>> tagLocation(
       HoodieData<HoodieRecord<R>> records, HoodieEngineContext context,
       HoodieTable hoodieTable)
       throws HoodieIndexException {
-    // initialize necessary information before tagging. e.g., hashing metadata
+    // Initialize necessary information before tagging. e.g., hashing metadata
     List<String> partitions = records.map(HoodieRecord::getPartitionPath).distinct().collectAsList();
     LOG.info("Initializing hashing metadata for partitions: " + partitions);
     initialize(hoodieTable, partitions);
 
     return records.mapPartitions(iterator ->
         new LazyIterableIterator<HoodieRecord<R>, HoodieRecord<R>>(iterator) {
-
-          @Override
-          protected void start() {
-
-          }
-
           @Override
           protected HoodieRecord<R> computeNext() {
             // TODO maybe batch the operation to improve performance
             HoodieRecord record = inputItr.next();
             HoodieRecordLocation loc = getBucket(record.getKey(), record.getPartitionPath());
             return HoodieIndexUtils.getTaggedRecord(record, Option.ofNullable(loc));
           }
-
-          @Override
-          protected void end() {
-
-          }
         }
     );
   }
@@ -137,13 +126,15 @@ public int getNumBuckets() {
 
   /**
    * Initialize necessary fields
+   *
    * @param table
    * @param partitions
    */
   protected abstract void initialize(HoodieTable table, List<String> partitions);
 
   /**
    * Get record location given the record key and its partition
+   *
    * @param key
    * @param partitionPath
    * @return

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bucket/HoodieSimpleBucketIndex.java

@@ -20,7 +20,6 @@
 
 import org.apache.hudi.common.model.HoodieKey;
 import org.apache.hudi.common.model.HoodieRecordLocation;
-import org.apache.hudi.common.util.collection.Pair;
 import org.apache.hudi.config.HoodieWriteConfig;
 import org.apache.hudi.exception.HoodieIOException;
 import org.apache.hudi.index.HoodieIndexUtils;
@@ -38,38 +37,38 @@
  */
 public class HoodieSimpleBucketIndex extends HoodieBucketIndex {
 
-  private static final Logger LOG =  LogManager.getLogger(HoodieSimpleBucketIndex.class);
+  private static final Logger LOG = LogManager.getLogger(HoodieSimpleBucketIndex.class);
 
   /**
-   * partitionPath -> bucketId -> fileInfo
+   * Mapping from partitionPath -> bucketId -> fileInfo
    */
-  Map<String, Map<Integer, Pair<String, String>>> partitionPathFileIDList;
+  private Map<String, Map<Integer, HoodieRecordLocation>> partitionPathFileIDList;
 
   public HoodieSimpleBucketIndex(HoodieWriteConfig config) {
     super(config);
   }
 
-  private Map<Integer, Pair<String, String>> loadPartitionBucketIdFileIdMapping(
+  private Map<Integer, HoodieRecordLocation> loadPartitionBucketIdFileIdMapping(
       HoodieTable hoodieTable,
       String partition) {
     // bucketId -> fileIds
-    Map<Integer, Pair<String, String>> fileIDList = new HashMap<>();
+    Map<Integer, HoodieRecordLocation> bucketIdToFileIdMapping = new HashMap<>();
     hoodieTable.getMetaClient().reloadActiveTimeline();
     HoodieIndexUtils
         .getLatestBaseFilesForPartition(partition, hoodieTable)
         .forEach(file -> {
           String fileId = file.getFileId();
           String commitTime = file.getCommitTime();
           int bucketId = BucketIdentifier.bucketIdFromFileId(fileId);
-          if (!fileIDList.containsKey(bucketId)) {
-            fileIDList.put(bucketId, Pair.of(fileId, commitTime));
+          if (!bucketIdToFileIdMapping.containsKey(bucketId)) {
+            bucketIdToFileIdMapping.put(bucketId, new HoodieRecordLocation(commitTime, fileId));
           } else {
-            // check if bucket data is valid
+            // Check if bucket data is valid
             throw new HoodieIOException("Find multiple files at partition path="
                 + partition + " belongs to the same bucket id = " + bucketId);
           }
         });
-    return fileIDList;
+    return bucketIdToFileIdMapping;
   }
 
   @Override
@@ -86,10 +85,7 @@ protected void initialize(HoodieTable table, List<String> partitions) {
   @Override
   protected HoodieRecordLocation getBucket(HoodieKey key, String partitionPath) {
     int bucketId = BucketIdentifier.getBucketId(key, config.getBucketIndexHashField(), numBuckets);
-    if (partitionPathFileIDList.get(partitionPath).containsKey(bucketId)) {
-      Pair<String, String> fileInfo = partitionPathFileIDList.get(partitionPath).get(bucketId);
-      return new HoodieRecordLocation(fileInfo.getRight(), fileInfo.getLeft());
-    }
-    return null;
+    Map<Integer, HoodieRecordLocation> bucketIdToFileIdMapping = partitionPathFileIDList.get(partitionPath);
+    return bucketIdToFileIdMapping.getOrDefault(bucketId, null);
   }
 }

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/storage/HoodieConsistentBucketLayout.java

@@ -19,28 +19,26 @@
 package org.apache.hudi.table.storage;
 
 import org.apache.hudi.common.model.WriteOperationType;
+import org.apache.hudi.common.util.CollectionUtils;
 import org.apache.hudi.common.util.Option;
 import org.apache.hudi.config.HoodieWriteConfig;
 
-import java.util.HashSet;
 import java.util.Set;
 
 /**
  * Storage layout when using consistent hashing bucket index.
  */
 public class HoodieConsistentBucketLayout extends HoodieStorageLayout {
-  public static final Set<WriteOperationType> SUPPORTED_OPERATIONS = new HashSet<WriteOperationType>() {
-    {
-      add(WriteOperationType.INSERT);
-      add(WriteOperationType.INSERT_PREPPED);
-      add(WriteOperationType.UPSERT);
-      add(WriteOperationType.UPSERT_PREPPED);
-      add(WriteOperationType.INSERT_OVERWRITE);
-      add(WriteOperationType.DELETE);
-      add(WriteOperationType.COMPACT);
-      add(WriteOperationType.DELETE_PARTITION);
-    }
-  };
+  public static final Set<WriteOperationType> SUPPORTED_OPERATIONS = CollectionUtils.createImmutableSet(
+      WriteOperationType.INSERT,
+      WriteOperationType.INSERT_PREPPED,
+      WriteOperationType.UPSERT,
+      WriteOperationType.UPSERT_PREPPED,
+      WriteOperationType.INSERT_OVERWRITE,
+      WriteOperationType.DELETE,
+      WriteOperationType.COMPACT,
+      WriteOperationType.DELETE_PARTITION
+  );
 
   public HoodieConsistentBucketLayout(HoodieWriteConfig config) {
     super(config);
@@ -56,6 +54,7 @@ public boolean determinesNumFileGroups() {
 
   /**
    * Consistent hashing will tag all incoming records, so we could go ahead reusing an existing Partitioner
+   *
    * @return
    */
   @Override

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/storage/HoodieSimpleBucketLayout.java

@@ -19,29 +19,28 @@
 package org.apache.hudi.table.storage;
 
 import org.apache.hudi.common.model.WriteOperationType;
+import org.apache.hudi.common.util.CollectionUtils;
 import org.apache.hudi.common.util.Option;
 import org.apache.hudi.config.HoodieLayoutConfig;
 import org.apache.hudi.config.HoodieWriteConfig;
 
-import java.util.HashSet;
 import java.util.Set;
 
 /**
  * Storage layout when using bucket index. Data distribution and files organization are in a specific way.
  */
 public class HoodieSimpleBucketLayout extends HoodieStorageLayout {
 
-  public static final Set<WriteOperationType> SUPPORTED_OPERATIONS = new HashSet<WriteOperationType>() {{
-      add(WriteOperationType.INSERT);
-      add(WriteOperationType.INSERT_PREPPED);
-      add(WriteOperationType.UPSERT);
-      add(WriteOperationType.UPSERT_PREPPED);
-      add(WriteOperationType.INSERT_OVERWRITE);
-      add(WriteOperationType.DELETE);
-      add(WriteOperationType.COMPACT);
-      add(WriteOperationType.DELETE_PARTITION);
-    }
-  };
+  public static final Set<WriteOperationType> SUPPORTED_OPERATIONS = CollectionUtils.createImmutableSet(
+      WriteOperationType.INSERT,
+      WriteOperationType.INSERT_PREPPED,
+      WriteOperationType.UPSERT,
+      WriteOperationType.UPSERT_PREPPED,
+      WriteOperationType.INSERT_OVERWRITE,
+      WriteOperationType.DELETE,
+      WriteOperationType.COMPACT,
+      WriteOperationType.DELETE_PARTITION
+  );
 
   public HoodieSimpleBucketLayout(HoodieWriteConfig config) {
     super(config);