Core, Hive: Double check commit status in case of commit conflict for NoLock

core/src/main/java/org/apache/iceberg/BaseMetastoreOperations.java

@@ -28,6 +28,7 @@
 import static org.apache.iceberg.TableProperties.COMMIT_STATUS_CHECKS_TOTAL_WAIT_MS_DEFAULT;
 
 import java.util.Map;
+import java.util.Optional;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Supplier;
 import org.apache.iceberg.util.PropertyUtil;
@@ -63,6 +64,31 @@ protected CommitStatus checkCommitStatus(
       String newMetadataLocation,
       Map<String, String> properties,
       Supplier<Boolean> commitStatusSupplier) {
+    if (metadataLocationCommitted(
+            tableOrViewName, newMetadataLocation, properties, commitStatusSupplier)
+        .orElse(false)) {
+      return CommitStatus.SUCCESS;
+    }
+    return CommitStatus.UNKNOWN;
+  }
+
+  /**
+   * Attempt to load the content and see if any current or past metadata location matches the one we
+   * were attempting to set.
+   *
+   * @param tableOrViewName full name of the Table/View
+   * @param newMetadataLocation the path of the new commit file
+   * @param properties properties for retry
+   * @param commitStatusSupplier check if the latest metadata presents or not using metadata
+   *     location for table.
+   * @return Empty if locations cannot be checked, e.g. unable to refresh. True if the new location
+   *     is committed, false otherwise.
+   */
+  protected Optional<Boolean> metadataLocationCommitted(
+      String tableOrViewName,
+      String newMetadataLocation,
+      Map<String, String> properties,
+      Supplier<Boolean> commitStatusSupplier) {
     int maxAttempts =
         PropertyUtil.propertyAsInt(
             properties, COMMIT_NUM_STATUS_CHECKS, COMMIT_NUM_STATUS_CHECKS_DEFAULT);
@@ -78,7 +104,7 @@ protected CommitStatus checkCommitStatus(
             COMMIT_STATUS_CHECKS_TOTAL_WAIT_MS,
             COMMIT_STATUS_CHECKS_TOTAL_WAIT_MS_DEFAULT);
 
-    AtomicReference<CommitStatus> status = new AtomicReference<>(CommitStatus.UNKNOWN);
+    AtomicReference<Boolean> res = new AtomicReference<>(null);
 
     Tasks.foreach(newMetadataLocation)
         .retry(maxAttempts)
@@ -96,23 +122,24 @@ protected CommitStatus checkCommitStatus(
                     "Commit status check: Commit to {} of {} succeeded",
                     tableOrViewName,
                     newMetadataLocation);
-                status.set(CommitStatus.SUCCESS);
+                res.set(true);
               } else {
                 LOG.warn(
                     "Commit status check: Commit to {} of {} unknown, new metadata location is not current "
                         + "or in history",
                     tableOrViewName,
                     newMetadataLocation);
+                res.set(false);
               }
             });
 
-    if (status.get() == CommitStatus.UNKNOWN) {
+    if (res.get() == null) {
       LOG.error(
           "Cannot determine commit state to {}. Failed during checking {} times. "
               + "Treating commit state as unknown.",
           tableOrViewName,
           maxAttempts);
     }
-    return status.get();
+    return Optional.ofNullable(res.get());
   }
 }

core/src/main/java/org/apache/iceberg/BaseMetastoreTableOperations.java

@@ -310,7 +310,7 @@ protected CommitStatus checkCommitStatus(String newMetadataLocation, TableMetada
    * @return true if the new metadata location is the current metadata location or present within
    *     previous metadata files.
    */
-  private boolean checkCurrentMetadataLocation(String newMetadataLocation) {
+  protected boolean checkCurrentMetadataLocation(String newMetadataLocation) {
     TableMetadata metadata = refresh();
     String currentMetadataFileLocation = metadata.metadataFileLocation();
     return currentMetadataFileLocation.equals(newMetadataLocation)

hive-metastore/src/main/java/org/apache/iceberg/hive/HiveTableOperations.java

@@ -268,16 +268,6 @@ protected void doCommit(TableMetadata base, TableMetadata metadata) {
         throw e;
 
       } catch (Throwable e) {
-        if (e.getMessage() != null
-            && e.getMessage()
-                .contains(
-                    "The table has been modified. The parameter value for key '"
-                        + HiveTableOperations.METADATA_LOCATION_PROP
-                        + "' is")) {
-          throw new CommitFailedException(
-              e, "The table %s.%s has been modified concurrently", database, tableName);
-        }
-
         if (e.getMessage() != null
             && e.getMessage().contains("Table/View 'HIVE_LOCKS' does not exist")) {
           throw new RuntimeException(
@@ -287,15 +277,32 @@ protected void doCommit(TableMetadata base, TableMetadata metadata) {
               e);
         }
 
-        LOG.error(
-            "Cannot tell if commit to {}.{} succeeded, attempting to reconnect and check.",
-            database,
-            tableName,
-            e);
         commitStatus = BaseMetastoreOperations.CommitStatus.UNKNOWN;
-        commitStatus =
-            BaseMetastoreOperations.CommitStatus.valueOf(
-                checkCommitStatus(newMetadataLocation, metadata).name());
+        if (e.getMessage() != null
+            && e.getMessage()
+                .contains(
+                    "The table has been modified. The parameter value for key '"
+                        + HiveTableOperations.METADATA_LOCATION_PROP
+                        + "' is")) {
+          // It's possible the HMS client incorrectly retries a successful operation, due to network
+          // issue for example, and triggers this exception. So we need double-check to
+          // make sure this is really a concurrent modification
+          commitStatus = handlePossibleConcurrentModification(newMetadataLocation, metadata);
+          if (commitStatus == BaseMetastoreOperations.CommitStatus.FAILURE) {
+            throw new CommitFailedException(
+                e, "The table %s.%s has been modified concurrently", database, tableName);
+          }
+        } else {
+          LOG.error(
+              "Cannot tell if commit to {}.{} succeeded, attempting to reconnect and check.",
+              database,
+              tableName,
+              e);
+          commitStatus =
+              BaseMetastoreOperations.CommitStatus.valueOf(
+                  checkCommitStatus(newMetadataLocation, metadata).name());
+        }
+
         switch (commitStatus) {
           case SUCCESS:
             break;
@@ -324,6 +331,24 @@ protected void doCommit(TableMetadata base, TableMetadata metadata) {
         "Committed to table {} with the new metadata location {}", fullName, newMetadataLocation);
   }
 
+  private BaseMetastoreOperations.CommitStatus handlePossibleConcurrentModification(
+      String newMetadataLocation, TableMetadata metadata) {
+    Optional<Boolean> locationCommitted =
+        metadataLocationCommitted(
+            tableName(),
+            newMetadataLocation,
+            metadata.properties(),
+            () -> checkCurrentMetadataLocation(newMetadataLocation));
+    if (locationCommitted.isPresent()) {
+      if (locationCommitted.get()) {
+        return BaseMetastoreOperations.CommitStatus.SUCCESS;
+      } else {
+        return BaseMetastoreOperations.CommitStatus.FAILURE;
+      }
+    }
+    return BaseMetastoreOperations.CommitStatus.UNKNOWN;
+  }
+
   private void setHmsTableParameters(
       String newMetadataLocation,
       Table tbl,

hive-metastore/src/test/java/org/apache/iceberg/hive/HiveMetastoreExtension.java

@@ -40,16 +40,10 @@ private HiveMetastoreExtension(String databaseName, Map<String, String> hiveConf
 
   @Override
   public void beforeAll(ExtensionContext extensionContext) throws Exception {
-    metastore = new TestHiveMetastore();
-    HiveConf hiveConfWithOverrides = new HiveConf(TestHiveMetastore.class);
-    if (hiveConfOverride != null) {
-      for (Map.Entry<String, String> kv : hiveConfOverride.entrySet()) {
-        hiveConfWithOverrides.set(kv.getKey(), kv.getValue());
-      }
-    }
+    metastore = new TestHiveMetastore(hiveConfOverride);
 
-    metastore.start(hiveConfWithOverrides);
-    metastoreClient = new HiveMetaStoreClient(hiveConfWithOverrides);
+    metastore.start(new HiveConf(TestHiveMetastore.class));
+    metastoreClient = new HiveMetaStoreClient(metastore.hiveConf());
     if (null != databaseName) {
       String dbPath = metastore.getDatabasePath(databaseName);
       Database db = new Database(databaseName, "description", dbPath, Maps.newHashMap());

hive-metastore/src/test/java/org/apache/iceberg/hive/TestHiveCommitLocks.java

@@ -22,6 +22,7 @@
 import static org.apache.iceberg.types.Types.NestedField.required;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.Mockito.any;
 import static org.mockito.Mockito.atLeastOnce;
 import static org.mockito.Mockito.doAnswer;
@@ -63,6 +64,7 @@
 import org.apache.iceberg.Schema;
 import org.apache.iceberg.Table;
 import org.apache.iceberg.TableMetadata;
+import org.apache.iceberg.TableProperties;
 import org.apache.iceberg.catalog.TableIdentifier;
 import org.apache.iceberg.exceptions.CommitFailedException;
 import org.apache.iceberg.hadoop.ConfigProperties;
@@ -109,7 +111,12 @@ public class TestHiveCommitLocks {
   private static final HiveMetastoreExtension HIVE_METASTORE_EXTENSION =
       HiveMetastoreExtension.builder()
           .withDatabase(DB_NAME)
-          .withConfig(ImmutableMap.of(HiveConf.ConfVars.HIVE_TXN_TIMEOUT.varname, "1s"))
+          .withConfig(
+              ImmutableMap.of(
+                  HiveConf.ConfVars.HIVE_TXN_TIMEOUT.varname,
+                  "1s",
+                  HiveConf.ConfVars.METASTORE_TRY_DIRECT_SQL.varname,
+                  "true"))
           .build();
 
   private static HiveCatalog catalog;
@@ -205,6 +212,37 @@ public static void cleanup() {
     }
   }
 
+  @Test
+  public void testMultipleAlterTableForNoLock() throws Exception {
+    Table table = catalog.loadTable(TABLE_IDENTIFIER);
+    table.updateProperties().set(TableProperties.HIVE_LOCK_ENABLED, "false").commit();
+    spyOps.refresh();
+    TableMetadata metadataV3 = spyOps.current();
+    AtomicReference<Throwable> alterTableException = new AtomicReference<>(null);
+    doAnswer(
+            i -> {
+              try {
+                // mock a situation where alter table is unexpectedly invoked more than once
+                i.callRealMethod();
+                return i.callRealMethod();
+              } catch (Throwable e) {
+                alterTableException.compareAndSet(null, e);
+                throw e;
+              }
+            })
+        .when(spyClient)
+        .alter_table_with_environmentContext(anyString(), anyString(), any(), any());
+    spyOps.commit(metadataV3, metadataV1);
+    verify(spyClient, times(1))
+        .alter_table_with_environmentContext(anyString(), anyString(), any(), any());
+    assertThat(alterTableException)
+        .as("Expecting to trigger an exception indicating table has been modified")
+        .hasValueMatching(
+            t ->
+                t.getMessage()
+                    .contains("The table has been modified. The parameter value for key '"));
+  }
+
   @Test
   public void testLockAcquisitionAtFirstTime() throws TException, InterruptedException {
     doReturn(acquiredLockResponse).when(spyClient).lock(any());

hive-metastore/src/test/java/org/apache/iceberg/hive/TestHiveMetastore.java

@@ -31,6 +31,7 @@
 import java.sql.Connection;
 import java.sql.DriverManager;
 import java.sql.SQLException;
+import java.util.Map;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import org.apache.hadoop.conf.Configuration;
@@ -128,6 +129,15 @@ public class TestHiveMetastore {
   private TServer server;
   private HiveMetaStore.HMSHandler baseHandler;
   private HiveClientPool clientPool;
+  private final Map<String, String> hiveConfOverride;
+
+  public TestHiveMetastore() {
+    this(null);
+  }
+
+  public TestHiveMetastore(Map<String, String> hiveConfOverride) {
+    this.hiveConfOverride = hiveConfOverride;
+  }
 
   /**
    * Starts a TestHiveMetastore with the default connection pool size (5) and the default HiveConf.
@@ -271,6 +281,11 @@ private void initConf(HiveConf conf, int port) {
     // Setting this to avoid thrift exception during running Iceberg tests outside Iceberg.
     conf.set(
         HiveConf.ConfVars.HIVE_IN_TEST.varname, HiveConf.ConfVars.HIVE_IN_TEST.getDefaultValue());
+    if (hiveConfOverride != null) {
+      for (Map.Entry<String, String> kv : hiveConfOverride.entrySet()) {
+        conf.set(kv.getKey(), kv.getValue());
+      }
+    }
   }
 
   private static void setupMetastoreDB(String dbURL) throws SQLException, IOException {

hive-metastore/src/test/resources/hive-schema-3.1.0.derby.sql

@@ -130,7 +130,7 @@ CREATE TABLE "APP"."TAB_COL_STATS"(
     "BIT_VECTOR" BLOB
 );
 
-CREATE TABLE "APP"."TABLE_PARAMS" ("TBL_ID" BIGINT NOT NULL, "PARAM_KEY" VARCHAR(256) NOT NULL, "PARAM_VALUE" CLOB);
+CREATE TABLE "APP"."TABLE_PARAMS" ("TBL_ID" BIGINT NOT NULL, "PARAM_KEY" VARCHAR(256) NOT NULL, "PARAM_VALUE" VARCHAR(32672));
 
 CREATE TABLE "APP"."BUCKETING_COLS" ("SD_ID" BIGINT NOT NULL, "BUCKET_COL_NAME" VARCHAR(256), "INTEGER_IDX" INTEGER NOT NULL);