[HUDI-9206] Support reading inflight instants with HoodieLogRecordReader (apache#13010)

(cherry picked from commit 1f52b4e)

Author: lokeshj1703

hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/clustering/run/strategy/MultipleSparkJobExecutionStrategy.java

@@ -473,20 +473,11 @@ public Iterator<InternalRow> call(ClusteringOperation clusteringOperation) throw
         Configuration conf = broadcastManager.retrieveStorageConfig().get();
 
         // instantiate FG reader
-        HoodieFileGroupReader<T> fileGroupReader = new HoodieFileGroupReader<>(
-            readerContextOpt.get(),
+        HoodieFileGroupReader<T> fileGroupReader = new HoodieFileGroupReader<>(readerContextOpt.get(),
             getHoodieTable().getMetaClient().getStorage().newInstance(new StoragePath(basePath), new HadoopStorageConfiguration(conf)),
-            basePath,
-            instantTime,
-            fileSlice,
-            readerSchema,
-            readerSchema,
-            internalSchemaOption,
-            getHoodieTable().getMetaClient(),
-            getHoodieTable().getMetaClient().getTableConfig().getProps(),
-            0,
-            Long.MAX_VALUE,
-            usePosition);
+            basePath, instantTime, fileSlice, readerSchema, readerSchema, internalSchemaOption,
+            getHoodieTable().getMetaClient(), getHoodieTable().getMetaClient().getTableConfig().getProps(),
+            0, Long.MAX_VALUE, usePosition, false);
         fileGroupReader.initRecordIterators();
         // read records from the FG reader
         HoodieFileGroupReader.HoodieFileGroupReaderIterator<InternalRow> recordIterator

hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/HoodieSparkFileGroupReaderBasedMergeHandle.java

@@ -185,20 +185,11 @@ public void write() {
     hoodieTable.getMetaClient().getTableConfig().getProps().forEach(props::putIfAbsent);
     config.getProps().forEach(props::putIfAbsent);
     // Initializes file group reader
-    try (HoodieFileGroupReader<T> fileGroupReader = new HoodieFileGroupReader<>(
-        readerContext,
+    try (HoodieFileGroupReader<T> fileGroupReader = new HoodieFileGroupReader<>(readerContext,
         storage.newInstance(hoodieTable.getMetaClient().getBasePath(), new HadoopStorageConfiguration(conf)),
-        hoodieTable.getMetaClient().getBasePath().toString(),
-        instantTime,
-        fileSlice,
-        writeSchemaWithMetaFields,
-        writeSchemaWithMetaFields,
-        internalSchemaOption,
-        hoodieTable.getMetaClient(),
-        props,
-        0,
-        Long.MAX_VALUE,
-        usePosition)) {
+        hoodieTable.getMetaClient().getBasePath().toString(), instantTime, fileSlice,
+        writeSchemaWithMetaFields, writeSchemaWithMetaFields, internalSchemaOption,
+        hoodieTable.getMetaClient(), props, 0, Long.MAX_VALUE, usePosition, false)) {
       fileGroupReader.initRecordIterators();
       // Reads the records from the file slice
       try (HoodieFileGroupReaderIterator<InternalRow> recordIterator

hudi-common/src/main/java/org/apache/hudi/common/table/log/BaseHoodieLogRecordReader.java

@@ -136,16 +136,14 @@ public abstract class BaseHoodieLogRecordReader<T> {
   // Use scanV2 method.
   private final boolean enableOptimizedLogBlocksScan;
   protected FileGroupRecordBuffer<T> recordBuffer;
+  // Allows to consider inflight instants while merging log records
+  protected boolean allowInflightInstants;
 
-  protected BaseHoodieLogRecordReader(HoodieReaderContext readerContext,
-                                      HoodieStorage storage,
-                                      List<String> logFilePaths,
+  protected BaseHoodieLogRecordReader(HoodieReaderContext readerContext, HoodieStorage storage, List<String> logFilePaths,
                                       boolean reverseReader, int bufferSize, Option<InstantRange> instantRange,
-                                      boolean withOperationField, boolean forceFullScan,
-                                      Option<String> partitionNameOverride,
-                                      Option<String> keyFieldOverride,
-                                      boolean enableOptimizedLogBlocksScan,
-                                      FileGroupRecordBuffer<T> recordBuffer) {
+                                      boolean withOperationField, boolean forceFullScan, Option<String> partitionNameOverride,
+                                      Option<String> keyFieldOverride, boolean enableOptimizedLogBlocksScan, FileGroupRecordBuffer<T> recordBuffer,
+                                      boolean allowInflightInstants) {
     this.readerContext = readerContext;
     this.readerSchema = readerContext.getSchemaHandler().getRequiredSchema();
     this.latestInstantTime = readerContext.getLatestCommitTime();
@@ -196,6 +194,8 @@ protected BaseHoodieLogRecordReader(HoodieReaderContext readerContext,
 
     this.partitionNameOverrideOpt = partitionNameOverride;
     this.recordBuffer = recordBuffer;
+    // When the allowInflightInstants flag is enabled, records written by inflight instants are also read
+    this.allowInflightInstants = allowInflightInstants;
   }
 
   /**
@@ -256,8 +256,8 @@ private void scanInternalV1(Option<KeySpec> keySpecOpt) {
             // Skip processing a data or delete block with the instant time greater than the latest instant time used by this log record reader
             continue;
           }
-          if (!completedInstantsTimeline.containsOrBeforeTimelineStarts(instantTime)
-              || inflightInstantsTimeline.containsInstant(instantTime)) {
+          if (!allowInflightInstants
+              && (inflightInstantsTimeline.containsInstant(instantTime) || !completedInstantsTimeline.containsOrBeforeTimelineStarts(instantTime))) {
             // hit an uncommitted block possibly from a failed write, move to the next one and skip processing this one
             continue;
           }
@@ -589,8 +589,8 @@ && compareTimestamps(logBlock.getLogBlockHeader().get(INSTANT_TIME), GREATER_THA
           continue;
         }
         if (logBlock.getBlockType() != COMMAND_BLOCK) {
-          if (!completedInstantsTimeline.containsOrBeforeTimelineStarts(instantTime)
-              || inflightInstantsTimeline.containsInstant(instantTime)) {
+          if (!allowInflightInstants
+              && (inflightInstantsTimeline.containsInstant(instantTime)) || !completedInstantsTimeline.containsOrBeforeTimelineStarts(instantTime)) {
             // hit an uncommitted block possibly from a failed write, move to the next one and skip processing this one
             continue;
           }

hudi-common/src/main/java/org/apache/hudi/common/table/log/HoodieMergedLogRecordReader.java

@@ -63,16 +63,12 @@ public class HoodieMergedLogRecordReader<T> extends BaseHoodieLogRecordReader<T>
   private long totalTimeTakenToReadAndMergeBlocks;
 
   @SuppressWarnings("unchecked")
-  private HoodieMergedLogRecordReader(HoodieReaderContext<T> readerContext,
-                                      HoodieStorage storage, List<String> logFilePaths, boolean reverseReader,
-                                      int bufferSize, Option<InstantRange> instantRange,
-                                      boolean withOperationField, boolean forceFullScan,
-                                      Option<String> partitionName,
-                                      Option<String> keyFieldOverride,
-                                      boolean enableOptimizedLogBlocksScan,
-                                      FileGroupRecordBuffer<T> recordBuffer) {
+  private HoodieMergedLogRecordReader(HoodieReaderContext<T> readerContext, HoodieStorage storage, List<String> logFilePaths, boolean reverseReader,
+                                      int bufferSize, Option<InstantRange> instantRange, boolean withOperationField, boolean forceFullScan,
+                                      Option<String> partitionName, Option<String> keyFieldOverride, boolean enableOptimizedLogBlocksScan,
+                                      FileGroupRecordBuffer<T> recordBuffer, boolean allowInflightInstants) {
     super(readerContext, storage, logFilePaths, reverseReader, bufferSize, instantRange, withOperationField,
-        forceFullScan, partitionName, keyFieldOverride, enableOptimizedLogBlocksScan, recordBuffer);
+        forceFullScan, partitionName, keyFieldOverride, enableOptimizedLogBlocksScan, recordBuffer, allowInflightInstants);
     this.scannedPrefixes = new HashSet<>();
 
     if (forceFullScan) {
@@ -220,6 +216,7 @@ public static class Builder<T> extends BaseHoodieLogRecordReader.Builder<T> {
     private boolean enableOptimizedLogBlocksScan = false;
 
     private FileGroupRecordBuffer<T> recordBuffer;
+    private boolean allowInflightInstants = false;
 
     @Override
     public Builder<T> withHoodieReaderContext(HoodieReaderContext<T> readerContext) {
@@ -292,6 +289,11 @@ public Builder<T> withRecordBuffer(FileGroupRecordBuffer<T> recordBuffer) {
       return this;
     }
 
+    public Builder<T> withAllowInflightInstants(boolean allowInflightInstants) {
+      this.allowInflightInstants = allowInflightInstants;
+      return this;
+    }
+
     @Override
     public HoodieMergedLogRecordReader<T> build() {
       ValidationUtils.checkArgument(recordBuffer != null, "Record Buffer is null in Merged Log Record Reader");
@@ -307,7 +309,8 @@ public HoodieMergedLogRecordReader<T> build() {
           withOperationField, forceFullScan,
           Option.ofNullable(partitionName),
           Option.ofNullable(keyFieldOverride),
-          enableOptimizedLogBlocksScan, recordBuffer);
+          enableOptimizedLogBlocksScan, recordBuffer,
+          allowInflightInstants);
     }
   }
 }

hudi-common/src/main/java/org/apache/hudi/common/table/read/HoodieFileGroupReader.java

@@ -82,20 +82,17 @@ public final class HoodieFileGroupReader<T> implements Closeable {
   private ClosableIterator<T> baseFileIterator;
   private final Option<UnaryOperator<T>> outputConverter;
   private final HoodieReadStats readStats;
+  // Allows to consider inflight instants while merging log records using HoodieMergedLogRecordReader
+  // The inflight instants need to be considered while updating RLI records. RLI needs to fetch the revived
+  // and deleted keys from the log files written as part of active data commit. During the RLI update,
+  // the allowInflightInstants flag would need to be set to true. This would ensure the HoodieMergedLogRecordReader
+  // considers the log records which are inflight.
+  private boolean allowInflightInstants;
 
-  public HoodieFileGroupReader(HoodieReaderContext<T> readerContext,
-                               HoodieStorage storage,
-                               String tablePath,
-                               String latestCommitTime,
-                               FileSlice fileSlice,
-                               Schema dataSchema,
-                               Schema requestedSchema,
-                               Option<InternalSchema> internalSchemaOpt,
-                               HoodieTableMetaClient hoodieTableMetaClient,
-                               TypedProperties props,
-                               long start,
-                               long length,
-                               boolean shouldUseRecordPosition) {
+  public HoodieFileGroupReader(HoodieReaderContext<T> readerContext, HoodieStorage storage, String tablePath,
+                               String latestCommitTime, FileSlice fileSlice, Schema dataSchema, Schema requestedSchema,
+                               Option<InternalSchema> internalSchemaOpt, HoodieTableMetaClient hoodieTableMetaClient, TypedProperties props,
+                               long start, long length, boolean shouldUseRecordPosition, boolean allowInflightInstants) {
     this.readerContext = readerContext;
     this.storage = storage;
     this.hoodieBaseFileOption = fileSlice.getBaseFile();
@@ -134,6 +131,7 @@ public HoodieFileGroupReader(HoodieReaderContext<T> readerContext,
     this.recordBuffer = getRecordBuffer(readerContext, hoodieTableMetaClient,
         recordMergeMode, props, hoodieBaseFileOption, this.logFiles.isEmpty(),
         isSkipMerge, shouldUseRecordPosition, readStats);
+    this.allowInflightInstants = allowInflightInstants;
   }
 
   /**
@@ -290,6 +288,7 @@ private void scanLogFiles() {
         .withPartition(getRelativePartitionPath(
             new StoragePath(path), logFiles.get(0).getPath().getParent()))
         .withRecordBuffer(recordBuffer)
+        .withAllowInflightInstants(allowInflightInstants)
         .build()) {
       readStats.setTotalLogReadTimeMs(logRecordReader.getTotalTimeTakenToReadAndMergeBlocks());
       readStats.setTotalUpdatedRecordsCompacted(logRecordReader.getNumMergedRecordsInLog());

hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadataUtil.java

@@ -997,6 +997,7 @@ private static Map<String, HoodieRecord> getLogRecords(List<String> logFilePaths
           Collections.emptyList(),
           datasetMetaClient.getTableConfig().getRecordMergeStrategyId());
 
+      // CRITICAL: Ensure allowInflightInstants is set to true while replacing the scanner with *LogRecordReader or HoodieFileGroupReader
       HoodieMergedLogRecordScanner mergedLogRecordScanner = HoodieMergedLogRecordScanner.newBuilder()
           .withStorage(datasetMetaClient.getStorage())
           .withBasePath(datasetMetaClient.getBasePath())

hudi-common/src/test/java/org/apache/hudi/common/table/read/TestHoodieFileGroupReaderBase.java

@@ -329,6 +329,7 @@ private List<T> readRecordsFromFileGroup(StorageConfiguration<?> storageConf,
           props,
           1,
           fileSlice.getTotalFileSize(),
+          false,
           false));
     }
     HoodieFileGroupReader<T> fileGroupReader = new HoodieFileGroupReader<>(
@@ -344,6 +345,7 @@ private List<T> readRecordsFromFileGroup(StorageConfiguration<?> storageConf,
         props,
         0,
         fileSlice.getTotalFileSize(),
+        false,
         false);
     fileGroupReader.initRecordIterators();
     while (fileGroupReader.hasNext()) {

hudi-common/src/test/java/org/apache/hudi/common/testutils/reader/HoodieFileGroupReaderTestUtils.java

@@ -45,8 +45,8 @@ public static HoodieFileGroupReader<IndexedRecord> createFileGroupReader(
       TypedProperties properties,
       HoodieStorage storage,
       HoodieReaderContext<IndexedRecord> readerContext,
-      HoodieTableMetaClient metaClient
-  ) {
+      HoodieTableMetaClient metaClient,
+      boolean allowInflightCommits) {
     assert (fileSliceOpt.isPresent());
     return new HoodieFileGroupReaderBuilder()
         .withReaderContext(readerContext)
@@ -55,6 +55,7 @@ public static HoodieFileGroupReader<IndexedRecord> createFileGroupReader(
         .withStart(start)
         .withLength(length)
         .withProperties(properties)
+        .withAllowInflightCommits(allowInflightCommits)
         .build(basePath, latestCommitTime, schema, shouldUseRecordPosition, metaClient);
   }
 
@@ -65,6 +66,7 @@ public static class HoodieFileGroupReaderBuilder {
     private TypedProperties props;
     private long start;
     private long length;
+    private boolean allowInflightCommits = false;
 
     public HoodieFileGroupReaderBuilder withReaderContext(
         HoodieReaderContext<IndexedRecord> context) {
@@ -97,6 +99,11 @@ public HoodieFileGroupReaderBuilder withLength(long length) {
       return this;
     }
 
+    public HoodieFileGroupReaderBuilder withAllowInflightCommits(boolean allowInflightCommits) {
+      this.allowInflightCommits = allowInflightCommits;
+      return this;
+    }
+
     public HoodieFileGroupReader<IndexedRecord> build(
         String basePath,
         String latestCommitTime,
@@ -108,20 +115,8 @@ public HoodieFileGroupReader<IndexedRecord> build(
       props.setProperty(HoodieMemoryConfig.SPILLABLE_MAP_BASE_PATH.key(),  basePath + "/" + HoodieTableMetaClient.TEMPFOLDER_NAME);
       props.setProperty(HoodieCommonConfig.SPILLABLE_DISK_MAP_TYPE.key(), ExternalSpillableMap.DiskMapType.ROCKS_DB.name());
       props.setProperty(HoodieCommonConfig.DISK_MAP_BITCASK_COMPRESSION_ENABLED.key(), "false");
-      return new HoodieFileGroupReader<>(
-          readerContext,
-          storage,
-          basePath,
-          latestCommitTime,
-          fileSlice,
-          schema,
-          schema,
-          Option.empty(),
-          metaClient,
-          props,
-          start,
-          length,
-          shouldUseRecordPosition);
+      return new HoodieFileGroupReader<>(readerContext, storage, basePath, latestCommitTime, fileSlice,
+          schema, schema, Option.empty(), metaClient, props, start, length, shouldUseRecordPosition, allowInflightCommits);
     }
   }
 }