HIVE-28121: Use direct SQL for transactional altering table parameter (#5129)

standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/HiveAlterHandler.java

@@ -56,7 +56,6 @@
 import org.apache.hadoop.hive.metastore.api.Table;
 import org.apache.hadoop.hive.metastore.api.hive_metastoreConstants;
 
-import javax.jdo.Constants;
 import java.io.IOException;
 import java.net.URI;
 import java.util.ArrayList;
@@ -183,12 +182,7 @@ public void alterTable(RawStore msdb, Warehouse wh, String catName, String dbnam
       String expectedValue = environmentContext != null && environmentContext.getProperties() != null ?
               environmentContext.getProperties().get(hive_metastoreConstants.EXPECTED_PARAMETER_VALUE) : null;
 
-      if (expectedKey != null) {
-        // If we have to check the expected state of the table we have to prevent nonrepeatable reads.
-        msdb.openTransaction(Constants.TX_REPEATABLE_READ);
-      } else {
-        msdb.openTransaction();
-      }
+      msdb.openTransaction();
       // get old table
       // Note: we don't verify stats here; it's done below in alterTableUpdateTableColumnStats.
       olddb = msdb.getDatabase(catName, dbname);
@@ -198,10 +192,20 @@ public void alterTable(RawStore msdb, Warehouse wh, String catName, String dbnam
             TableName.getQualified(catName, dbname, name) + " doesn't exist");
       }
 
-      if (expectedKey != null && expectedValue != null
-              && !expectedValue.equals(oldt.getParameters().get(expectedKey))) {
-        throw new MetaException("The table has been modified. The parameter value for key '" + expectedKey + "' is '"
-                + oldt.getParameters().get(expectedKey) + "'. The expected was value was '" + expectedValue + "'");
+      if (expectedKey != null && expectedValue != null) {
+        String newValue = newt.getParameters().get(expectedKey);
+        if (newValue == null) {
+          throw new MetaException(String.format("New value for expected key %s is not set", expectedKey));
+        }
+        if (!expectedValue.equals(oldt.getParameters().get(expectedKey))) {
+          throw new MetaException("The table has been modified. The parameter value for key '" + expectedKey + "' is '"
+              + oldt.getParameters().get(expectedKey) + "'. The expected was value was '" + expectedValue + "'");
+        }
+        long affectedRows = msdb.updateParameterWithExpectedValue(oldt, expectedKey, expectedValue, newValue);
+        if (affectedRows != 1) {
+          // make sure concurrent modification exception messages have the same prefix
+          throw new MetaException("The table has been modified. The parameter value for key '" + expectedKey + "' is different");
+        }
       }
 
       validateTableChangesOnReplSource(olddb, oldt, newt, environmentContext);

standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/MetaStoreDirectSql.java

@@ -53,6 +53,7 @@
 import javax.jdo.Transaction;
 import javax.jdo.datastore.JDOConnection;
 
+import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableSet;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.hadoop.conf.Configuration;
@@ -3273,4 +3274,13 @@ private List<Long> getFunctionIds(String catName) throws MetaException {
       return result;
     }
   }
+
+  long updateTableParam(Table table, String key, String expectedValue, String newValue) {
+    String statement = TxnUtils.createUpdatePreparedStmt(
+        "\"TABLE_PARAMS\"",
+        ImmutableList.of("\"PARAM_VALUE\""),
+        ImmutableList.of("\"TBL_ID\"", "\"PARAM_KEY\"", "\"PARAM_VALUE\""));
+    Query query = pm.newQuery("javax.jdo.query.SQL", statement);
+    return (long) query.executeWithArray(newValue, table.getId(), key, expectedValue);
+  }
 }

standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/ObjectStore.java

@@ -624,6 +624,29 @@ public boolean openTransaction(String isolationLevel) {
     return result;
   }
 
+  @Override
+  public long updateParameterWithExpectedValue(Table table, String key, String expectedValue, String newValue)
+      throws MetaException, NoSuchObjectException {
+    return new GetHelper<Long>(table.getCatName(), table.getDbName(), table.getTableName(), true, false) {
+
+      @Override
+      protected String describeResult() {
+        return "Affected rows";
+      }
+
+      @Override
+      protected Long getSqlResult(GetHelper<Long> ctx) throws MetaException {
+        return directSql.updateTableParam(table, key, expectedValue, newValue);
+      }
+
+      @Override
+      protected Long getJdoResult(GetHelper<Long> ctx) throws MetaException, NoSuchObjectException, InvalidObjectException {
+        throw new UnsupportedOperationException(
+            "Cannot update parameter with JDO, make sure direct SQL is enabled");
+      }
+    }.run(false);
+  }
+
   /**
    * if this is the commit of the first open call then an actual commit is
    * called.

standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/RawStore.java

@@ -2317,4 +2317,14 @@ default PropertyStore getPropertyStore() {
     return null;
   }
 
+  /**
+   * Updates a given table parameter with expected value.
+   *
+   * @return the number of rows updated
+   */
+  default long updateParameterWithExpectedValue(Table table, String key, String expectedValue, String newValue)
+      throws MetaException, NoSuchObjectException {
+    throw new UnsupportedOperationException("This Store doesn't support updating table parameter with expected value");
+  }
+
 }

standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/client/TestTablesCreateDropAlterTruncate.java

@@ -63,6 +63,7 @@
 import org.apache.thrift.protocol.TProtocolException;
 import org.junit.After;
 import org.junit.Assert;
+import org.junit.Assume;
 import org.junit.Before;
 import org.junit.BeforeClass;
 import org.junit.Test;
@@ -1196,6 +1197,8 @@ public void testAlterTableAlreadyExists() throws Exception {
 
   @Test
   public void testAlterTableExpectedPropertyMatch() throws Exception {
+    Assume.assumeTrue(MetastoreConf.getBoolVar(metaStore.getConf(), ConfVars.TRY_DIRECT_SQL));
+    Assume.assumeTrue(MetastoreConf.getBoolVar(metaStore.getConf(), ConfVars.TRY_DIRECT_SQL_DDL));
     Table originalTable = testTables[0];
 
     EnvironmentContext context = new EnvironmentContext();
@@ -1209,6 +1212,8 @@ public void testAlterTableExpectedPropertyMatch() throws Exception {
 
   @Test(expected = MetaException.class)
   public void testAlterTableExpectedPropertyDifferent() throws Exception {
+    Assume.assumeTrue(MetastoreConf.getBoolVar(metaStore.getConf(), ConfVars.TRY_DIRECT_SQL));
+    Assume.assumeTrue(MetastoreConf.getBoolVar(metaStore.getConf(), ConfVars.TRY_DIRECT_SQL_DDL));
     Table originalTable = testTables[0];
 
     EnvironmentContext context = new EnvironmentContext();
@@ -1228,6 +1233,8 @@ public void testAlterTableExpectedPropertyDifferent() throws Exception {
    */
   @Test
   public void testAlterTableExpectedPropertyConcurrent() throws Exception {
+    Assume.assumeTrue(MetastoreConf.getBoolVar(metaStore.getConf(), ConfVars.TRY_DIRECT_SQL));
+    Assume.assumeTrue(MetastoreConf.getBoolVar(metaStore.getConf(), ConfVars.TRY_DIRECT_SQL_DDL));
     Table originalTable = testTables[0];
 
     originalTable.getParameters().put("snapshot", "0");