feat: [HUDI-9766] Support for show_timeline Procedure with appropriate start and end time for both active and archive timelines

hudi-common/src/main/java/org/apache/hudi/common/table/timeline/HoodieArchivedTimeline.java

@@ -29,10 +29,16 @@ public interface HoodieArchivedTimeline extends HoodieTimeline {
 
   void loadCompletedInstantDetailsInMemory();
 
+  void loadCompletedInstantDetailsInMemory(String startTs, String endTs);
+
+  void loadCompletedInstantDetailsInMemory(int limit);
+
   void loadCompactionDetailsInMemory(String compactionInstantTime);
 
   void loadCompactionDetailsInMemory(String startTs, String endTs);
 
+  void loadCompactionDetailsInMemory(int limit);
+
   void clearInstantDetailsFromMemory(String instantTime);
 
   void clearInstantDetailsFromMemory(String startTs, String endTs);

hudi-common/src/main/java/org/apache/hudi/common/table/timeline/versioning/v1/ArchivedTimelineV1.java

@@ -218,6 +218,34 @@ record -> {
         });
   }
 
+  @Override
+  public void loadCompactionDetailsInMemory(int limit) {
+    loadInstantsWithLimit(limit, true,
+        record -> {
+          Object action = record.get(ACTION_STATE);
+          return record.get(ACTION_TYPE_KEY).toString().equals(HoodieTimeline.COMPACTION_ACTION)
+              && (action == null || org.apache.hudi.common.table.timeline.HoodieInstant.State.INFLIGHT.toString().equals(action.toString()));
+        });
+  }
+
+  @Override
+  public void loadCompletedInstantDetailsInMemory(String startTs, String endTs) {
+    loadInstants(new ClosedClosedTimeRangeFilter(startTs, endTs), null, true,
+        record -> {
+          Object action = record.get(ACTION_STATE);
+          return action == null || org.apache.hudi.common.table.timeline.HoodieInstant.State.COMPLETED.toString().equals(action.toString());
+        });
+  }
+
+  @Override
+  public void loadCompletedInstantDetailsInMemory(int limit) {
+    loadInstantsWithLimit(limit, true,
+        record -> {
+          Object action = record.get(ACTION_STATE);
+          return action == null || org.apache.hudi.common.table.timeline.HoodieInstant.State.COMPLETED.toString().equals(action.toString());
+        });
+  }
+
   @Override
   public void clearInstantDetailsFromMemory(String instantTime) {
     this.readCommits.remove(instantTime);
@@ -249,6 +277,12 @@ private List<HoodieInstant> loadInstants(HoodieArchivedTimeline.TimeRangeFilter
         .stream().flatMap(Collection::stream).sorted().collect(Collectors.toList());
   }
 
+  private void loadInstantsWithLimit(int limit, boolean loadInstantDetails, Function<GenericRecord, Boolean> commitsFilter) {
+    InstantsLoaderWithLimit loader = new InstantsLoaderWithLimit(loadInstantDetails, limit);
+    timelineLoader.loadInstants(
+        metaClient, null, Option.empty(), LoadMode.PLAN, commitsFilter, loader);
+  }
+
   /**
    * Callback to read instant details.
    */
@@ -274,6 +308,39 @@ public Map<String, List<HoodieInstant>> getInstantsInRangeCollected() {
     }
   }
 
+  public class InstantsLoaderWithLimit implements BiConsumer<String, GenericRecord> {
+    private final Map<String, List<HoodieInstant>> instantsInRange = new ConcurrentHashMap<>();
+    private final boolean loadInstantDetails;
+    private final int limit;
+    private volatile int loadedCount = 0;
+
+    private InstantsLoaderWithLimit(boolean loadInstantDetails, int limit) {
+      this.loadInstantDetails = loadInstantDetails;
+      this.limit = limit;
+    }
+
+    @Override
+    public void accept(String instantTime, GenericRecord record) {
+      if (loadedCount >= limit) {
+        return;
+      }
+      Option<HoodieInstant> instant = readCommit(instantTime, record, loadInstantDetails, null);
+      if (instant.isPresent()) {
+        synchronized (this) {
+          if (loadedCount < limit) {
+            instantsInRange.computeIfAbsent(instant.get().requestedTime(), s -> new ArrayList<>())
+                .add(instant.get());
+            loadedCount++;
+          }
+        }
+      }
+    }
+
+    public Map<String, List<HoodieInstant>> getInstantsInRangeCollected() {
+      return instantsInRange;
+    }
+  }
+
   private Option<HoodieInstant> readCommit(String instantTime, GenericRecord record, boolean loadDetails,
                                            TimeRangeFilter timeRangeFilter) {
     final String action = record.get(ACTION_TYPE_KEY).toString();

hudi-common/src/main/java/org/apache/hudi/common/table/timeline/versioning/v2/ArchivedTimelineV2.java

@@ -137,6 +137,24 @@ record -> record.get(ACTION_ARCHIVED_META_FIELD).toString().equals(COMMIT_ACTION
     );
   }
 
+  @Override
+  public void loadCompactionDetailsInMemory(int limit) {
+    loadInstantsWithLimit(limit, HoodieArchivedTimeline.LoadMode.PLAN,
+        record -> record.get(ACTION_ARCHIVED_META_FIELD).toString().equals(COMMIT_ACTION)
+            && record.get(PLAN_ARCHIVED_META_FIELD) != null
+    );
+  }
+
+  @Override
+  public void loadCompletedInstantDetailsInMemory(String startTs, String endTs) {
+    loadInstants(new HoodieArchivedTimeline.TimeRangeFilter(startTs, endTs), HoodieArchivedTimeline.LoadMode.METADATA);
+  }
+
+  @Override
+  public void loadCompletedInstantDetailsInMemory(int limit) {
+    loadInstantsWithLimit(limit, HoodieArchivedTimeline.LoadMode.METADATA, r -> true);
+  }
+
   @Override
   public void clearInstantDetailsFromMemory(String instantTime) {
     this.readCommits.remove(instantTime);
@@ -242,6 +260,47 @@ private List<HoodieInstant> loadInstants(
     return result;
   }
 
+  /**
+   * Loads instants with a limit on the number of instants to load.
+   * This is used for limit-based loading where we only want to load the N most recent instants.
+   */
+  private void loadInstantsWithLimit(int limit, HoodieArchivedTimeline.LoadMode loadMode,
+      Function<GenericRecord, Boolean> commitsFilter) {
+    InstantsLoaderWithLimit loader = new InstantsLoaderWithLimit(limit, loadMode);
+    timelineLoader.loadInstants(metaClient, null, loadMode, commitsFilter, loader);
+  }
+
+  /**
+   * Callback to read instant details with a limit on the number of instants to load.
+   * Extends BiConsumer to be used as a callback in the timeline loader.
+   * The BiConsumer interface allows it to be passed as a lambda/function that accepts
+   * (instantTime, GenericRecord) pairs during the loading process.
+   */
+  private class InstantsLoaderWithLimit implements BiConsumer<String, GenericRecord> {
+    private final int limit;
+    private final HoodieArchivedTimeline.LoadMode loadMode;
+    private volatile int loadedCount = 0;
+
+    private InstantsLoaderWithLimit(int limit, HoodieArchivedTimeline.LoadMode loadMode) {
+      this.limit = limit;
+      this.loadMode = loadMode;
+    }
+
+    @Override
+    public void accept(String instantTime, GenericRecord record) {
+      if (loadedCount >= limit) {
+        return;
+      }
+      Option<BiConsumer<String, GenericRecord>> instantDetailsConsumer = Option.ofNullable(getInstantDetailsFunc(loadMode));
+      readCommit(instantTime, record, instantDetailsConsumer);
+      synchronized (this) {
+        if (loadedCount < limit) {
+          loadedCount++;
+        }
+      }
+    }
+  }
+
   @Override
   public HoodieTimeline getWriteTimeline() {
     // filter in-memory instants

hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/HoodieProcedures.scala

@@ -100,6 +100,7 @@ object HoodieProcedures {
       ,(ShowCleansProcedure.NAME, ShowCleansProcedure.builder)
       ,(ShowCleansPartitionMetadataProcedure.NAME, ShowCleansPartitionMetadataProcedure.builder)
       ,(ShowCleansPlanProcedure.NAME, ShowCleansPlanProcedure.builder)
+      ,(ShowTimelineProcedure.NAME, ShowTimelineProcedure.builder)
       ,(SetAuditLockProcedure.NAME, SetAuditLockProcedure.builder)
       ,(ShowAuditLockStatusProcedure.NAME, ShowAuditLockStatusProcedure.builder)
       ,(ValidateAuditLockProcedure.NAME, ValidateAuditLockProcedure.builder)