[HUDI-2798] Fix flink query operation fields (#4041)

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieMergeHandle.java

@@ -250,11 +250,17 @@ public class HoodieMergeHandle<T extends HoodieRecordPayload, I, K, O> extends H
         + ((ExternalSpillableMap) keyToNewRecords).getSizeOfFileOnDiskInBytes());
   }
 
-  private boolean writeUpdateRecord(HoodieRecord<T> hoodieRecord, Option<IndexedRecord> indexedRecord) {
+  private boolean writeUpdateRecord(HoodieRecord<T> hoodieRecord, GenericRecord oldRecord, Option<IndexedRecord> indexedRecord) {
+    boolean isDelete = false;
     if (indexedRecord.isPresent()) {
       updatedRecordsWritten++;
+      GenericRecord record = (GenericRecord) indexedRecord.get();
+      if (oldRecord != record) {
+        // the incoming record is chosen
+        isDelete = HoodieOperation.isDelete(hoodieRecord.getOperation());
       }
-    return writeRecord(hoodieRecord, indexedRecord);
+    }
+    return writeRecord(hoodieRecord, indexedRecord, isDelete);
   }
 
   protected void writeInsertRecord(HoodieRecord<T> hoodieRecord) throws IOException {
@@ -264,12 +270,16 @@ public class HoodieMergeHandle<T extends HoodieRecordPayload, I, K, O> extends H
     if (insertRecord.isPresent() && insertRecord.get().equals(IGNORE_RECORD)) {
       return;
     }
-    if (writeRecord(hoodieRecord, insertRecord)) {
+    if (writeRecord(hoodieRecord, insertRecord, HoodieOperation.isDelete(hoodieRecord.getOperation()))) {
       insertRecordsWritten++;
     }
   }
 
   protected boolean writeRecord(HoodieRecord<T> hoodieRecord, Option<IndexedRecord> indexedRecord) {
+    return writeRecord(hoodieRecord, indexedRecord, false);
+  }
+
+  protected boolean writeRecord(HoodieRecord<T> hoodieRecord, Option<IndexedRecord> indexedRecord, boolean isDelete) {
     Option recordMetadata = hoodieRecord.getData().getMetadata();
     if (!partitionPath.equals(hoodieRecord.getPartitionPath())) {
       HoodieUpsertException failureEx = new HoodieUpsertException("mismatched partition path, record partition: "
@@ -277,11 +287,8 @@ public class HoodieMergeHandle<T extends HoodieRecordPayload, I, K, O> extends H
       writeStatus.markFailure(hoodieRecord, failureEx, recordMetadata);
       return false;
     }
-    if (HoodieOperation.isDelete(hoodieRecord.getOperation())) {
-      indexedRecord = Option.empty();
-    }
     try {
-      if (indexedRecord.isPresent()) {
+      if (indexedRecord.isPresent() && !isDelete) {
         // Convert GenericRecord to GenericRecord with hoodie commit metadata in schema
         IndexedRecord recordWithMetadataInSchema = rewriteRecord((GenericRecord) indexedRecord.get());
         fileWriter.writeAvroWithMetadata(recordWithMetadataInSchema, hoodieRecord);
@@ -321,7 +328,7 @@ public class HoodieMergeHandle<T extends HoodieRecordPayload, I, K, O> extends H
         if (combinedAvroRecord.isPresent() && combinedAvroRecord.get().equals(IGNORE_RECORD)) {
           // If it is an IGNORE_RECORD, just copy the old record, and do not update the new record.
           copyOldRecord = true;
-        } else if (writeUpdateRecord(hoodieRecord, combinedAvroRecord)) {
+        } else if (writeUpdateRecord(hoodieRecord, oldRecord, combinedAvroRecord)) {
           /*
            * ONLY WHEN 1) we have an update for this key AND 2) We are able to successfully
            * write the the combined new

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/compact/HoodieCompactor.java

@@ -179,6 +179,7 @@ public abstract class HoodieCompactor<T extends HoodieRecordPayload, I, K, O> im
         .withSpillableMapBasePath(config.getSpillableMapBasePath())
         .withDiskMapType(config.getCommonConfig().getSpillableDiskMapType())
         .withBitCaskDiskMapCompressionEnabled(config.getCommonConfig().isBitCaskDiskMapCompressionEnabled())
+        .withOperationField(config.allowOperationMetadataField())
         .build();
     if (!scanner.iterator().hasNext()) {
       scanner.close();

hudi-flink/src/main/java/org/apache/hudi/source/IncrementalInputSplits.java

@@ -44,6 +44,7 @@ import javax.annotation.Nullable;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Objects;
 import java.util.Set;
@@ -156,6 +157,24 @@ public class IncrementalInputSplits implements Serializable {
     }
 
     String tableName = conf.getString(FlinkOptions.TABLE_NAME);
+
+    Set<String> writePartitions;
+    final FileStatus[] fileStatuses;
+
+    if (instantRange == null) {
+      // reading from the earliest, scans the partitions and files directly.
+      FileIndex fileIndex = FileIndex.instance(new org.apache.hadoop.fs.Path(path.toUri()), conf);
+      if (this.requiredPartitions != null) {
+        // apply partition push down
+        fileIndex.setPartitionPaths(this.requiredPartitions);
+      }
+      writePartitions = new HashSet<>(fileIndex.getOrBuildPartitionPaths());
+      if (writePartitions.size() == 0) {
+        LOG.warn("No partitions found for reading in user provided path.");
+        return Result.EMPTY;
+      }
+      fileStatuses = fileIndex.getFilesInPartitions();
+    } else {
       List<HoodieCommitMetadata> activeMetadataList = instants.stream()
           .map(instant -> WriteProfiles.getCommitMetadata(tableName, path, instant, commitTimeline)).collect(Collectors.toList());
       List<HoodieCommitMetadata> archivedMetadataList = getArchivedMetadata(metaClient, instantRange, commitTimeline, tableName);
@@ -171,13 +190,19 @@ public class IncrementalInputSplits implements Serializable {
           ? mergeList(archivedMetadataList, activeMetadataList)
           : activeMetadataList;
 
-    Set<String> writePartitions = HoodieInputFormatUtils.getWritePartitionPaths(metadataList);
+      writePartitions = HoodieInputFormatUtils.getWritePartitionPaths(metadataList);
       // apply partition push down
       if (this.requiredPartitions != null) {
         writePartitions = writePartitions.stream()
             .filter(this.requiredPartitions::contains).collect(Collectors.toSet());
       }
-    FileStatus[] fileStatuses = WriteProfiles.getWritePathsOfInstants(path, hadoopConf, metadataList, metaClient.getTableType());
+      if (writePartitions.size() == 0) {
+        LOG.warn("No partitions found for reading in user provided path.");
+        return Result.EMPTY;
+      }
+      fileStatuses = WriteProfiles.getWritePathsOfInstants(path, hadoopConf, metadataList, metaClient.getTableType());
+    }
+
     if (fileStatuses.length == 0) {
       LOG.warn("No files found for reading in user provided path.");
       return Result.EMPTY;

hudi-flink/src/main/java/org/apache/hudi/table/format/mor/MergeOnReadInputFormat.java

@@ -683,13 +683,18 @@ public class MergeOnReadInputFormat
             // deleted
             continue;
           } else {
+            final RowKind rowKind = FormatUtils.getRowKindSafely(mergedAvroRecord.get(), this.operationPos);
+            if (!emitDelete && rowKind == RowKind.DELETE) {
+              // deleted
+              continue;
+            }
             GenericRecord avroRecord = buildAvroRecordBySchema(
                 mergedAvroRecord.get(),
                 requiredSchema,
                 requiredPos,
                 recordBuilder);
             this.currentRecord = (RowData) avroToRowDataConverter.convert(avroRecord);
-            FormatUtils.setRowKind(this.currentRecord, mergedAvroRecord.get(), this.operationPos);
+            this.currentRecord.setRowKind(rowKind);
             return false;
           }
         }
@@ -746,9 +751,6 @@ public class MergeOnReadInputFormat
         RowData curRow,
         String curKey) throws IOException {
       final HoodieRecord<?> record = scanner.getRecords().get(curKey);
-      if (!emitDelete && HoodieOperation.isDelete(record.getOperation())) {
-        return Option.empty();
-      }
       GenericRecord historyAvroRecord = (GenericRecord) rowDataToAvroConverter.convert(tableSchema, curRow);
       return record.getData().combineAndGetUpdateValue(historyAvroRecord, tableSchema);
     }

hudi-flink/src/test/java/org/apache/hudi/table/format/TestInputFormat.java

@@ -39,6 +39,7 @@ import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.io.TempDir;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.EnumSource;
+import org.junit.jupiter.params.provider.ValueSource;
 
 import java.io.File;
 import java.io.IOException;
@@ -221,8 +222,9 @@ public class TestInputFormat {
     assertThat(actual2, is(expected2));
   }
 
-  @Test
+  @ParameterizedTest
-  void testReadBaseAndLogFilesWithDisorderUpdateDelete() throws Exception {
+  @ValueSource(booleans = {true, false})
+  void testReadBaseAndLogFilesWithDisorderUpdateDelete(boolean compact) throws Exception {
     Map<String, String> options = new HashMap<>();
     options.put(FlinkOptions.CHANGELOG_ENABLED.key(), "true");
     beforeEach(HoodieTableType.MERGE_ON_READ, options);
@@ -233,7 +235,7 @@ public class TestInputFormat {
     TestData.writeData(TestData.DATA_SET_SINGLE_INSERT, conf);
 
     // write another commit using logs and read again.
-    conf.setBoolean(FlinkOptions.COMPACTION_ASYNC_ENABLED, false);
+    conf.setBoolean(FlinkOptions.COMPACTION_ASYNC_ENABLED, compact);
     TestData.writeData(TestData.DATA_SET_DISORDER_UPDATE_DELETE, conf);
 
     InputFormat<RowData, ?> inputFormat = this.tableSource.getInputFormat();
@@ -242,9 +244,11 @@ public class TestInputFormat {
     // when isEmitDelete is false.
     List<RowData> result1 = readData(inputFormat);
 
+    final String rowKind = compact ? "I" : "U";
+    final String expected = "[+" + rowKind + "[id1, Danny, 22, 1970-01-01T00:00:00.004, par1]]";
+
     final String actual1 = TestData.rowDataToString(result1);
-    final String expected1 = "[+U[id1, Danny, 22, 1970-01-01T00:00:00.004, par1]]";
+    assertThat(actual1, is(expected));
-    assertThat(actual1, is(expected1));
 
     // refresh the input format and set isEmitDelete to true.
     this.tableSource.reset();
@@ -254,8 +258,7 @@ public class TestInputFormat {
     List<RowData> result2 = readData(inputFormat);
 
     final String actual2 = TestData.rowDataToString(result2);
-    final String expected2 = "[+U[id1, Danny, 22, 1970-01-01T00:00:00.004, par1]]";
+    assertThat(actual2, is(expected));
-    assertThat(actual2, is(expected2));
   }
 
   @Test