[HUDI-1502] MOR rollback and restore support for metadata sync

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieAppendHandle.java

@@ -164,6 +164,7 @@ private void init(HoodieRecord record) {
         // Since the actual log file written to can be different based on when rollover happens, we use the
         // base file to denote some log appends happened on a slice. writeToken will still fence concurrent
         // writers.
+        // TODO? are these sufficient?
         createMarkerFile(partitionPath, FSUtils.makeDataFileName(baseInstantTime, writeToken, fileId, hoodieTable.getBaseFileExtension()));
 
         this.writer = createLogWriter(fileSlice, baseInstantTime);

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

@@ -32,6 +32,7 @@
 import org.apache.hudi.config.HoodieWriteConfig;
 import org.apache.hudi.exception.HoodieIOException;
 import org.apache.hudi.table.HoodieTable;
+
 import org.apache.log4j.LogManager;
 import org.apache.log4j.Logger;
 
@@ -53,9 +54,9 @@ public abstract class AbstractMarkerBasedRollbackStrategy<T extends HoodieRecord
 
   protected final HoodieWriteConfig config;
 
-  private final String basePath;
+  protected final String basePath;
 
-  private final String instantTime;
+  protected final String instantTime;
 
   public AbstractMarkerBasedRollbackStrategy(HoodieTable<T, I, K, O> table, HoodieEngineContext context, HoodieWriteConfig config, String instantTime) {
     this.table = table;
@@ -90,6 +91,7 @@ protected HoodieRollbackStat undoAppend(String appendBaseFilePath, HoodieInstant
     String fileId = FSUtils.getFileIdFromFilePath(baseFilePathForAppend);
     String baseCommitTime = FSUtils.getCommitTime(baseFilePathForAppend.getName());
     String partitionPath = FSUtils.getRelativePartitionPath(new Path(basePath), new Path(basePath, appendBaseFilePath).getParent());
+    final Map<FileStatus, Long> probableLogFileMap = getProbableFileSizeMap(partitionPath, baseCommitTime);
 
     HoodieLogFormat.Writer writer = null;
     try {
@@ -129,9 +131,23 @@ protected HoodieRollbackStat undoAppend(String appendBaseFilePath, HoodieInstant
           1L);
     }
 
-    return HoodieRollbackStat.newBuilder()
+    HoodieRollbackStat.Builder builder = HoodieRollbackStat.newBuilder()
         .withPartitionPath(partitionPath)
-        .withRollbackBlockAppendResults(filesToNumBlocksRollback)
-        .build();
+        .withRollbackBlockAppendResults(filesToNumBlocksRollback);
+    if (probableLogFileMap != null) {
+      builder.withProbableLogFileToSizeMap(probableLogFileMap);
+    }
+    return builder.build();
+  }
+
+  /**
+   * Returns probable log files for the respective baseCommitTime to assist in metadata table syncing.
+   * @param partitionPath partition path of interest
+   * @param baseCommitTime base commit time of interest
+   * @return Map<FileStatus, File size>
+   * @throws IOException
+   */
+  protected Map<FileStatus, Long> getProbableFileSizeMap(String partitionPath, String baseCommitTime) throws IOException {
+    return null;
   }
 }

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

@@ -28,7 +28,6 @@
 import org.apache.log4j.LogManager;
 import org.apache.log4j.Logger;
 
-import java.io.IOException;
 import java.util.ArrayList;
 import java.util.List;
 
@@ -57,7 +56,7 @@ public BaseMergeOnReadRollbackActionExecutor(HoodieEngineContext context,
   }
 
   @Override
-  protected List<HoodieRollbackStat> executeRollback() throws IOException {
+  protected List<HoodieRollbackStat> executeRollback() {
     HoodieTimer rollbackTimer = new HoodieTimer();
     rollbackTimer.startTimer();
 

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

@@ -74,14 +74,16 @@ static HoodieRollbackStat mergeRollbackStat(HoodieRollbackStat stat1, HoodieRoll
     final List<String> successDeleteFiles = new ArrayList<>();
     final List<String> failedDeleteFiles = new ArrayList<>();
     final Map<FileStatus, Long> commandBlocksCount = new HashMap<>();
-    final List<FileStatus> filesToRollback = new ArrayList<>();
+    final Map<FileStatus, Long> probableLogFileMap = new HashMap<>();
     Option.ofNullable(stat1.getSuccessDeleteFiles()).ifPresent(successDeleteFiles::addAll);
     Option.ofNullable(stat2.getSuccessDeleteFiles()).ifPresent(successDeleteFiles::addAll);
     Option.ofNullable(stat1.getFailedDeleteFiles()).ifPresent(failedDeleteFiles::addAll);
     Option.ofNullable(stat2.getFailedDeleteFiles()).ifPresent(failedDeleteFiles::addAll);
     Option.ofNullable(stat1.getCommandBlocksCount()).ifPresent(commandBlocksCount::putAll);
     Option.ofNullable(stat2.getCommandBlocksCount()).ifPresent(commandBlocksCount::putAll);
-    return new HoodieRollbackStat(stat1.getPartitionPath(), successDeleteFiles, failedDeleteFiles, commandBlocksCount);
+    Option.ofNullable(stat1.getProbableLogFileToSizeMap()).ifPresent(probableLogFileMap::putAll);
+    Option.ofNullable(stat2.getProbableLogFileToSizeMap()).ifPresent(probableLogFileMap::putAll);
+    return new HoodieRollbackStat(stat1.getPartitionPath(), successDeleteFiles, failedDeleteFiles, commandBlocksCount, probableLogFileMap);
   }
 
   /**

hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/rollback/ListingBasedRollbackHelper.java

@@ -98,8 +98,9 @@ public List<HoodieRollbackStat> collectRollbackStats(HoodieEngineContext context
    * @return stats collected with or w/o actual deletions.
    */
   JavaPairRDD<String, HoodieRollbackStat> maybeDeleteAndCollectStats(HoodieEngineContext context, HoodieInstant instantToRollback, List<ListingBasedRollbackRequest> rollbackRequests,
-                                                                     int sparkPartitions, boolean doDelete) {
+      int sparkPartitions, boolean doDelete) {
     JavaSparkContext jsc = HoodieSparkEngineContext.getSparkContext(context);
+
     return jsc.parallelize(rollbackRequests, sparkPartitions).mapToPair(rollbackRequest -> {
       switch (rollbackRequest.getType()) {
         case DELETE_DATA_FILES_ONLY: {
@@ -116,14 +117,31 @@ JavaPairRDD<String, HoodieRollbackStat> maybeDeleteAndCollectStats(HoodieEngineC
                   .withDeletedFileResults(filesToDeletedStatus).build());
         }
         case APPEND_ROLLBACK_BLOCK: {
+          // collect all log files that is supposed to be deleted with this rollback
+          String baseCommit = rollbackRequest.getLatestBaseInstant().get();
+          SerializablePathFilter filter = (path) -> {
+            if (FSUtils.isLogFile(path)) {
+              // Since the baseCommitTime is the only commit for new log files, it's okay here
+              String fileCommitTime = FSUtils.getBaseCommitTimeFromLogPath(path);
+              return baseCommit.equals(fileCommitTime);
+            }
+            return false;
+          };
+
+          final Map<FileStatus, Long> probableLogFileMap = new HashMap<>();
+          FileSystem fs = metaClient.getFs();
+          FileStatus[] probableLogFiles = fs.listStatus(FSUtils.getPartitionPath(config.getBasePath(), rollbackRequest.getPartitionPath()), filter);
+          for (FileStatus fileStatus : probableLogFiles) {
+            probableLogFileMap.put(fileStatus, fileStatus.getLen());
+          }
+
           Writer writer = null;
           try {
             writer = HoodieLogFormat.newWriterBuilder()
                 .onParentPath(FSUtils.getPartitionPath(metaClient.getBasePath(), rollbackRequest.getPartitionPath()))
                 .withFileId(rollbackRequest.getFileId().get())
                 .overBaseCommit(rollbackRequest.getLatestBaseInstant().get()).withFs(metaClient.getFs())
                 .withFileExtension(HoodieLogFile.DELTA_EXTENSION).build();
-
             // generate metadata
             if (doDelete) {
               Map<HeaderMetadataType, String> header = generateHeader(instantToRollback.getTimestamp());
@@ -151,15 +169,15 @@ JavaPairRDD<String, HoodieRollbackStat> maybeDeleteAndCollectStats(HoodieEngineC
           );
           return new Tuple2<>(rollbackRequest.getPartitionPath(),
               HoodieRollbackStat.newBuilder().withPartitionPath(rollbackRequest.getPartitionPath())
-                  .withRollbackBlockAppendResults(filesToNumBlocksRollback).build());
+                  .withRollbackBlockAppendResults(filesToNumBlocksRollback)
+                  .withProbableLogFileToSizeMap(probableLogFileMap).build());
         }
         default:
           throw new IllegalStateException("Unknown Rollback action " + rollbackRequest);
       }
     });
   }
 
-
   /**
    * Common method used for cleaning out base files under a partition path during rollback of a set of commits.
    */

hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/rollback/SparkMarkerBasedRollbackStrategy.java

@@ -22,6 +22,7 @@
 import org.apache.hudi.client.common.HoodieSparkEngineContext;
 import org.apache.hudi.common.HoodieRollbackStat;
 import org.apache.hudi.common.engine.HoodieEngineContext;
+import org.apache.hudi.common.fs.FSUtils;
 import org.apache.hudi.common.model.HoodieKey;
 import org.apache.hudi.common.model.HoodieRecord;
 import org.apache.hudi.common.model.HoodieRecordPayload;
@@ -31,17 +32,25 @@
 import org.apache.hudi.exception.HoodieRollbackException;
 import org.apache.hudi.table.HoodieTable;
 import org.apache.hudi.table.MarkerFiles;
+import org.apache.hudi.table.action.rollback.ListingBasedRollbackHelper.SerializablePathFilter;
 
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
 import org.apache.spark.api.java.JavaRDD;
 import org.apache.spark.api.java.JavaSparkContext;
 
+import java.io.IOException;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 
 import scala.Tuple2;
 
 @SuppressWarnings("checkstyle:LineLength")
 public class SparkMarkerBasedRollbackStrategy<T extends HoodieRecordPayload> extends AbstractMarkerBasedRollbackStrategy<T, JavaRDD<HoodieRecord<T>>, JavaRDD<HoodieKey>, JavaRDD<WriteStatus>> {
-  public SparkMarkerBasedRollbackStrategy(HoodieTable<T, JavaRDD<HoodieRecord<T>>, JavaRDD<HoodieKey>, JavaRDD<WriteStatus>> table, HoodieEngineContext context, HoodieWriteConfig config, String instantTime) {
+
+  public SparkMarkerBasedRollbackStrategy(HoodieTable<T, JavaRDD<HoodieRecord<T>>, JavaRDD<HoodieKey>, JavaRDD<WriteStatus>> table, HoodieEngineContext context, HoodieWriteConfig config,
+      String instantTime) {
     super(table, context, config, instantTime);
   }
 
@@ -75,4 +84,23 @@ public List<HoodieRollbackStat> execute(HoodieInstant instantToRollback) {
       throw new HoodieRollbackException("Error rolling back using marker files written for " + instantToRollback, e);
     }
   }
+
+  protected Map<FileStatus, Long> getProbableFileSizeMap(String partitionPath, String baseCommitTime) throws IOException {
+    // collect all log files that is supposed to be deleted with this rollback
+    SerializablePathFilter filter = (path) -> {
+      if (FSUtils.isLogFile(path)) {
+        String fileCommitTime = FSUtils.getBaseCommitTimeFromLogPath(path);
+        return baseCommitTime.equals(fileCommitTime);
+      }
+      return false;
+    };
+
+    final Map<FileStatus, Long> probableLogFileMap = new HashMap<>();
+    FileSystem fs = table.getMetaClient().getFs();
+    FileStatus[] toBeDeleted = fs.listStatus(FSUtils.getPartitionPath(config.getBasePath(), partitionPath), filter);
+    for (FileStatus fileStatus : toBeDeleted) {
+      probableLogFileMap.put(fileStatus, fileStatus.getLen());
+    }
+    return probableLogFileMap;
+  }
 }

hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/metadata/TestHoodieBackedMetadata.java

@@ -78,6 +78,7 @@
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
 public class TestHoodieBackedMetadata extends HoodieClientTestHarness {
+
   private static final Logger LOG = LogManager.getLogger(TestHoodieBackedMetadata.class);
 
   @TempDir
@@ -259,13 +260,11 @@ public void testTableOperations(HoodieTableType tableType) throws Exception {
   /**
    * Test rollback of various table operations sync to Metadata Table correctly.
    */
-  //@ParameterizedTest
-  //@EnumSource(HoodieTableType.class)
-  //public void testRollbackOperations(HoodieTableType tableType) throws Exception {
-  @Test
-  public void testRollbackOperations() throws Exception {
+  @ParameterizedTest
+  @EnumSource(HoodieTableType.class)
+  public void testRollbackOperations(HoodieTableType tableType) throws Exception {
     //FIXME(metadata): This is broken for MOR, until HUDI-1434 is fixed
-    init(HoodieTableType.COPY_ON_WRITE);
+    init(tableType);
     HoodieSparkEngineContext engineContext = new HoodieSparkEngineContext(jsc);
 
     try (SparkRDDWriteClient client = new SparkRDDWriteClient(engineContext, getWriteConfig(true, true))) {
@@ -369,13 +368,12 @@ public void testRollbackOperations() throws Exception {
   }
 
   /**
-   * Test when syncing rollback to metadata if the commit being rolled back has not been synced that essentially a no-op
-   * occurs to metadata.
-   * @throws Exception
+   * Test when syncing rollback to metadata if the commit being rolled back has not been synced that essentially a no-op occurs to metadata.
    */
-  @Test
-  public void testRollbackUnsyncedCommit() throws Exception {
-    init(HoodieTableType.COPY_ON_WRITE);
+  @ParameterizedTest
+  @EnumSource(HoodieTableType.class)
+  public void testRollbackUnsyncedCommit(HoodieTableType tableType) throws Exception {
+    init(tableType);
     HoodieSparkEngineContext engineContext = new HoodieSparkEngineContext(jsc);
 
     try (SparkRDDWriteClient client = new SparkRDDWriteClient(engineContext, getWriteConfig(true, true))) {
@@ -517,8 +515,7 @@ public void testSync(HoodieTableType tableType) throws Exception {
   }
 
   /**
-   * Instants on Metadata Table should be archived as per config.
-   * Metadata Table should be automatically compacted as per config.
+   * Instants on Metadata Table should be archived as per config. Metadata Table should be automatically compacted as per config.
    */
   @Test
   public void testCleaningArchivingAndCompaction() throws Exception {
@@ -741,8 +738,6 @@ public void testMetadataOutOfSync() throws Exception {
 
   /**
    * Validate the metadata tables contents to ensure it matches what is on the file system.
-   *
-   * @throws IOException
    */
   private void validateMetadata(SparkRDDWriteClient client) throws IOException {
     HoodieWriteConfig config = client.getConfig();
@@ -794,7 +789,19 @@ private void validateMetadata(SparkRDDWriteClient client) throws IOException {
         if ((fsFileNames.size() != metadataFilenames.size()) || (!fsFileNames.equals(metadataFilenames))) {
           LOG.info("*** File system listing = " + Arrays.toString(fsFileNames.toArray()));
           LOG.info("*** Metadata listing = " + Arrays.toString(metadataFilenames.toArray()));
+
+          for (String fileName : fsFileNames) {
+            if (!metadataFilenames.contains(fileName)) {
+              System.out.println(partition + "FsFilename " + fileName + " not found in Meta data");
+            }
+          }
+          for (String fileName : metadataFilenames) {
+            if (!fsFileNames.contains(fileName)) {
+              System.out.println(partition + "Metadata file " + fileName + " not found in original FS");
+            }
+          }
         }
+
         assertEquals(fsFileNames.size(), metadataFilenames.size(), "Files within partition " + partition + " should match");
         assertTrue(fsFileNames.equals(metadataFilenames), "Files within partition " + partition + " should match");
 

hudi-common/src/main/avro/HoodieRollbackMetadata.avsc

@@ -31,18 +31,22 @@
             {"name": "partitionPath", "type": "string"},
             {"name": "successDeleteFiles", "type": {"type": "array", "items": "string"}},
             {"name": "failedDeleteFiles", "type": {"type": "array", "items": "string"}},
-            {"name": "appendFiles", "type": {
+            {"name": "rollbackLogFiles", "type": {
                 "type": "map",
-                "doc": "Files to which append blocks were written",
                 "values": {
                     "type": "long",
                     "doc": "Size of this file in bytes"
                 }
+            }},
+            {"name": "writtenLogFiles", "type": {
+                "type": "map",
+                 "values": {
+                    "type": "long",
+                    "doc": "Size of this file in bytes"
+                }
             }}
         ]
-     }
-     }
-     },
+     }}},
      {
         "name":"version",
         "type":["int", "null"],