[HUDI-2285] Metadata Table synchronous design

hudi-cli/src/main/java/org/apache/hudi/cli/HoodieCLI.java

@@ -42,7 +42,7 @@ public class HoodieCLI {
   public static CLIState state = CLIState.INIT;
   public static String basePath;
   protected static HoodieTableMetaClient tableMetadata;
-  public static HoodieTableMetaClient syncTableMetadata;
+  public static HoodieTableMetaClient metaClient;
   public static TimelineLayoutVersion layoutVersion;
   private static TempViewProvider tempViewProvider;
 

hudi-cli/src/main/java/org/apache/hudi/cli/HoodiePrompt.java

@@ -40,7 +40,7 @@ public String getPrompt() {
         case TABLE:
           return "hudi:" + tableName + "->";
         case SYNC:
-          return "hudi:" + tableName + " <==> " + HoodieCLI.syncTableMetadata.getTableConfig().getTableName() + "->";
+          return "hudi:" + tableName + " <==> " + HoodieCLI.metaClient.getTableConfig().getTableName() + "->";
         default:
           return "hudi:" + tableName + "->";
       }

hudi-cli/src/main/java/org/apache/hudi/cli/commands/CommitsCommand.java

@@ -442,10 +442,10 @@ public String compareCommits(@CliOption(key = {"path"}, help = "Path of the tabl
 
   @CliCommand(value = "commits sync", help = "Compare commits with another Hoodie table")
   public String syncCommits(@CliOption(key = {"path"}, help = "Path of the table to compare to") final String path) {
-    HoodieCLI.syncTableMetadata = HoodieTableMetaClient.builder().setConf(HoodieCLI.conf).setBasePath(path).build();
+    HoodieCLI.metaClient = HoodieTableMetaClient.builder().setConf(HoodieCLI.conf).setBasePath(path).build();
     HoodieCLI.state = HoodieCLI.CLIState.SYNC;
     return "Load sync state between " + HoodieCLI.getTableMetaClient().getTableConfig().getTableName() + " and "
-        + HoodieCLI.syncTableMetadata.getTableConfig().getTableName();
+        + HoodieCLI.metaClient.getTableConfig().getTableName();
   }
 
   /*

hudi-cli/src/main/java/org/apache/hudi/cli/commands/HoodieSyncCommand.java

@@ -56,10 +56,10 @@ public String validateSync(
       @CliOption(key = {"hivePass"}, mandatory = true, unspecifiedDefaultValue = "",
           help = "hive password to connect to") final String hivePass)
       throws Exception {
-    if (HoodieCLI.syncTableMetadata == null) {
+    if (HoodieCLI.metaClient == null) {
       throw new HoodieException("Sync validate request target table not null.");
     }
-    HoodieTableMetaClient target = HoodieCLI.syncTableMetadata;
+    HoodieTableMetaClient target = HoodieCLI.metaClient;
     HoodieTimeline targetTimeline = target.getActiveTimeline().getCommitsTimeline();
     HoodieTableMetaClient source = HoodieCLI.getTableMetaClient();
     HoodieTimeline sourceTimeline = source.getActiveTimeline().getCommitsTimeline();

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/AbstractHoodieWriteClient.java

@@ -60,6 +60,7 @@
 import org.apache.hudi.exception.HoodieRollbackException;
 import org.apache.hudi.exception.HoodieSavepointException;
 import org.apache.hudi.index.HoodieIndex;
+import org.apache.hudi.metadata.HoodieTableMetadataWriter;
 import org.apache.hudi.metrics.HoodieMetrics;
 import org.apache.hudi.table.BulkInsertPartitioner;
 import org.apache.hudi.table.HoodieTable;
@@ -187,6 +188,8 @@ public boolean commitStats(String instantTime, List<HoodieWriteStat> stats, Opti
         lastCompletedTxnAndMetadata.isPresent() ? Option.of(lastCompletedTxnAndMetadata.get().getLeft()) : Option.empty());
     try {
       preCommit(instantTime, metadata);
+      // TODO: Should this be under its own transaction
+      table.getMetadataWriter().ifPresent(w -> ((HoodieTableMetadataWriter)w).update(metadata, instantTime));
       commit(table, commitActionType, instantTime, metadata, stats);
       postCommit(table, metadata, instantTime, extraMetadata);
       LOG.info("Committed " + instantTime);
@@ -241,10 +244,6 @@ protected void preCommit(String instantTime, HoodieCommitMetadata metadata) {
     // TODO : Conflict resolution is not supported for Flink & Java engines
   }
 
-  protected void syncTableMetadata() {
-    // no-op
-  }
-
   /**
    * Filter out HoodieRecords that already exists in the output folder. This is useful in deduplication.
    *
@@ -399,14 +398,6 @@ protected void preWrite(String instantTime, WriteOperationType writeOperationTyp
       HoodieTableMetaClient metaClient) {
     setOperationType(writeOperationType);
     this.lastCompletedTxnAndMetadata = TransactionUtils.getLastCompletedTxnInstantAndMetadata(metaClient);
-    this.txnManager.beginTransaction(Option.of(new HoodieInstant(State.INFLIGHT, metaClient.getCommitActionType(), instantTime)), lastCompletedTxnAndMetadata
-        .isPresent()
-        ? Option.of(lastCompletedTxnAndMetadata.get().getLeft()) : Option.empty());
-    try {
-      syncTableMetadata();
-    } finally {
-      this.txnManager.endTransaction();
-    }
     this.asyncCleanerService = AsyncCleanerService.startAsyncCleaningIfEnabled(this);
   }
 
@@ -435,7 +426,6 @@ protected void postCommit(HoodieTable<T, I, K, O> table, HoodieCommitMetadata me
       HoodieTimelineArchiveLog archiveLog = new HoodieTimelineArchiveLog(config, table);
       archiveLog.archiveIfRequired(context);
       autoCleanOnCommit();
-      syncTableMetadata();
     } catch (IOException ioe) {
       throw new HoodieIOException(ioe.getMessage(), ioe);
     } finally {

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadataWriter.java

@@ -32,7 +32,8 @@
  */
 public interface HoodieTableMetadataWriter extends Serializable, AutoCloseable {
 
-  void update(HoodieCommitMetadata commitMetadata, String instantTime);
+  // Update the metadata table due to a COMMIT operation
+  void update(HoodieCommitMetadata option, String instantTime);
 
   void update(HoodieCleanerPlan cleanerPlan, String instantTime);
 

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/HoodieTable.java

@@ -58,6 +58,7 @@
 import org.apache.hudi.common.table.view.TableFileSystemView.BaseFileOnlyView;
 import org.apache.hudi.common.table.view.TableFileSystemView.SliceView;
 import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.ValidationUtils;
 import org.apache.hudi.common.util.collection.Pair;
 import org.apache.hudi.config.HoodieWriteConfig;
 import org.apache.hudi.exception.HoodieException;
@@ -66,6 +67,7 @@
 import org.apache.hudi.exception.HoodieUpsertException;
 import org.apache.hudi.index.HoodieIndex;
 import org.apache.hudi.metadata.HoodieTableMetadata;
+import org.apache.hudi.metadata.HoodieTableMetadataWriter;
 import org.apache.hudi.table.action.HoodieWriteMetadata;
 import org.apache.hudi.table.action.bootstrap.HoodieBootstrapWriteMetadata;
 import org.apache.log4j.LogManager;
@@ -679,4 +681,9 @@ public HoodieEngineContext getContext() {
     // to engine context, and it ends up being null (as its not serializable and marked transient here).
     return context == null ? new HoodieLocalEngineContext(hadoopConfiguration.get()) : context;
   }
+
+  public Option<HoodieTableMetadataWriter> getMetadataWriter() {
+    ValidationUtils.checkArgument(!config.useFileListingMetadata(), "Metadata Table support not enabled in this Table");
+    return Option.empty();
+  }
 }

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/BaseActionExecutor.java

@@ -21,7 +21,11 @@
 import java.io.Serializable;
 
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hudi.avro.model.HoodieCleanMetadata;
+import org.apache.hudi.avro.model.HoodieRestoreMetadata;
+import org.apache.hudi.avro.model.HoodieRollbackMetadata;
 import org.apache.hudi.common.engine.HoodieEngineContext;
+import org.apache.hudi.common.model.HoodieCommitMetadata;
 import org.apache.hudi.common.model.HoodieRecordPayload;
 import org.apache.hudi.config.HoodieWriteConfig;
 import org.apache.hudi.table.HoodieTable;
@@ -46,4 +50,24 @@ public BaseActionExecutor(HoodieEngineContext context, HoodieWriteConfig config,
   }
 
   public abstract R execute();
+
+  protected final void syncTableMetadata(HoodieCommitMetadata metadata) {
+    // TODO: Should this be under its own transaction
+    table.getMetadataWriter().ifPresent(w -> w.update(metadata, instantTime));
+  }
+
+  protected final void syncTableMetadata(HoodieCleanMetadata metadata) {
+    // TODO: Should this be under its own transaction
+    table.getMetadataWriter().ifPresent(w -> w.update(metadata, instantTime));
+  }
+
+  protected final void syncTableMetadata(HoodieRollbackMetadata metadata) {
+    // TODO: Should this be under its own transaction
+    table.getMetadataWriter().ifPresent(w -> w.update(metadata, instantTime));
+  }
+
+  protected final void syncTableMetadata(HoodieRestoreMetadata metadata) {
+    // TODO: Should this be under its own transaction
+    table.getMetadataWriter().ifPresent(w -> w.update(metadata, instantTime));
+  }
 }

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/clean/BaseCleanActionExecutor.java

@@ -115,6 +115,8 @@ private HoodieCleanMetadata runClean(HoodieTable<T, I, K, O> table, HoodieInstan
           cleanStats
       );
 
+      syncTableMetadata(metadata);
+
       table.getActiveTimeline().transitionCleanInflightToComplete(inflightInstant,
           TimelineMetadataUtils.serializeCleanMetadata(metadata));
       LOG.info("Marked clean started on " + inflightInstant.getTimestamp() + " as complete");

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/commit/BaseCommitActionExecutor.java

@@ -175,10 +175,6 @@ protected void finalizeWrite(String instantTime, List<HoodieWriteStat> stats, Ho
     }
   }
 
-  protected void syncTableMetadata() {
-    // No Op
-  }
-
   /**
    * By default, return the writer schema in Write Config for storing in commit.
    */

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/restore/BaseRestoreActionExecutor.java

@@ -92,6 +92,7 @@ private HoodieRestoreMetadata finishRestore(Map<String, List<HoodieRollbackMetad
 
     HoodieRestoreMetadata restoreMetadata = TimelineMetadataUtils.convertRestoreMetadata(
         instantTime, durationInMs, instantsRolledBack, instantToMetadata);
+    syncTableMetadata(restoreMetadata);
     table.getActiveTimeline().saveAsComplete(new HoodieInstant(true, HoodieTimeline.RESTORE_ACTION, instantTime),
         TimelineMetadataUtils.serializeRestoreMetadata(restoreMetadata));
     LOG.info("Commits " + instantsRolledBack + " rollback is complete. Restored table to " + restoreInstantTime);

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/BaseRollbackActionExecutor.java

@@ -197,6 +197,11 @@ public List<HoodieRollbackStat> doRollbackAndGetStats() {
 
   protected void finishRollback(HoodieRollbackMetadata rollbackMetadata) throws HoodieIOException {
     try {
+      // TODO: Potential error here - rollbacks have already completed here so if the syncTableMetadata fails,
+      // metadata table will be left in an inconsistent state. This is because we do not use the inflight
+      // state for rollback.
+      syncTableMetadata(rollbackMetadata);
+
       table.getActiveTimeline().createNewInstant(
           new HoodieInstant(HoodieInstant.State.INFLIGHT, HoodieTimeline.ROLLBACK_ACTION, instantTime));
       table.getActiveTimeline().saveAsComplete(

hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/HoodieFlinkWriteClient.java

@@ -42,7 +42,6 @@
 import org.apache.hudi.exception.HoodieCommitException;
 import org.apache.hudi.exception.HoodieException;
 import org.apache.hudi.exception.HoodieIOException;
-import org.apache.hudi.exception.HoodieMetadataException;
 import org.apache.hudi.exception.HoodieNotSupportedException;
 import org.apache.hudi.index.FlinkHoodieIndex;
 import org.apache.hudi.index.HoodieIndex;
@@ -52,8 +51,6 @@
 import org.apache.hudi.io.FlinkMergeHandle;
 import org.apache.hudi.io.HoodieWriteHandle;
 import org.apache.hudi.io.MiniBatchHandle;
-import org.apache.hudi.metadata.FlinkHoodieBackedTableMetadataWriter;
-import org.apache.hudi.metadata.HoodieTableMetadataWriter;
 import org.apache.hudi.table.BulkInsertPartitioner;
 import org.apache.hudi.table.HoodieFlinkTable;
 import org.apache.hudi.table.HoodieTable;
@@ -389,16 +386,6 @@ protected HoodieTable<T, List<HoodieRecord<T>>, List<HoodieKey>, List<WriteStatu
     return getTableAndInitCtx(metaClient, operationType);
   }
 
-  @Override
-  public void syncTableMetadata() {
-    // Open up the metadata table again, for syncing
-    try (HoodieTableMetadataWriter writer = FlinkHoodieBackedTableMetadataWriter.create(hadoopConf, config, context)) {
-      LOG.info("Successfully synced to metadata table");
-    } catch (Exception e) {
-      throw new HoodieMetadataException("Error syncing to metadata table.", e);
-    }
-  }
-
   /**
    * Clean the write handles within a checkpoint interval.
    * All the handles should have been closed already.

hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/table/action/commit/BaseFlinkCommitActionExecutor.java

@@ -141,13 +141,14 @@ protected void commit(Option<Map<String, String>> extraMetadata, HoodieWriteMeta
     result.setWriteStats(writeStats);
     // Finalize write
     finalizeWrite(instantTime, writeStats, result);
-    syncTableMetadata();
     try {
       LOG.info("Committing " + instantTime + ", action Type " + getCommitActionType());
       HoodieActiveTimeline activeTimeline = table.getActiveTimeline();
       HoodieCommitMetadata metadata = CommitUtils.buildMetadata(writeStats, result.getPartitionToReplaceFileIds(),
           extraMetadata, operationType, getSchemaToStoreInCommit(), getCommitActionType());
 
+      syncTableMetadata(metadata);
+
       activeTimeline.saveAsComplete(new HoodieInstant(true, getCommitActionType(), instantTime),
           Option.of(metadata.toJsonString().getBytes(StandardCharsets.UTF_8)));
       LOG.info("Committed " + instantTime);

hudi-client/hudi-java-client/src/main/java/org/apache/hudi/table/action/commit/BaseJavaCommitActionExecutor.java

@@ -206,6 +206,8 @@ protected void commit(Option<Map<String, String>> extraMetadata, HoodieWriteMeta
       HoodieCommitMetadata metadata = CommitUtils.buildMetadata(writeStats, result.getPartitionToReplaceFileIds(),
           extraMetadata, operationType, getSchemaToStoreInCommit(), getCommitActionType());
 
+      syncTableMetadata(metadata);
+
       activeTimeline.saveAsComplete(new HoodieInstant(true, getCommitActionType(), instantTime),
           Option.of(metadata.toJsonString().getBytes(StandardCharsets.UTF_8)));
       LOG.info("Committed " + instantTime);

hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/SparkRDDWriteClient.java

@@ -42,10 +42,8 @@
 import org.apache.hudi.config.HoodieWriteConfig;
 import org.apache.hudi.exception.HoodieClusteringException;
 import org.apache.hudi.exception.HoodieCommitException;
-import org.apache.hudi.exception.HoodieMetadataException;
 import org.apache.hudi.index.HoodieIndex;
 import org.apache.hudi.index.SparkHoodieIndex;
-import org.apache.hudi.metadata.HoodieTableMetadataWriter;
 import org.apache.hudi.metadata.SparkHoodieBackedTableMetadataWriter;
 import org.apache.hudi.metrics.DistributedRegistry;
 import org.apache.hudi.table.BulkInsertPartitioner;
@@ -96,6 +94,11 @@ public SparkRDDWriteClient(HoodieEngineContext context, HoodieWriteConfig writeC
   public SparkRDDWriteClient(HoodieEngineContext context, HoodieWriteConfig writeConfig,
                              Option<EmbeddedTimelineService> timelineService) {
     super(context, writeConfig, timelineService);
+    if (config.useFileListingMetadata()) {
+      // If the metadata table does not exist, it should be bootstrapped here
+      // TODO: Check if we can remove this requirement - auto bootstrap on commit
+      SparkHoodieBackedTableMetadataWriter.create(context.getHadoopConf().get(), config, context);
+    }
   }
 
   /**
@@ -444,16 +447,6 @@ private HoodieTable<T, JavaRDD<HoodieRecord<T>>, JavaRDD<HoodieKey>, JavaRDD<Wri
     return table;
   }
 
-  @Override
-  public void syncTableMetadata() {
-    // Open up the metadata table again, for syncing
-    try (HoodieTableMetadataWriter writer = SparkHoodieBackedTableMetadataWriter.create(hadoopConf, config, context)) {
-      LOG.info("Successfully synced to metadata table");
-    } catch (Exception e) {
-      throw new HoodieMetadataException("Error syncing to metadata table.", e);
-    }
-  }
-
   @Override
   protected void preCommit(String instantTime, HoodieCommitMetadata metadata) {
     // Create a Hoodie table after startTxn which encapsulated the commits and files visible.

hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/metadata/SparkHoodieBackedTableMetadataWriter.java

@@ -114,8 +114,14 @@ protected void commit(List<HoodieRecord> records, String partitionName, String i
       // trigger cleaning, compaction, with suffixes based on the same instant time. This ensures that any future
       // delta commits synced over will not have an instant time lesser than the last completed instant on the
       // metadata table.
-      if (writeClient.scheduleCompactionAtInstant(instantTime + "001", Option.empty())) {
-        writeClient.compact(instantTime + "001");
+      // TODO: This does not work with parallel operations because the operations having a larger timestamp
+      // may have completed earlier and hence instantTime is not the latest commit.
+      try {
+        if (writeClient.scheduleCompactionAtInstant(instantTime + "001", Option.empty())) {
+          writeClient.compact(instantTime + "001");
+        }
+      } catch (IllegalArgumentException e) {
+        LOG.info("Ignoring error in compaction: " + e);
       }
       writeClient.clean(instantTime + "002");
     }

hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/HoodieSparkTable.java

@@ -18,6 +18,7 @@
 
 package org.apache.hudi.table;
 
+import org.apache.hadoop.fs.Path;
 import org.apache.hudi.client.WriteStatus;
 import org.apache.hudi.client.common.HoodieSparkEngineContext;
 import org.apache.hudi.common.engine.HoodieEngineContext;
@@ -29,9 +30,12 @@
 import org.apache.hudi.common.util.Option;
 import org.apache.hudi.config.HoodieWriteConfig;
 import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.exception.HoodieMetadataException;
 import org.apache.hudi.index.HoodieIndex;
 import org.apache.hudi.index.SparkHoodieIndex;
-
+import org.apache.hudi.metadata.HoodieTableMetadata;
+import org.apache.hudi.metadata.HoodieTableMetadataWriter;
+import org.apache.hudi.metadata.SparkHoodieBackedTableMetadataWriter;
 import org.apache.spark.api.java.JavaRDD;
 
 public abstract class HoodieSparkTable<T extends HoodieRecordPayload>
@@ -66,4 +70,21 @@ public static <T extends HoodieRecordPayload> HoodieSparkTable<T> create(HoodieW
   protected HoodieIndex<T, JavaRDD<HoodieRecord<T>>, JavaRDD<HoodieKey>, JavaRDD<WriteStatus>> getIndex(HoodieWriteConfig config, HoodieEngineContext context) {
     return SparkHoodieIndex.createIndex(config);
   }
+
+  @Override
+  public Option<HoodieTableMetadataWriter> getMetadataWriter() {
+    if (!config.useFileListingMetadata()) {
+      return Option.empty();
+    }
+
+    try {
+      if (!metaClient.getFs().exists(new Path(HoodieTableMetadata.getMetadataTableBasePath(metaClient.getBasePath())))) {
+        return Option.empty();
+      }
+    } catch (Exception e) {
+      throw new HoodieMetadataException("Could not create metadata table writer", e);
+    }
+
+    return Option.of(SparkHoodieBackedTableMetadataWriter.create(context.getHadoopConf().get(), config, context));
+  }
 }