feature(web): 增加时间线 Rollback 和 Clean 的查询

2023-07-06 19:26:49 +08:00
parent f0c4031365
commit a5a9f600f1
8 changed files with 571 additions and 26 deletions
--- a/service-hudi-query/src/main/java/com/lanyuanxiaoyao/service/hudi/controller/TimelineController.java
+++ b/service-hudi-query/src/main/java/com/lanyuanxiaoyao/service/hudi/controller/TimelineController.java
@@ -1,8 +1,10 @@
 package com.lanyuanxiaoyao.service.hudi.controller;

 import com.lanyuanxiaoyao.service.configuration.entity.PageResponse;
+import com.lanyuanxiaoyao.service.configuration.entity.hudi.HudiCleanerPlan;
 import com.lanyuanxiaoyao.service.configuration.entity.hudi.HudiCompactionPlan;
 import com.lanyuanxiaoyao.service.configuration.entity.hudi.HudiInstant;
+import com.lanyuanxiaoyao.service.configuration.entity.hudi.HudiRollbackPlan;
 import com.lanyuanxiaoyao.service.hudi.service.TimelineService;
 import java.io.IOException;
 import java.util.List;
@@ -90,13 +92,47 @@ public class TimelineController {
    }

    @GetMapping("read_compaction_plan_hdfs")
-    public HudiCompactionPlan readCompactionPlan(
+    public HudiCompactionPlan readCompactionPlanHdfs(
            @RequestParam("hdfs") String hdfs,
            @RequestParam("instant") String instant
    ) throws IOException {
        return timelineService.readCompactionPlan(hdfs, instant);
    }

+    @GetMapping("read_rollback_plan")
+    public HudiRollbackPlan readRollbackPlan(
+            @RequestParam("flink_job_id") Long flinkJobId,
+            @RequestParam("alias") String alias,
+            @RequestParam("instant") String instant
+    ) throws IOException {
+        return timelineService.readRollbackPlan(flinkJobId, alias, instant);
+    }
+
+    @GetMapping("read_rollback_plan_hdfs")
+    public HudiRollbackPlan readRollbackPlanHdfs(
+            @RequestParam("hdfs") String hdfs,
+            @RequestParam("instant") String instant
+    ) throws IOException {
+        return timelineService.readRollbackPlan(hdfs, instant);
+    }
+
+    @GetMapping("read_cleaner_plan")
+    public HudiCleanerPlan readCleanerPlan(
+            @RequestParam("flink_job_id") Long flinkJobId,
+            @RequestParam("alias") String alias,
+            @RequestParam("instant") String instant
+    ) throws IOException {
+        return timelineService.readCleanerPlan(flinkJobId, alias, instant);
+    }
+
+    @GetMapping("read_cleaner_plan_hdfs")
+    public HudiCleanerPlan readCleanerPlanHdfs(
+            @RequestParam("hdfs") String hdfs,
+            @RequestParam("instant") String instant
+    ) throws IOException {
+        return timelineService.readCleanerPlan(hdfs, instant);
+    }
+
    @GetMapping("list_pending_compaction")
    public ImmutableList<HudiInstant> pendingCompactionInstants(
            @RequestParam("flink_job_id") Long flinkJobId,
--- a/service-hudi-query/src/main/java/com/lanyuanxiaoyao/service/hudi/service/TimelineService.java
+++ b/service-hudi-query/src/main/java/com/lanyuanxiaoyao/service/hudi/service/TimelineService.java
@@ -5,8 +5,10 @@ import cn.hutool.core.util.StrUtil;
 import com.eshore.odcp.hudi.connector.entity.TableMeta;
 import com.lanyuanxiaoyao.service.configuration.ExecutorProvider;
 import com.lanyuanxiaoyao.service.configuration.entity.PageResponse;
+import com.lanyuanxiaoyao.service.configuration.entity.hudi.HudiCleanerPlan;
 import com.lanyuanxiaoyao.service.configuration.entity.hudi.HudiCompactionPlan;
 import com.lanyuanxiaoyao.service.configuration.entity.hudi.HudiInstant;
+import com.lanyuanxiaoyao.service.configuration.entity.hudi.HudiRollbackPlan;
 import com.lanyuanxiaoyao.service.configuration.utils.ComparatorUtil;
 import com.lanyuanxiaoyao.service.forest.service.InfoService;
 import com.lanyuanxiaoyao.service.hudi.utils.HoodieUtils;
@@ -16,11 +18,15 @@ import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
+import org.apache.hudi.avro.model.HoodieCleanerPlan;
 import org.apache.hudi.avro.model.HoodieCompactionPlan;
+import org.apache.hudi.avro.model.HoodieRollbackPlan;
 import org.apache.hudi.common.table.HoodieTableMetaClient;
 import org.apache.hudi.common.table.timeline.HoodieInstant;
 import org.apache.hudi.common.table.timeline.HoodieTimeline;
+import org.apache.hudi.common.util.CleanerUtils;
 import org.apache.hudi.common.util.CompactionUtils;
+import org.apache.hudi.table.action.rollback.RollbackUtils;
 import org.eclipse.collections.api.factory.Lists;
 import org.eclipse.collections.api.factory.Maps;
 import org.eclipse.collections.api.list.ImmutableList;
@@ -142,14 +148,14 @@ public class TimelineService {
        return new PageResponse<>(result.toList(), hudiInstants.size());
    }

-    @Cacheable(value = "read_compaction_plan", sync = true)
+    @Cacheable(value = "read-compaction-plan", sync = true)
    @Retryable(Throwable.class)
    public HudiCompactionPlan readCompactionPlan(Long flinkJobId, String alias, String instant) throws IOException {
        TableMeta meta = infoService.tableMetaDetail(flinkJobId, alias);
        return readCompactionPlan(meta.getHudi().getTargetHdfsPath(), instant);
    }

-    @Cacheable(value = "read_compaction_plan", sync = true)
+    @Cacheable(value = "read-compaction-plan", sync = true)
    @Retryable(Throwable.class)
    public HudiCompactionPlan readCompactionPlan(String hdfs, String instant) throws IOException {
        HoodieTableMetaClient client = HoodieTableMetaClient.builder()
@@ -158,6 +164,7 @@ public class TimelineService {
                .build();
        HoodieCompactionPlan plan = CompactionUtils.getCompactionPlan(client, instant);
        return new HudiCompactionPlan(
+                plan.getVersion(),
                ObjectUtil.isNotNull(plan.getOperations())
                        ? Lists.immutable.ofAll(plan.getOperations())
                        .collect(o -> new HudiCompactionPlan.Operation(
@@ -170,8 +177,73 @@ public class TimelineService {
                                o.getBootstrapFilePath()
                        ))
                        : Lists.immutable.empty(),
-                ObjectUtil.isNotNull(plan.getExtraMetadata()) ? Maps.immutable.ofAll(plan.getExtraMetadata()) : Maps.immutable.empty(),
-                plan.getVersion()
+                ObjectUtil.isNotNull(plan.getExtraMetadata()) ? Maps.immutable.ofAll(plan.getExtraMetadata()) : Maps.immutable.empty()
+        );
+    }
+
+    @Cacheable(value = "read-rollback-plan", sync = true)
+    @Retryable(Throwable.class)
+    public HudiRollbackPlan readRollbackPlan(Long flinkJobId, String alias, String instant) throws IOException {
+        TableMeta meta = infoService.tableMetaDetail(flinkJobId, alias);
+        return readRollbackPlan(meta.getHudi().getTargetHdfsPath(), instant);
+    }
+
+    @Cacheable(value = "read-rollback-plan", sync = true)
+    @Retryable(Throwable.class)
+    public HudiRollbackPlan readRollbackPlan(String hdfs, String instant) throws IOException {
+        HoodieTableMetaClient client = HoodieTableMetaClient.builder()
+                .setConf(new Configuration())
+                .setBasePath(hdfs)
+                .build();
+        HoodieRollbackPlan plan = RollbackUtils.getRollbackPlan(client, new HoodieInstant(HoodieInstant.State.INFLIGHT, HoodieTimeline.ROLLBACK_ACTION, instant));
+        return new HudiRollbackPlan(
+                plan.getVersion(),
+                ObjectUtil.isNotNull(plan.getInstantToRollback())
+                        ? new HudiRollbackPlan.Info(plan.getInstantToRollback().getAction(), plan.getInstantToRollback().getCommitTime())
+                        : null,
+                ObjectUtil.isNotNull(plan.getRollbackRequests())
+                        ? Lists.immutable.ofAll(plan.getRollbackRequests())
+                        .collect(r -> new HudiRollbackPlan.Request(
+                                r.getFileId(),
+                                r.getPartitionPath(),
+                                r.getLatestBaseInstant(),
+                                Lists.immutable.ofAll(r.getFilesToBeDeleted()),
+                                Maps.immutable.ofAll(r.getLogBlocksToBeDeleted())
+                        ))
+                        : Lists.immutable.empty()
+        );
+    }
+
+    @Cacheable(value = "read-cleaner-plan", sync = true)
+    @Retryable(Throwable.class)
+    public HudiCleanerPlan readCleanerPlan(Long flinkJobId, String alias, String instant) throws IOException {
+        TableMeta meta = infoService.tableMetaDetail(flinkJobId, alias);
+        return readCleanerPlan(meta.getHudi().getTargetHdfsPath(), instant);
+    }
+
+    @Cacheable(value = "read-cleaner-plan", sync = true)
+    @Retryable(Throwable.class)
+    public HudiCleanerPlan readCleanerPlan(String hdfs, String instant) throws IOException {
+        HoodieTableMetaClient client = HoodieTableMetaClient.builder()
+                .setConf(new Configuration())
+                .setBasePath(hdfs)
+                .build();
+        HoodieCleanerPlan plan = CleanerUtils.getCleanerPlan(client, HoodieTimeline.getCleanInflightInstant(instant));
+        return new HudiCleanerPlan(
+                plan.getVersion(),
+                plan.getPolicy(),
+                ObjectUtil.isNotNull(plan.getEarliestInstantToRetain())
+                        ? new HudiCleanerPlan.Instant(plan.getEarliestInstantToRetain().getAction(), plan.getEarliestInstantToRetain().getState(), plan.getEarliestInstantToRetain().getTimestamp())
+                        : null,
+                Maps.immutable.ofAll(plan.getFilePathsToBeDeletedPerPartition())
+                        .collectValues((key, value) -> Lists.immutable.ofAll(value)
+                                .collect(i -> new HudiCleanerPlan.Info(
+                                        i.getFilePath(),
+                                        i.getIsBootstrapBaseFile()
+                                ))),
+                Maps.immutable.ofAll(plan.getFilesToBeDeletedPerPartition())
+                        .collectValues((key, value) -> Lists.immutable.ofAll(value)),
+                Lists.immutable.ofAll(plan.getPartitionsToBeDeleted())
        );
    }