Merge branch 'master' into fix_optimizer_table_loading

apache · Aug 6, 2024 · 94addfa · 94addfa
2 parents b0e88e9 + df81844
commit 94addfa
Show file tree

Hide file tree

Showing 2 changed files with 20 additions and 16 deletions.
diff --git a/...oro-ams-server/src/main/java/org/apache/amoro/server/dashboard/PaimonTableDescriptor.java b/...oro-ams-server/src/main/java/org/apache/amoro/server/dashboard/PaimonTableDescriptor.java
@@ -525,14 +525,24 @@ public List<TagOrBranchInfo> getTableBranches(AmoroTable<?> amoroTable) {
   @Override
   public List<ConsumerInfo> getTableConsumerInfos(AmoroTable<?> amoroTable) {
     FileStoreTable table = getTable(amoroTable);
+    FileStore<?> store = table.store();
     ConsumerManager consumerManager = new ConsumerManager(table.fileIO(), table.location());
     List<ConsumerInfo> consumerInfos = new ArrayList<>();
     try {
       consumerManager
           .consumers()
           .forEach(
-              (consumerId, nextSnapshot) ->
-                  consumerInfos.add(new ConsumerInfo(consumerId, nextSnapshot)));
+              (consumerId, nextSnapshotId) -> {
+                long currentSnapshotId = nextSnapshotId;
+                if (!table.snapshotManager().snapshotExists(currentSnapshotId)) {
+                  // if not exits,maybe steaming scan is running,so need to nextSnapshotId -1
+                  currentSnapshotId = nextSnapshotId - 1;
+                }
+                Snapshot snapshot = table.snapshotManager().snapshot(currentSnapshotId);
+                AmoroSnapshotsOfTable amoroSnapshotsOfTable = getSnapshotsOfTable(store, snapshot);
+                consumerInfos.add(
+                    new ConsumerInfo(consumerId, nextSnapshotId, amoroSnapshotsOfTable));
+              });
     } catch (IOException e) {
       throw new RuntimeException(e);
     }

diff --git a/.../amoro-ams-server/src/main/java/org/apache/amoro/server/dashboard/model/ConsumerInfo.java b/.../amoro-ams-server/src/main/java/org/apache/amoro/server/dashboard/model/ConsumerInfo.java
@@ -20,32 +20,26 @@
 
 public class ConsumerInfo {
 
-  public static final String CONSUMER_ID = "consumer_id";
-  public static final String NEXT_SNAPSHOT_ID = "next_snapshot_id";
+  private final String consumerId;
+  private final long nextSnapshotId;
+  private final AmoroSnapshotsOfTable amoroCurrentSnapshotsOfTable;
 
-  private String consumerId;
-  private long nextSnapshotId;
-
-  public ConsumerInfo() {}
-
-  public ConsumerInfo(String consumerId, long nextSnapshotId) {
+  public ConsumerInfo(
+      String consumerId, long nextSnapshotId, AmoroSnapshotsOfTable amoroCurrentSnapshotsOfTable) {
     this.consumerId = consumerId;
     this.nextSnapshotId = nextSnapshotId;
+    this.amoroCurrentSnapshotsOfTable = amoroCurrentSnapshotsOfTable;
   }
 
   public String getConsumerId() {
     return consumerId;
   }
 
-  public void setConsumerId(String consumerId) {
-    this.consumerId = consumerId;
-  }
-
   public long getNextSnapshotId() {
     return nextSnapshotId;
   }
 
-  public void setNextSnapshotId(long nextSnapshotId) {
-    this.nextSnapshotId = nextSnapshotId;
+  public AmoroSnapshotsOfTable getAmoroCurrentSnapshotsOfTable() {
+    return amoroCurrentSnapshotsOfTable;
   }
 }