[HUDI-9622] Add implementation of MergeHandle backed by the HoodieFileGroupReader (#13699)

The goal of this PR is to ensure consistent behavior while reading and writing data across our Merge-on-Read and Copy-on-Write tables by leveraging the existing HoodieFileGroupReader to manage the merging of records. The FileGroupReaderBasedMergeHandle that is currently used for compaction is updated to allow merging with an incoming stream of records.

Summary of changes:

- FileGroupReaderBasedMergeHandle.java is updated to allow incoming records in the form of an iterator of records directly instead of reading changes exclusively from log files. New callbacks are added to support creating the required outputs for updates to Record Level and Secondary indexes.
- The merge handle is also updated to account for preserving the metadata of records that are not updated while also generating the metadata for updated records. This does not impact the compaction workflow which will preserve the metadata of the records.
- The FileGroupReaderBasedMergeHandle is set as the default merge handle
- New test cases are added for RLI including a test where records move between partitions and deletes are sent to partitions that do not contain the original record
- The delete record ordering value is now converted to the engine specific type so there are no issues when performing comparisons

Differences between FileGroupReaderBasedMergeHandle and HoodieWriteMergeHandle
- Currently the HoodieWriteMergeHandle can handle applying a single update to multiple records with the same key. This functionality does not exist in the FileGroupReaderBasedMergeHandle
- The FileGroupReaderBasedMergeHandle does not support the shouldFlush functionality in the HoodieRecordMerger

---------

Co-authored-by: Sivabalan Narayanan <[email protected]>
Co-authored-by: Lokesh Jain <[email protected]>
Co-authored-by: Lokesh Jain <[email protected]>
Co-authored-by: danny0405 <[email protected]>

Author: 5 people

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/WriteStatus.java

@@ -73,7 +73,7 @@ public class WriteStatus implements Serializable {
   private long totalErrorRecords = 0;
 
   private final double failureFraction;
-  private final boolean trackSuccessRecords;
+  private boolean trackSuccessRecords;
   private final transient Random random;
   private IndexStats indexStats = new IndexStats();
 
@@ -109,6 +109,17 @@ public void markSuccess(HoodieRecord record, Option<Map<String, String>> optiona
     updateStatsForSuccess(optionalRecordMetadata);
   }
 
+  /**
+   * Allows the writer to manually add record delegates to the index stats.
+   */
+  public void manuallyTrackSuccess() {
+    this.trackSuccessRecords = false;
+  }
+
+  public void addRecordDelegate(HoodieRecordDelegate recordDelegate) {
+    indexStats.addHoodieRecordDelegate(recordDelegate);
+  }
+
   /**
    * Used by native write handles like HoodieRowCreateHandle and HoodieRowDataCreateHandle.
    *

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

@@ -73,7 +73,6 @@
 import org.apache.hudi.index.HoodieIndex;
 import org.apache.hudi.io.FileGroupReaderBasedMergeHandle;
 import org.apache.hudi.io.HoodieConcatHandle;
-import org.apache.hudi.io.HoodieWriteMergeHandle;
 import org.apache.hudi.keygen.SimpleAvroKeyGenerator;
 import org.apache.hudi.keygen.constant.KeyGeneratorOptions;
 import org.apache.hudi.keygen.constant.KeyGeneratorType;
@@ -858,7 +857,7 @@ public class HoodieWriteConfig extends HoodieConfig {
 
   public static final ConfigProperty<String> MERGE_HANDLE_CLASS_NAME = ConfigProperty
       .key("hoodie.write.merge.handle.class")
-      .defaultValue(HoodieWriteMergeHandle.class.getName())
+      .defaultValue(FileGroupReaderBasedMergeHandle.class.getName())
       .markAdvanced()
       .sinceVersion("1.1.0")
       .withDocumentation("The merge handle class that implements interface{@link HoodieMergeHandle} to merge the records "

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

@@ -430,16 +430,11 @@ private static <R> Option<HoodieRecord<R>> mergeIncomingWithExistingRecordWithEx
       //the record was deleted
       return Option.empty();
     }
-    if (mergeResult.getRecord() == null) {
-      // SENTINEL case: the record did not match and merge case and should not be modified
+    if (mergeResult.getRecord() == null || mergeResult == existingBufferedRecord) {
+      // SENTINEL case: the record did not match the merge case and should not be modified or there is no update to the record
       return Option.of((HoodieRecord<R>) new HoodieAvroIndexedRecord(HoodieRecord.SENTINEL));
     }
 
-    if (mergeResult.getRecord().equals(HoodieRecord.SENTINEL)) {
-      //the record did not match and merge case and should not be modified
-      return Option.of(existingRecordContext.constructHoodieRecord(mergeResult, incoming.getPartitionPath()));
-    }
-
     //record is inserted or updated
     String partitionPath = inferPartitionPath(incoming, existing, writeSchemaWithMetaFields, keyGenerator, existingRecordContext, mergeResult);
     HoodieRecord<R> result = existingRecordContext.constructHoodieRecord(mergeResult, partitionPath);
@@ -526,14 +521,14 @@ public static <R> HoodieData<HoodieRecord<R>> mergeForPartitionUpdatesIfNeeded(
         .distinct(updatedConfig.getGlobalIndexReconcileParallelism());
     // define the buffered record merger.
     ReaderContextFactory<R> readerContextFactory = (ReaderContextFactory<R>) hoodieTable.getContext()
-        .getReaderContextFactoryForWrite(hoodieTable.getMetaClient(), config.getRecordMerger().getRecordType(), config.getProps());
+        .getReaderContextFactoryForWrite(hoodieTable.getMetaClient(), config.getRecordMerger().getRecordType(), config.getProps(), true);
     HoodieReaderContext<R> readerContext = readerContextFactory.getContext();
     RecordContext<R> incomingRecordContext = readerContext.getRecordContext();
     readerContext.initRecordMergerForIngestion(config.getProps());
     // Create a reader context for the existing records. In the case of merge-into commands, the incoming records
     // can be using an expression payload so here we rely on the table's configured payload class if it is required.
     ReaderContextFactory<R> readerContextFactoryForExistingRecords = (ReaderContextFactory<R>) hoodieTable.getContext()
-        .getReaderContextFactoryForWrite(hoodieTable.getMetaClient(), config.getRecordMerger().getRecordType(), hoodieTable.getMetaClient().getTableConfig().getProps());
+        .getReaderContextFactoryForWrite(hoodieTable.getMetaClient(), config.getRecordMerger().getRecordType(), hoodieTable.getMetaClient().getTableConfig().getProps(), true);
     RecordContext<R> existingRecordContext = readerContextFactoryForExistingRecords.getContext().getRecordContext();
     // merged existing records with current locations being set
     SerializableSchema writerSchema = new SerializableSchema(hoodieTable.getConfig().getWriteSchema());
@@ -551,7 +546,7 @@ public static <R> HoodieData<HoodieRecord<R>> mergeForPartitionUpdatesIfNeeded(
         false,
         readerContext.getRecordMerger(),
         orderingFieldNames,
-        writerSchema.get(),
+        isExpressionPayload ? writerSchema.get() : writerSchemaWithMetaFields.get(),
         Option.ofNullable(Pair.of(hoodieTable.getMetaClient().getTableConfig().getPayloadClass(), hoodieTable.getConfig().getPayloadClass())),
         hoodieTable.getConfig().getProps(),
         hoodieTable.getMetaClient().getTableConfig().getPartialUpdateMode());