resolve merge conflicts for HUDI-9667

Author: rahil-c

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

@@ -70,7 +70,7 @@ public RestorePlanActionExecutor(HoodieEngineContext context,
   public Option<HoodieRestorePlan> execute() {
     final HoodieInstant restoreInstant = instantGenerator.createNewInstant(HoodieInstant.State.REQUESTED, HoodieTimeline.RESTORE_ACTION, instantTime);
     HoodieTableMetaClient metaClient = table.getMetaClient();
-    try (CompletionTimeQueryView completionTimeQueryView = metaClient.getTableFormat().getTimelineFactory().createCompletionTimeQueryView(metaClient)) {
+    try (CompletionTimeQueryView completionTimeQueryView = metaClient.getTimelineLayout().getTimelineFactory().createCompletionTimeQueryView(metaClient)) {
       // Get all the commits on the timeline after the provided commit time
       // rollback pending clustering instants first before other instants (See HUDI-3362)
       List<HoodieInstant> pendingClusteringInstantsToRollback = table.getActiveTimeline().filterPendingReplaceOrClusteringTimeline()

hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestHoodieFileSystemViews.java

@@ -190,7 +190,7 @@ private void assertFileSystemViews(HoodieWriteConfig writeConfig, boolean enable
     }
   }
 
-  private void assertForFSVEquality(HoodieTableFileSystemView fsv1, HoodieTableFileSystemView fsv2, boolean enableMdt) {
+  public static void assertForFSVEquality(HoodieTableFileSystemView fsv1, HoodieTableFileSystemView fsv2, boolean enableMdt) {
     List<String> allPartitionNames = Arrays.asList(DEFAULT_FIRST_PARTITION_PATH, DEFAULT_SECOND_PARTITION_PATH, DEFAULT_THIRD_PARTITION_PATH);
     fsv1.loadPartitions(allPartitionNames);
     if (enableMdt) {
@@ -239,7 +239,7 @@ static void assertBaseFileEquality(HoodieBaseFile baseFile1, HoodieBaseFile base
     assertEquals(baseFile1.getFileSize(), baseFile2.getFileSize());
   }
 
-  private void assertFileSliceListEquality(List<FileSlice> fileSlices1, List<FileSlice> fileSlices2) {
+  static void assertFileSliceListEquality(List<FileSlice> fileSlices1, List<FileSlice> fileSlices2) {
     assertEquals(fileSlices1.size(), fileSlices1.size());
     Map<Pair<String, String>, FileSlice> fileNameToFileSliceMap1 = new HashMap<>();
     fileSlices1.forEach(entry -> {
@@ -255,7 +255,7 @@ private void assertFileSliceListEquality(List<FileSlice> fileSlices1, List<FileS
     });
   }
 
-  private void assertFileSliceEquality(FileSlice fileSlice1, FileSlice fileSlice2) {
+  static void assertFileSliceEquality(FileSlice fileSlice1, FileSlice fileSlice2) {
     assertEquals(fileSlice1.getBaseFile().isPresent(), fileSlice2.getBaseFile().isPresent());
     if (fileSlice1.getBaseFile().isPresent()) {
       assertBaseFileEquality(fileSlice1.getBaseFile().get(), fileSlice2.getBaseFile().get());

hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/testutils/HoodieClientTestBase.java

@@ -623,8 +623,7 @@ protected void updateBatchWithoutCommit(String newCommitTime, List<HoodieRecord>
         .withAutoUpgradeVersion(false)
         .withWriteTableVersion(tableVersion.versionCode())
         .withMetadataConfig(HoodieMetadataConfig.newBuilder()
-            .withStreamingWriteEnabled(tableVersion.greaterThanOrEquals(HoodieTableVersion.EIGHT))
-            .build())
+        .build())
         .build();
 
     try (SparkRDDWriteClient client = getHoodieWriteClient(hoodieWriteConfig)) {

hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/client/functional/TestSavepointRestoreMergeOnRead.java

@@ -24,6 +24,7 @@
 import org.apache.hudi.common.config.HoodieMetadataConfig;
 import org.apache.hudi.common.fs.FSUtils;
 import org.apache.hudi.common.function.SerializableFunctionUnchecked;
+import org.apache.hudi.common.model.HoodieCommitMetadata;
 import org.apache.hudi.common.model.HoodieFailedWritesCleaningPolicy;
 import org.apache.hudi.common.model.HoodieRecord;
 import org.apache.hudi.common.model.HoodieTableType;
@@ -42,6 +43,7 @@
 import org.apache.hudi.metadata.HoodieTableMetadata;
 import org.apache.hudi.storage.StoragePath;
 import org.apache.hudi.storage.StoragePathFilter;
+import org.apache.hudi.table.action.HoodieWriteMetadata;
 import org.apache.hudi.testutils.HoodieClientTestBase;
 
 import org.apache.spark.api.java.JavaRDD;
@@ -62,8 +64,8 @@
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
+import static org.apache.hudi.client.functional.TestHoodieFileSystemViews.assertForFSVEquality;
 import static org.apache.hudi.common.table.timeline.HoodieTimeline.DELTA_COMMIT_ACTION;
-import static org.apache.hudi.functional.TestHoodieFileSystemViews.assertForFSVEquality;
 import static org.apache.hudi.testutils.Assertions.assertNoWriteErrors;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
@@ -249,7 +251,6 @@ public void testRestoreWithFileGroupCreatedWithDeltaCommits(HoodieTableVersion t
         .withAutoUpgradeVersion(false)
         .withWriteTableVersion(tableVersion.versionCode())
         .withMetadataConfig(HoodieMetadataConfig.newBuilder()
-            .withStreamingWriteEnabled(tableVersion.greaterThanOrEquals(HoodieTableVersion.EIGHT))
             .build())
         .build();
 
@@ -420,7 +421,7 @@ void testCleaningPendingCompaction(HoodieTableVersion tableVersion) throws Excep
    * No files will be cleaned up. Only rollback log appends.
    */
   @ParameterizedTest
-  @EnumSource(value = HoodieTableVersion.class, names = {"SIX", "NINE"})
+  @EnumSource(value = HoodieTableVersion.class, names = {"SIX", "EIGHT"})
   void testCleaningCompletedRollback(HoodieTableVersion tableVersion) throws Exception {
     HoodieWriteConfig hoodieWriteConfig = getHoodieWriteConfigAndInitializeTable(HoodieCompactionConfig.newBuilder()
         .withMaxNumDeltaCommitsBeforeCompaction(3) // the 3rd delta_commit triggers compaction
@@ -456,7 +457,7 @@ void testCleaningCompletedRollback(HoodieTableVersion tableVersion) throws Excep
 
       assertRowNumberEqualsTo(20);
       // write a delta_commit but does not commit
-      updateBatchWithoutCommit(WriteClientTestUtils.createNewInstantTime(),
+      updateBatchWithoutCommit(client.createNewInstantTime(),
           Objects.requireNonNull(baseRecordsToUpdate, "The records to update should not be null"), tableVersion);
       // rollback the delta_commit
       assertTrue(writeClient.rollbackFailedWrites(metaClient), "The last delta_commit should be rolled back");
@@ -485,7 +486,7 @@ void rollbackWithAsyncServices_compactionCompletesDuringCommit() {
       // Run compaction while delta-commit is in-flight
       Option<String> compactionInstant = client.scheduleCompaction(Option.empty());
       HoodieWriteMetadata result = client.compact(compactionInstant.get());
-      client.commitCompaction(compactionInstant.get(), result, Option.empty());
+      client.commitCompaction(compactionInstant.get(), (HoodieCommitMetadata) result.getCommitMetadata().get(), Option.empty());
       // commit the inflight delta commit
       client.commit(inflightCommit, writeStatus);
 
@@ -522,7 +523,7 @@ void rollbackWithAsyncServices_commitCompletesDuringCompaction() {
       // commit the inflight delta commit
       client.commit(inflightCommit, writeStatus);
       // commit the compaction instant after the delta commit
-      client.commitCompaction(compactionInstant.get(), result, Option.empty());
+      client.commitCompaction(compactionInstant.get(), (HoodieCommitMetadata) result.getCommitMetadata().get(), Option.empty());
 
       client.savepoint(inflightCommit, "user1", "Savepoint for commit that completed during compaction");
 
@@ -543,7 +544,7 @@ void rollbackWithAsyncServices_commitCompletesDuringCompaction() {
   }
 
   @ParameterizedTest
-  @EnumSource(value = HoodieTableVersion.class, names = {"SIX", "NINE"})
+  @EnumSource(value = HoodieTableVersion.class, names = {"SIX", "EIGHT"})
   void rollbackWithAsyncServices_commitStartsAndFinishesDuringCompaction(HoodieTableVersion tableVersion) {
     HoodieWriteConfig hoodieWriteConfig = getHoodieWriteConfigWithCompactionAndConcurrencyControl(tableVersion);
     try (SparkRDDWriteClient client = getHoodieWriteClient(hoodieWriteConfig)) {
@@ -560,7 +561,7 @@ void rollbackWithAsyncServices_commitStartsAndFinishesDuringCompaction(HoodieTab
       // commit the inflight delta commit
       client.commit(inflightCommit, writeStatus);
       // commit the compaction instant after the delta commit
-      client.commitCompaction(compactionInstant.get(), result, Option.empty());
+      client.commitCompaction(compactionInstant.get(), (HoodieCommitMetadata) result.getCommitMetadata().get(), Option.empty());
 
       client.savepoint(inflightCommit, "user1", "Savepoint for commit that completed during compaction");
 
@@ -581,7 +582,7 @@ void rollbackWithAsyncServices_commitStartsAndFinishesDuringCompaction(HoodieTab
   }
 
   @ParameterizedTest
-  @EnumSource(value = HoodieTableVersion.class, names = {"SIX", "NINE"})
+  @EnumSource(value = HoodieTableVersion.class, names = {"SIX", "EIGHT"})
   void testMissingFileDoesNotFallRestore(HoodieTableVersion tableVersion) throws Exception {
     HoodieWriteConfig hoodieWriteConfig = getHoodieWriteConfigAndInitializeTable(HoodieCompactionConfig.newBuilder()
         .withMaxNumDeltaCommitsBeforeCompaction(4)
@@ -657,7 +658,6 @@ private HoodieWriteConfig getHoodieWriteConfigWithCompactionAndConcurrencyContro
         .withAutoUpgradeVersion(false)
         .withWriteTableVersion(tableVersion.versionCode())
         .withMetadataConfig(HoodieMetadataConfig.newBuilder()
-            .withStreamingWriteEnabled(tableVersion.greaterThanOrEquals(HoodieTableVersion.EIGHT))
             .build())
         .withProps(Collections.singletonMap(HoodieCompactionConfig.PARQUET_SMALL_FILE_LIMIT.key(), "0"))
         .build();
@@ -677,9 +677,9 @@ private void validateFilesMetadata(HoodieWriteConfig writeConfig) {
     HoodieTableFileSystemView metadataBasedView = (HoodieTableFileSystemView) FileSystemViewManager
         .createViewManager(context, writeConfig.getMetadataConfig(), viewStorageConfig, writeConfig.getCommonConfig(),
             (SerializableFunctionUnchecked<HoodieTableMetaClient, HoodieTableMetadata>) v1 ->
-                metaClient.getTableFormat().getMetadataFactory().create(context, metaClient.getStorage(), writeConfig.getMetadataConfig(), writeConfig.getBasePath()))
+                HoodieTableMetadata.create(context, metaClient.getStorage(), writeConfig.getMetadataConfig(), writeConfig.getBasePath()))
         .getFileSystemView(basePath);
-    assertForFSVEquality(fileListingBasedView, metadataBasedView, true, Option.empty());
+    assertForFSVEquality(fileListingBasedView, metadataBasedView, true);
   }
 
   private String upsertBatch(SparkRDDWriteClient client, List<HoodieRecord> baseRecordsToUpdate) throws IOException {
@@ -698,7 +698,6 @@ private void compactWithoutCommit(String compactionInstantTime, HoodieTableVersi
         .withAutoUpgradeVersion(false)
         .withWriteTableVersion(tableVersion.versionCode())
         .withMetadataConfig(HoodieMetadataConfig.newBuilder()
-            .withStreamingWriteEnabled(tableVersion.greaterThanOrEquals(HoodieTableVersion.EIGHT))
             .build())
         .build();
 
@@ -726,7 +725,6 @@ private HoodieWriteConfig getHoodieWriteConfigAndInitializeTable(HoodieCompactio
         .withAutoUpgradeVersion(false)
         .withWriteTableVersion(tableVersion.versionCode())
         .withMetadataConfig(HoodieMetadataConfig.newBuilder()
-            .withStreamingWriteEnabled(tableVersion.greaterThanOrEquals(HoodieTableVersion.EIGHT))
             .build())
         .build();
     initMetaClient(HoodieTableType.MERGE_ON_READ, hoodieWriteConfig.getProps());