[HUDI-9622] Add implementation of MergeHandle backed by the HoodieFileGroupReader

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());