lanyuanxiaoyao/hudi
1
0
Fork 0
Code Activity

[MINOR] Fix result missing information issue in commits_compare Procedure (#6165)

Browse Source 
Co-authored-by: superche <superche@tencent.com>
This commit is contained in:
superche
2022-07-22 07:25:22 +08:00
committed by GitHub
parent 36e656aa77
commit 8e0b47e360
1 changed files with 4 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/CommitsCompareProcedure.scala
View File

@@ -54,17 +54,15 @@ class CommitsCompareProcedure() extends BaseProcedure with ProcedureBuilder {
val sourceTimeline = source.getActiveTimeline.getCommitsTimeline.filterCompletedInstants
val targetTimeline = target.getActiveTimeline.getCommitsTimeline.filterCompletedInstants
val targetLatestCommit =
if (targetTimeline.getInstants.iterator.hasNext) targetTimeline.lastInstant.get.getTimestamp
else "0"
if (targetTimeline.getInstants.iterator.hasNext) targetTimeline.lastInstant.get.getTimestamp else "0"
val sourceLatestCommit =
if (sourceTimeline.getInstants.iterator.hasNext) sourceTimeline.lastInstant.get.getTimestamp
else "0"
if (sourceTimeline.getInstants.iterator.hasNext) sourceTimeline.lastInstant.get.getTimestamp else "0"
if (sourceLatestCommit != null && HoodieTimeline.compareTimestamps(targetLatestCommit, HoodieTimeline.GREATER_THAN, sourceLatestCommit)) { // source is behind the target
val commitsToCatchup = targetTimeline.findInstantsAfter(sourceLatestCommit, Integer.MAX_VALUE).getInstants.iterator().asScala.map(instant => instant.getTimestamp)
val commitsToCatchup = targetTimeline.findInstantsAfter(sourceLatestCommit, Integer.MAX_VALUE).getInstants.iterator().asScala.map(instant => instant.getTimestamp).toList.asJava
Seq(Row("Source " + source.getTableConfig.getTableName + " is behind by " + commitsToCatchup.size + " commits. Commits to catch up - " + commitsToCatchup))
} else {
val commitsToCatchup = sourceTimeline.findInstantsAfter(targetLatestCommit, Integer.MAX_VALUE).getInstants.iterator().asScala.map(instant => instant.getTimestamp)
val commitsToCatchup = sourceTimeline.findInstantsAfter(targetLatestCommit, Integer.MAX_VALUE).getInstants.iterator().asScala.map(instant => instant.getTimestamp).toList.asJava
Seq(Row("Source " + source.getTableConfig.getTableName + " is ahead by " + commitsToCatchup.size + " commits. Commits to catch up - " + commitsToCatchup))
}
}