(hudi) branch master updated: [HUDI-7710] Use compaction.requested during conflict resolution (#11151)

[jo...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

jonvex pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git


The following commit(s) were added to refs/heads/master by this push:
     new bc07e3ed252 [HUDI-7710] Use compaction.requested during conflict resolution (#11151)
bc07e3ed252 is described below

commit bc07e3ed252f0334e98f1c4daa2928a3da05a088
Author: Lin Liu <14...@users.noreply.github.com>
AuthorDate: Mon May 6 13:38:26 2024 -0700

    [HUDI-7710] Use compaction.requested during conflict resolution (#11151)
    
    * [HUDI-7710] Replace compaction.inflight with compaction.requested during conflict resolution
    
    * Remove an unused import
    
    * Replace in ConcurrentOperation class instead
    
    * Use MOR table
    
    * Address some comments
    
    * Remove an unnecessary change
---
 .../client/transaction/ConcurrentOperation.java    |  4 +
 ...urrentFileWritesConflictResolutionStrategy.java |  2 -
 .../TestConflictResolutionStrategyUtil.java        |  6 +-
 ...itesConflictResolutionStrategyWithMORTable.java | 89 ++++++++++++++++++++++
 4 files changed, 96 insertions(+), 5 deletions(-)

diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/ConcurrentOperation.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/ConcurrentOperation.java
index 2655cc2b535..5c02592d329 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/ConcurrentOperation.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/ConcurrentOperation.java
@@ -60,6 +60,10 @@ public class ConcurrentOperation {
   private Set<Pair<String, String>> mutatedPartitionAndFileIds = Collections.emptySet();
 
   public ConcurrentOperation(HoodieInstant instant, HoodieTableMetaClient metaClient) throws IOException {
+    // Replace compaction.inflight to compaction.request since inflight does not contain compaction plan.
+    if (instant.getAction().equals(COMPACTION_ACTION) && instant.getState().equals(HoodieInstant.State.INFLIGHT)) {
+      instant = new HoodieInstant(HoodieInstant.State.REQUESTED, COMPACTION_ACTION, instant.getTimestamp());
+    }
     this.metadataWrapper = new HoodieMetadataWrapper(MetadataConversionUtils.createMetaWrapper(instant, metaClient));
     this.commitMetadataOption = Option.empty();
     this.actionState = instant.getState().name();
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/SimpleConcurrentFileWritesConflictResolutionStrategy.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/SimpleConcurrentFileWritesConflictResolutionStrategy.java
index 8bef9e49152..ce16e14af22 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/SimpleConcurrentFileWritesConflictResolutionStrategy.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/SimpleConcurrentFileWritesConflictResolutionStrategy.java
@@ -38,7 +38,6 @@ import java.util.HashSet;
 import java.util.Set;
 import java.util.stream.Stream;
 
-import static org.apache.hudi.common.table.timeline.HoodieInstant.State.REQUESTED;
 import static org.apache.hudi.common.table.timeline.HoodieTimeline.COMPACTION_ACTION;
 import static org.apache.hudi.common.table.timeline.HoodieTimeline.REPLACE_COMMIT_ACTION;
 
@@ -69,7 +68,6 @@ public class SimpleConcurrentFileWritesConflictResolutionStrategy
         .getTimelineOfActions(CollectionUtils.createSet(REPLACE_COMMIT_ACTION, COMPACTION_ACTION))
         .findInstantsAfter(currentInstant.getTimestamp())
         .filterInflightsAndRequested()
-        .filter(i -> (!i.getAction().equals(COMPACTION_ACTION)) || i.getState().equals(REQUESTED))
         .getInstantsAsStream();
     return Stream.concat(completedCommitsInstantStream, compactionAndClusteringPendingTimeline);
   }
diff --git a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/client/transaction/TestConflictResolutionStrategyUtil.java b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/client/transaction/TestConflictResolutionStrategyUtil.java
index 2569ff434eb..63f251b0b41 100644
--- a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/client/transaction/TestConflictResolutionStrategyUtil.java
+++ b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/client/transaction/TestConflictResolutionStrategyUtil.java
@@ -211,14 +211,14 @@ public class TestConflictResolutionStrategyUtil {
   }
 
   public static void createPendingCompaction(String instantTime, HoodieTableMetaClient metaClient) throws Exception {
-    String fileId1 = "file-2";
+    String fileId1 = "file-1";
     HoodieCompactionPlan compactionPlan = new HoodieCompactionPlan();
     compactionPlan.setVersion(TimelineLayoutVersion.CURR_VERSION);
     HoodieCompactionOperation operation = new HoodieCompactionOperation();
     operation.setFileId(fileId1);
     operation.setPartitionPath(HoodieTestDataGenerator.DEFAULT_FIRST_PARTITION_PATH);
-    operation.setDataFilePath("/file-2");
-    operation.setDeltaFilePaths(Arrays.asList("/file-2"));
+    operation.setDataFilePath("/file-1");
+    operation.setDeltaFilePaths(Arrays.asList("/file-1-log1"));
     compactionPlan.setOperations(Arrays.asList(operation));
     HoodieTestTable.of(metaClient)
         .addRequestedCompaction(instantTime, compactionPlan);
diff --git a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/client/transaction/TestSimpleConcurrentFileWritesConflictResolutionStrategyWithMORTable.java b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/client/transaction/TestSimpleConcurrentFileWritesConflictResolutionStrategyWithMORTable.java
new file mode 100644
index 00000000000..8f5311b14c0
--- /dev/null
+++ b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/client/transaction/TestSimpleConcurrentFileWritesConflictResolutionStrategyWithMORTable.java
@@ -0,0 +1,89 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.hudi.client.transaction;
+
+import org.apache.hudi.common.model.HoodieCommitMetadata;
+import org.apache.hudi.common.model.HoodieTableType;
+import org.apache.hudi.common.table.timeline.HoodieActiveTimeline;
+import org.apache.hudi.common.table.timeline.HoodieInstant;
+import org.apache.hudi.common.table.timeline.HoodieTimeline;
+import org.apache.hudi.common.testutils.HoodieCommonTestHarness;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.exception.HoodieWriteConflictException;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import static org.apache.hudi.client.transaction.TestConflictResolutionStrategyUtil.createCommit;
+import static org.apache.hudi.client.transaction.TestConflictResolutionStrategyUtil.createCommitMetadata;
+import static org.apache.hudi.client.transaction.TestConflictResolutionStrategyUtil.createInflightCommit;
+import static org.apache.hudi.client.transaction.TestConflictResolutionStrategyUtil.createPendingCompaction;
+
+public class TestSimpleConcurrentFileWritesConflictResolutionStrategyWithMORTable extends HoodieCommonTestHarness {
+  @Override
+  protected HoodieTableType getTableType() {
+    return HoodieTableType.MERGE_ON_READ;
+  }
+
+  @BeforeEach
+  public void init() throws IOException {
+    initMetaClient();
+  }
+
+  @Test
+  public void testConcurrentWritesWithInterleavingInflightCompaction() throws Exception {
+    createCommit(metaClient.createNewInstantTime(), metaClient);
+    HoodieActiveTimeline timeline = metaClient.getActiveTimeline();
+    // Consider commits before this are all successful.
+    Option<HoodieInstant> lastSuccessfulInstant = timeline.getCommitsTimeline().filterCompletedInstants().lastInstant();
+
+    // Writer 1 starts.
+    String currentWriterInstant = metaClient.createNewInstantTime();
+    createInflightCommit(currentWriterInstant, metaClient);
+
+    // Compaction 1 gets scheduled and becomes inflight.
+    String newInstantTime = metaClient.createNewInstantTime();
+    createPendingCompaction(newInstantTime, metaClient);
+
+    // Writer 1 tries to commit.
+    Option<HoodieInstant> currentInstant = Option.of(
+        new HoodieInstant(HoodieInstant.State.INFLIGHT, HoodieTimeline.DELTA_COMMIT_ACTION, currentWriterInstant));
+    HoodieCommitMetadata currentMetadata = createCommitMetadata(currentWriterInstant);
+    metaClient.reloadActiveTimeline();
+
+    // Do conflict resolution.
+    SimpleConcurrentFileWritesConflictResolutionStrategy strategy =
+        new SimpleConcurrentFileWritesConflictResolutionStrategy();
+    List<HoodieInstant> candidateInstants = strategy.getCandidateInstants(
+        metaClient, currentInstant.get(), lastSuccessfulInstant).collect(Collectors.toList());
+    Assertions.assertEquals(1, candidateInstants.size());
+    ConcurrentOperation thatCommitOperation = new ConcurrentOperation(candidateInstants.get(0), metaClient);
+    ConcurrentOperation thisCommitOperation = new ConcurrentOperation(currentInstant.get(), currentMetadata);
+    Assertions.assertTrue(strategy.hasConflict(thisCommitOperation, thatCommitOperation));
+    Assertions.assertThrows(
+        HoodieWriteConflictException.class,
+        () -> strategy.resolveConflict(null, thisCommitOperation, thatCommitOperation));
+  }
+}