[HUDI-2474] Refreshing timeline for every operation in Hudi when metadata is enabled (apache#3698)

nsivabalan · web-flow · commit f0585facd6e8 · 2021-09-28T05:16:52.000-04:00
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/AbstractHoodieWriteClient.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/AbstractHoodieWriteClient.java
@@ -220,7 +220,11 @@ protected void commit(HoodieTable table, String commitActionType, String instant
         Option.of(metadata.toJsonString().getBytes(StandardCharsets.UTF_8)));
   }
 
-  protected abstract HoodieTable<T, I, K, O> createTable(HoodieWriteConfig config, Configuration hadoopConf);
+  protected HoodieTable<T, I, K, O> createTable(HoodieWriteConfig config, Configuration hadoopConf) {
+    return createTable(config, hadoopConf, false);
+  }
+
+  protected abstract HoodieTable<T, I, K, O> createTable(HoodieWriteConfig config, Configuration hadoopConf, boolean refreshTimeline);
 
   void emitCommitMetrics(String instantTime, HoodieCommitMetadata metadata, String actionType) {
     try {
@@ -272,7 +276,7 @@ public void bootstrap(Option<Map<String, String>> extraMetadata) {
    */
   public void rollbackFailedBootstrap() {
     LOG.info("Rolling back pending bootstrap if present");
-    HoodieTable<T, I, K, O> table = createTable(config, hadoopConf);
+    HoodieTable<T, I, K, O> table = createTable(config, hadoopConf, config.isMetadataTableEnabled());
     HoodieTimeline inflightTimeline = table.getMetaClient().getCommitsTimeline().filterPendingExcludingCompaction();
     Option<String> instant = Option.fromJavaOptional(
         inflightTimeline.getReverseOrderedInstants().map(HoodieInstant::getTimestamp).findFirst());
@@ -451,6 +455,9 @@ protected void postCommit(HoodieTable<T, I, K, O> table, HoodieCommitMetadata me
 
   protected void runTableServicesInline(HoodieTable<T, I, K, O> table, HoodieCommitMetadata metadata, Option<Map<String, String>> extraMetadata) {
     if (config.inlineTableServices()) {
+      if (config.isMetadataTableEnabled()) {
+        table.getHoodieView().sync();
+      }
       // Do an inline compaction if enabled
       if (config.inlineCompactionEnabled()) {
         runAnyPendingCompactions(table);
@@ -515,7 +522,7 @@ protected void autoCleanOnCommit() {
    * @param comment - Comment for the savepoint
    */
   public void savepoint(String user, String comment) {
-    HoodieTable<T, I, K, O> table = createTable(config, hadoopConf);
+    HoodieTable<T, I, K, O> table = createTable(config, hadoopConf, config.isMetadataTableEnabled());
     if (table.getCompletedCommitsTimeline().empty()) {
       throw new HoodieSavepointException("Could not savepoint. Commit timeline is empty");
     }
@@ -539,7 +546,7 @@ public void savepoint(String user, String comment) {
    * @param comment - Comment for the savepoint
    */
   public void savepoint(String instantTime, String user, String comment) {
-    HoodieTable<T, I, K, O> table = createTable(config, hadoopConf);
+    HoodieTable<T, I, K, O> table = createTable(config, hadoopConf, config.isMetadataTableEnabled());
     table.savepoint(context, instantTime, user, comment);
   }
 
@@ -551,7 +558,7 @@ public void savepoint(String instantTime, String user, String comment) {
    * @return true if the savepoint was deleted successfully
    */
   public void deleteSavepoint(String savepointTime) {
-    HoodieTable<T, I, K, O> table = createTable(config, hadoopConf);
+    HoodieTable<T, I, K, O> table = createTable(config, hadoopConf, config.isMetadataTableEnabled());
     SavepointHelpers.deleteSavepoint(table, savepointTime);
   }
 
@@ -566,7 +573,7 @@ public void deleteSavepoint(String savepointTime) {
    * @return true if the savepoint was restored to successfully
    */
   public void restoreToSavepoint(String savepointTime) {
-    HoodieTable<T, I, K, O> table = createTable(config, hadoopConf);
+    HoodieTable<T, I, K, O> table = createTable(config, hadoopConf, config.isMetadataTableEnabled());
     SavepointHelpers.validateSavepointPresence(table, savepointTime);
     restoreToInstant(savepointTime);
     SavepointHelpers.validateSavepointRestore(table, savepointTime);
@@ -624,7 +631,7 @@ public HoodieRestoreMetadata restoreToInstant(final String instantTime) throws H
     final String restoreInstantTime = HoodieActiveTimeline.createNewInstantTime();
     Timer.Context timerContext = metrics.getRollbackCtx();
     try {
-      HoodieTable<T, I, K, O> table = createTable(config, hadoopConf);
+      HoodieTable<T, I, K, O> table = createTable(config, hadoopConf, config.isMetadataTableEnabled());
       HoodieRestoreMetadata restoreMetadata = table.restore(context, restoreInstantTime, instantTime);
       if (timerContext != null) {
         final long durationInMs = metrics.getDurationInMs(timerContext.stop());
@@ -957,17 +964,17 @@ private Option<String> scheduleTableServiceInternal(String instantTime, Option<M
     switch (tableServiceType) {
       case CLUSTER:
         LOG.info("Scheduling clustering at instant time :" + instantTime);
-        Option<HoodieClusteringPlan> clusteringPlan = createTable(config, hadoopConf)
+        Option<HoodieClusteringPlan> clusteringPlan = createTable(config, hadoopConf, config.isMetadataTableEnabled())
             .scheduleClustering(context, instantTime, extraMetadata);
         return clusteringPlan.isPresent() ? Option.of(instantTime) : Option.empty();
       case COMPACT:
         LOG.info("Scheduling compaction at instant time :" + instantTime);
-        Option<HoodieCompactionPlan> compactionPlan = createTable(config, hadoopConf)
+        Option<HoodieCompactionPlan> compactionPlan = createTable(config, hadoopConf, config.isMetadataTableEnabled())
             .scheduleCompaction(context, instantTime, extraMetadata);
         return compactionPlan.isPresent() ? Option.of(instantTime) : Option.empty();
       case CLEAN:
         LOG.info("Scheduling cleaning at instant time :" + instantTime);
-        Option<HoodieCleanerPlan> cleanerPlan = createTable(config, hadoopConf)
+        Option<HoodieCleanerPlan> cleanerPlan = createTable(config, hadoopConf, config.isMetadataTableEnabled())
             .scheduleCleaning(context, instantTime, extraMetadata);
         return cleanerPlan.isPresent() ? Option.of(instantTime) : Option.empty();
       default:
diff --git a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/HoodieFlinkWriteClient.java b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/HoodieFlinkWriteClient.java
@@ -121,7 +121,8 @@ public boolean commit(String instantTime, List<WriteStatus> writeStatuses, Optio
   }
 
   @Override
-  protected HoodieTable<T, List<HoodieRecord<T>>, List<HoodieKey>, List<WriteStatus>> createTable(HoodieWriteConfig config, Configuration hadoopConf) {
+  protected HoodieTable<T, List<HoodieRecord<T>>, List<HoodieKey>, List<WriteStatus>> createTable(HoodieWriteConfig config, Configuration hadoopConf,
+                                                                                                  boolean refreshTimeline) {
     return HoodieFlinkTable.create(config, (HoodieFlinkEngineContext) context);
   }
 
diff --git a/hudi-client/hudi-java-client/src/main/java/org/apache/hudi/client/HoodieJavaWriteClient.java b/hudi-client/hudi-java-client/src/main/java/org/apache/hudi/client/HoodieJavaWriteClient.java
@@ -89,7 +89,8 @@ public boolean commit(String instantTime,
 
   @Override
   protected HoodieTable<T, List<HoodieRecord<T>>, List<HoodieKey>, List<WriteStatus>> createTable(HoodieWriteConfig config,
-                                                                                                  Configuration hadoopConf) {
+                                                                                                  Configuration hadoopConf,
+                                                                                                  boolean refreshTimeline) {
     return HoodieJavaTable.create(config, context);
   }
 
diff --git a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/SparkRDDWriteClient.java b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/SparkRDDWriteClient.java
@@ -126,8 +126,9 @@ public boolean commit(String instantTime, JavaRDD<WriteStatus> writeStatuses, Op
 
   @Override
   protected HoodieTable<T, JavaRDD<HoodieRecord<T>>, JavaRDD<HoodieKey>, JavaRDD<WriteStatus>> createTable(HoodieWriteConfig config,
-                                                                                                           Configuration hadoopConf) {
-    return HoodieSparkTable.create(config, context);
+                                                                                                           Configuration hadoopConf,
+                                                                                                           boolean refreshTimeline) {
+    return HoodieSparkTable.create(config, context, refreshTimeline);
   }
 
   @Override
@@ -319,7 +320,7 @@ protected void completeCompaction(HoodieCommitMetadata metadata, JavaRDD<WriteSt
 
   @Override
   protected JavaRDD<WriteStatus> compact(String compactionInstantTime, boolean shouldComplete) {
-    HoodieSparkTable<T> table = HoodieSparkTable.create(config, context);
+    HoodieSparkTable<T> table = HoodieSparkTable.create(config, context, config.isMetadataTableEnabled());
     preWrite(compactionInstantTime, WriteOperationType.COMPACT, table.getMetaClient());
     HoodieTimeline pendingCompactionTimeline = table.getActiveTimeline().filterPendingCompactionTimeline();
     HoodieInstant inflightInstant = HoodieTimeline.getCompactionInflightInstant(compactionInstantTime);
@@ -338,7 +339,7 @@ protected JavaRDD<WriteStatus> compact(String compactionInstantTime, boolean sho
 
   @Override
   public HoodieWriteMetadata<JavaRDD<WriteStatus>> cluster(String clusteringInstant, boolean shouldComplete) {
-    HoodieSparkTable<T> table = HoodieSparkTable.create(config, context);
+    HoodieSparkTable<T> table = HoodieSparkTable.create(config, context, config.isMetadataTableEnabled());
     preWrite(clusteringInstant, WriteOperationType.CLUSTER, table.getMetaClient());
     HoodieTimeline pendingClusteringTimeline = table.getActiveTimeline().filterPendingReplaceTimeline();
     HoodieInstant inflightInstant = HoodieTimeline.getReplaceCommitInflightInstant(clusteringInstant);
@@ -438,7 +439,7 @@ private HoodieTable<T, JavaRDD<HoodieRecord<T>>, JavaRDD<HoodieKey>, JavaRDD<Wri
       setWriteSchemaForDeletes(metaClient);
     }
     // Create a Hoodie table which encapsulated the commits and files visible
-    HoodieSparkTable<T> table = HoodieSparkTable.create(config, (HoodieSparkEngineContext) context, metaClient);
+    HoodieSparkTable<T> table = HoodieSparkTable.create(config, (HoodieSparkEngineContext) context, metaClient, config.isMetadataTableEnabled());
     if (table.getMetaClient().getCommitActionType().equals(HoodieTimeline.COMMIT_ACTION)) {
       writeTimer = metrics.getCommitCtx();
     } else {
diff --git a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/HoodieSparkTable.java b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/HoodieSparkTable.java
@@ -42,24 +42,43 @@ protected HoodieSparkTable(HoodieWriteConfig config, HoodieEngineContext context
   }
 
   public static <T extends HoodieRecordPayload> HoodieSparkTable<T> create(HoodieWriteConfig config, HoodieEngineContext context) {
+    return create(config, context, false);
+  }
+
+  public static <T extends HoodieRecordPayload> HoodieSparkTable<T> create(HoodieWriteConfig config, HoodieEngineContext context,
+                                                                           boolean refreshTimeline) {
     HoodieTableMetaClient metaClient =
         HoodieTableMetaClient.builder().setConf(context.getHadoopConf().get()).setBasePath(config.getBasePath())
             .setLoadActiveTimelineOnLoad(true).setConsistencyGuardConfig(config.getConsistencyGuardConfig())
             .setLayoutVersion(Option.of(new TimelineLayoutVersion(config.getTimelineLayoutVersion()))).build();
-    return HoodieSparkTable.create(config, (HoodieSparkEngineContext) context, metaClient);
+    return HoodieSparkTable.create(config, (HoodieSparkEngineContext) context, metaClient, refreshTimeline);
   }
 
   public static <T extends HoodieRecordPayload> HoodieSparkTable<T> create(HoodieWriteConfig config,
                                                                            HoodieSparkEngineContext context,
                                                                            HoodieTableMetaClient metaClient) {
+    return create(config, context, metaClient, false);
+  }
+
+  public static <T extends HoodieRecordPayload> HoodieSparkTable<T> create(HoodieWriteConfig config,
+                                                                           HoodieSparkEngineContext context,
+                                                                           HoodieTableMetaClient metaClient,
+                                                                           boolean refreshTimeline) {
+    HoodieSparkTable hoodieSparkTable;
     switch (metaClient.getTableType()) {
       case COPY_ON_WRITE:
-        return new HoodieSparkCopyOnWriteTable<>(config, context, metaClient);
+        hoodieSparkTable = new HoodieSparkCopyOnWriteTable<>(config, context, metaClient);
+        break;
       case MERGE_ON_READ:
-        return new HoodieSparkMergeOnReadTable<>(config, context, metaClient);
+        hoodieSparkTable = new HoodieSparkMergeOnReadTable<>(config, context, metaClient);
+        break;
       default:
         throw new HoodieException("Unsupported table type :" + metaClient.getTableType());
     }
+    if (refreshTimeline) {
+      hoodieSparkTable.getHoodieView().sync();
+    }
+    return hoodieSparkTable;
   }
 
   @Override
diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestHoodieBackedMetadata.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestHoodieBackedMetadata.java
@@ -71,6 +71,7 @@
 import org.apache.spark.api.java.JavaRDD;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Tag;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.Arguments;
@@ -547,6 +548,7 @@ public void testSync(HoodieTableType tableType) throws Exception {
    */
   @ParameterizedTest
   @EnumSource(HoodieTableType.class)
+  @Disabled
   public void testCleaningArchivingAndCompaction(HoodieTableType tableType) throws Exception {
     init(tableType);
     doWriteOperationsAndBootstrapMetadata(testTable);
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/view/RemoteHoodieTableFileSystemView.java b/hudi-common/src/main/java/org/apache/hudi/common/table/view/RemoteHoodieTableFileSystemView.java
@@ -479,7 +479,7 @@ public HoodieTimeline getTimeline() {
 
   @Override
   public void sync() {
-    // noop
+    refresh();
   }
 
   @Override

Original file line number	Diff line number	Diff line change
`@@ -121,7 +121,8 @@ public boolean commit(String instantTime, List<WriteStatus> writeStatuses, Optio`
`121`	`121`	`}`
`122`	`122`
`123`	`123`	`@Override`
`124`		`- protected HoodieTable<T, List<HoodieRecord<T>>, List<HoodieKey>, List<WriteStatus>> createTable(HoodieWriteConfig config, Configuration hadoopConf) {`
	`124`	`+ protected HoodieTable<T, List<HoodieRecord<T>>, List<HoodieKey>, List<WriteStatus>> createTable(HoodieWriteConfig config, Configuration hadoopConf,`
	`125`	`+ boolean refreshTimeline) {`
`125`	`126`	`return HoodieFlinkTable.create(config, (HoodieFlinkEngineContext) context);`
`126`	`127`	`}`
`127`	`128`
Original file line number	Diff line number	Diff line change
`@@ -89,7 +89,8 @@ public boolean commit(String instantTime,`
`89`	`89`
`90`	`90`	`@Override`
`91`	`91`	`protected HoodieTable<T, List<HoodieRecord<T>>, List<HoodieKey>, List<WriteStatus>> createTable(HoodieWriteConfig config,`
`92`		`- Configuration hadoopConf) {`
	`92`	`+ Configuration hadoopConf,`
	`93`	`+ boolean refreshTimeline) {`
`93`	`94`	`return HoodieJavaTable.create(config, context);`
`94`	`95`	`}`
`95`	`96`
Original file line number	Diff line number	Diff line change
`@@ -479,7 +479,7 @@ public HoodieTimeline getTimeline() {`
`479`	`479`
`480`	`480`	`@Override`
`481`	`481`	`public void sync() {`
`482`		`- // noop`
	`482`	`+ refresh();`
`483`	`483`	`}`
`484`	`484`
`485`	`485`	`@Override`