HIVE-29520: Add configuration to enable/disable ACID functionality in Hive Metastore

ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Initiator.java

@@ -213,8 +213,10 @@ public void init(AtomicBoolean stop) throws Exception {
     checkInterval = conf.getTimeVar(HiveConf.ConfVars.HIVE_COMPACTOR_CHECK_INTERVAL, TimeUnit.MILLISECONDS);
     metricsEnabled = MetastoreConf.getBoolVar(conf, MetastoreConf.ConfVars.METRICS_ENABLED) &&
         MetastoreConf.getBoolVar(conf, MetastoreConf.ConfVars.METASTORE_ACIDMETRICS_EXT_ON);
+    boolean isSupportAcid = MetastoreConf.getBoolVar(conf, MetastoreConf.ConfVars.METASTORE_SUPPORT_ACID);
     optimizers = Arrays.stream(MetastoreConf.getTrimmedStringsVar(conf,
             MetastoreConf.ConfVars.COMPACTOR_INITIATOR_TABLE_OPTIMIZERS))
+        .filter(e -> isSupportAcid || !e.equalsIgnoreCase(MetastoreConf.ACID_TABLE_OPTIMIZER_CLASS))
         .map(this::instantiateTableOptimizer).toList();
   }
 

standalone-metastore/metastore-common/src/main/java/org/apache/hadoop/hive/metastore/conf/MetastoreConf.java

@@ -112,6 +112,10 @@ public class MetastoreConf {
           "metastore.authentication.ldap.userMembershipKey";
   public static final String METASTORE_RETRYING_HANDLER_CLASS =
           "org.apache.hadoop.hive.metastore.RetryingHMSHandler";
+  public static final String  ACID_TABLE_OPTIMIZER_CLASS =
+      "org.apache.hadoop.hive.ql.txn.compactor.AcidTableOptimizer";
+  public static final String  ICEBERG_TABLE_OPTIMIZER_CLASS =
+      "org.apache.iceberg.mr.hive.compaction.IcebergTableOptimizer";
 
   private static final Map<String, ConfVars> metaConfs = new HashMap<>();
   private static volatile URL hiveSiteURL = null;
@@ -662,8 +666,7 @@ public enum ConfVars {
       "Enable table caching in the initiator. Currently the cache is cleaned after each cycle."),
     COMPACTOR_INITIATOR_TABLE_OPTIMIZERS("compactor.table.optimizers",
         "hive.compactor.table.optimizers",
-        "org.apache.hadoop.hive.ql.txn.compactor.AcidTableOptimizer," +
-            "org.apache.iceberg.mr.hive.compaction.IcebergTableOptimizer",
+        ACID_TABLE_OPTIMIZER_CLASS + "," + ICEBERG_TABLE_OPTIMIZER_CLASS,
         "Comma separated list of table optimizers executed by compaction Initiator."),
     COMPACTOR_WORKER_THREADS("metastore.compactor.worker.threads",
         "hive.compactor.worker.threads", 0,
@@ -1988,6 +1991,8 @@ public enum ConfVars {
         "The maximum non-native tables allowed per table type during collecting the summary."),
     METADATA_SUMMARY_NONNATIVE_THREADS("hive.metatool.summary.nonnative.threads", "hive.metatool.summary.nonnative.threads", 20,
         "Number of threads to be allocated for MetaToolTaskMetadataSummary for collecting the non-native table's summary."),
+    METASTORE_SUPPORT_ACID("metastore.support.acid", "hive.metastore.support.acid", true,
+        "Whether to support acid functionality in Hive metastore server."),
 
     // These are all values that we put here just for testing
     STR_TEST_ENTRY("test.str", "hive.test.str", "defaultval", "comment"),

standalone-metastore/metastore-server/pom.xml

@@ -385,6 +385,11 @@
       <artifactId>junit</artifactId>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>org.junit.jupiter</groupId>
+      <artifactId>junit-jupiter</artifactId>
+      <scope>test</scope>
+    </dependency>
     <dependency>
       <groupId>org.apache.directory.server</groupId>
       <artifactId>apacheds-server-integ</artifactId>

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

@@ -42,7 +42,8 @@ public class MaterializationsRebuildLockCleanerTask implements MetastoreTaskThre
 
   @Override
   public long runFrequency(TimeUnit unit) {
-    return MetastoreConf.getTimeVar(conf, MetastoreConf.ConfVars.TXN_TIMEOUT, unit) / 2;
+    return MetastoreConf.getBoolVar(conf, MetastoreConf.ConfVars.METASTORE_SUPPORT_ACID) ?
+        MetastoreConf.getTimeVar(conf, MetastoreConf.ConfVars.TXN_TIMEOUT, unit) / 2 : 0;
   }
 
   @Override

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

@@ -614,6 +614,26 @@ private static final Path getDefaultPath(IHMSHandler hmsHandler, Database db, St
 
   }
 
+  private void transformToExternalIfAcidNotSupported(Table table) throws MetaException {
+    Map<String, String> params = table.getParameters();
+    boolean isSupportAcid = MetastoreConf.getBoolVar(hmsHandler.getConf(),
+        ConfVars.METASTORE_SUPPORT_ACID);
+    if (!isSupportAcid) {
+      if (Boolean.parseBoolean(params.get(TABLE_IS_TRANSACTIONAL))) {
+        throw new MetaException("ACID tables are not permitted when the "
+            + ConfVars.METASTORE_SUPPORT_ACID.getHiveName() + " property is set to false");
+      }
+      if (TableType.MANAGED_TABLE.name().equals(table.getTableType())) {
+        table.setTableType(TableType.EXTERNAL_TABLE.toString());
+      }
+      if (TableType.EXTERNAL_TABLE.name().equals(table.getTableType())) {
+        params.put(HiveMetaHook.EXTERNAL, "TRUE");
+      }
+      params.remove(TABLE_IS_TRANSACTIONAL);
+      params.remove(TABLE_TRANSACTIONAL_PROPERTIES);
+    }
+  }
+
   @Override
   public Table transformCreateTable(Table table, List<String> processorCapabilities, String processorId) throws MetaException {
     if (!defaultCatalog.equalsIgnoreCase(table.getCatName())) {
@@ -628,6 +648,7 @@ public Table transformCreateTable(Table table, List<String> processorCapabilitie
     if (params == null) {
       params = new HashMap<>();
     }
+    transformToExternalIfAcidNotSupported(newTable);
     String tableType = newTable.getTableType();
     String dbName = table.getDbName();
     Database db = null;
@@ -673,8 +694,8 @@ public Table transformCreateTable(Table table, List<String> processorCapabilitie
           throw new MetaException("Processor has no capabilities, cannot create an ACID table.");
         }
 
-        newTable = validateTablePaths(table);
-        if (MetaStoreUtils.isInsertOnlyTableParam(table.getParameters())) { // MICRO_MANAGED Tables
+        validateTablePaths(newTable);
+        if (MetaStoreUtils.isInsertOnlyTableParam(newTable.getParameters())) { // MICRO_MANAGED Tables
           if (processorCapabilities.contains(HIVEMANAGEDINSERTWRITE)) {
             LOG.debug("Processor has required capabilities to be able to create INSERT-only tables");
             return newTable;
@@ -694,7 +715,7 @@ public Table transformCreateTable(Table table, List<String> processorCapabilitie
       }
     } else if (TableType.EXTERNAL_TABLE.name().equals(tableType)) {
       LOG.debug("Table to be created is of type " + tableType);
-      newTable = validateTablePaths(table);
+      validateTablePaths(newTable);
     }
     LOG.info("Transformer returning table:" + newTable.toString());
     return newTable;
@@ -734,7 +755,7 @@ public Table transformAlterTable(Table oldTable, Table newTable, List<String> pr
     LOG.info("Starting translation for Alter table for processor " + processorId + " with " + processorCapabilities
         + " on table " + newTable.getTableName());
 
-
+    transformToExternalIfAcidNotSupported(newTable);
     if (tableLocationChanged(oldTable, newTable)) {
       validateTablePaths(newTable);
     }

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

@@ -65,8 +65,13 @@ public List<MetaStoreThread> getCompactorThreads() throws Exception {
       compactors.add(initiator);
     }
     if (MetastoreConf.getBoolVar(configuration, MetastoreConf.ConfVars.COMPACTOR_CLEANER_ON)) {
-      MetaStoreThread cleaner = instantiateThread("org.apache.hadoop.hive.ql.txn.compactor.Cleaner");
-      compactors.add(cleaner);
+      if (MetastoreConf.getBoolVar(configuration, MetastoreConf.ConfVars.METASTORE_SUPPORT_ACID)) {
+        MetaStoreThread cleaner = instantiateThread("org.apache.hadoop.hive.ql.txn.compactor.Cleaner");
+        compactors.add(cleaner);
+      } else {
+        HiveMetaStore.LOG.warn("Compactor Cleaner is turned On. But, automatic compaction cleaner will not run " +
+            "when the {} property is set to false.", MetastoreConf.ConfVars.METASTORE_SUPPORT_ACID.getHiveName());
+      }
     }
     return compactors;
   }

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

@@ -20,7 +20,6 @@
 
 import com.cronutils.utils.VisibleForTesting;
 import com.google.common.util.concurrent.ThreadFactoryBuilder;
-
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hive.metastore.HiveMetaStore;
 import org.apache.hadoop.hive.metastore.MetastoreTaskThread;
@@ -107,20 +106,19 @@ public void takeLeadership(LeaderElection election) throws Exception {
     } else {
       tasks = new ArrayList<>(getRemoteOnlyTasks());
     }
-    int poolSize = Math.min(MetastoreConf.getIntVar(configuration,
-        MetastoreConf.ConfVars.THREAD_POOL_SIZE), tasks.size());
-    metastoreTaskThreadPool = Executors.newScheduledThreadPool(poolSize, threadFactory);
-    for (MetastoreTaskThread task : tasks) {
+    tasks.forEach(task -> {
       task.setConf(configuration);
       task.enforceMutex(election.enforceMutex());
-      long freq = task.runFrequency(TimeUnit.MILLISECONDS);
-      if (freq > 0) {
+      if (task.runFrequency(TimeUnit.MILLISECONDS) > 0) {
         runningTasks.add(task);
-        metastoreTaskThreadPool.scheduleAtFixedRate(task, freq, freq, TimeUnit.MILLISECONDS);
       }
-    }
-
+    });
+    int poolSize = Math.min(MetastoreConf.getIntVar(configuration,
+        MetastoreConf.ConfVars.THREAD_POOL_SIZE), runningTasks.size());
+    metastoreTaskThreadPool = Executors.newScheduledThreadPool(poolSize, threadFactory);
     runningTasks.forEach(task -> {
+      long freq = task.runFrequency(TimeUnit.MILLISECONDS);
+      metastoreTaskThreadPool.scheduleAtFixedRate(task, freq, freq, TimeUnit.MILLISECONDS);
       HiveMetaStore.LOG.info("Scheduling for " + task.getClass().getCanonicalName() + " service.");
     });
   }

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

@@ -50,8 +50,8 @@ public class AcidMetricLogger implements MetastoreTaskThread {
 
   @Override
   public long runFrequency(TimeUnit timeUnit) {
-    return MetastoreConf
-        .getTimeVar(conf, MetastoreConf.ConfVars.COMPACTOR_ACID_METRICS_LOGGER_FREQUENCY, timeUnit);
+    return MetastoreConf.getBoolVar(conf, MetastoreConf.ConfVars.METASTORE_SUPPORT_ACID) ?
+        MetastoreConf.getTimeVar(conf, MetastoreConf.ConfVars.COMPACTOR_ACID_METRICS_LOGGER_FREQUENCY, timeUnit) : 0;
   }
 
   @Override

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

@@ -103,7 +103,8 @@ public class AcidMetricService implements MetastoreTaskThread {
 
   @Override
   public long runFrequency(TimeUnit unit) {
-    return MetastoreConf.getTimeVar(conf, MetastoreConf.ConfVars.METASTORE_ACIDMETRICS_CHECK_INTERVAL, unit);
+    return MetastoreConf.getBoolVar(conf, MetastoreConf.ConfVars.METASTORE_SUPPORT_ACID) ?
+        MetastoreConf.getTimeVar(conf, MetastoreConf.ConfVars.METASTORE_ACIDMETRICS_CHECK_INTERVAL, unit) : 0;
   }
 
   @Override

standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/service/AcidHouseKeeperService.java

@@ -75,7 +75,8 @@ public Configuration getConf() {
 
   @Override
   public long runFrequency(TimeUnit unit) {
-    return MetastoreConf.getTimeVar(conf, MetastoreConf.ConfVars.ACID_HOUSEKEEPER_SERVICE_INTERVAL, unit);
+    return MetastoreConf.getBoolVar(conf, MetastoreConf.ConfVars.METASTORE_SUPPORT_ACID) ?
+        MetastoreConf.getTimeVar(conf, MetastoreConf.ConfVars.ACID_HOUSEKEEPER_SERVICE_INTERVAL, unit) : 0;
   }
 
   @Override

standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/service/AcidOpenTxnsCounterService.java

@@ -42,7 +42,8 @@ public class AcidOpenTxnsCounterService implements MetastoreTaskThread {
 
   @Override
   public long runFrequency(TimeUnit unit) {
-    return MetastoreConf.getTimeVar(conf, MetastoreConf.ConfVars.COUNT_OPEN_TXNS_INTERVAL, unit);
+    return MetastoreConf.getBoolVar(conf, MetastoreConf.ConfVars.METASTORE_SUPPORT_ACID) ?
+        MetastoreConf.getTimeVar(conf, MetastoreConf.ConfVars.COUNT_OPEN_TXNS_INTERVAL, unit) : 0;
   }
 
   @Override

standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/service/AcidTxnCleanerService.java

@@ -53,7 +53,8 @@ public Configuration getConf() {
 
   @Override
   public long runFrequency(TimeUnit unit) {
-    return MetastoreConf.getTimeVar(conf, MetastoreConf.ConfVars.ACID_TXN_CLEANER_INTERVAL, unit);
+    return MetastoreConf.getBoolVar(conf, MetastoreConf.ConfVars.METASTORE_SUPPORT_ACID) ?
+        MetastoreConf.getTimeVar(conf, MetastoreConf.ConfVars.ACID_TXN_CLEANER_INTERVAL, unit) : 0;
   }
 
   @Override

standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/service/CompactionHouseKeeperService.java

@@ -36,11 +36,14 @@ public CompactionHouseKeeperService() {
 
   @Override
   protected void initTasks(){
-    tasks = ImmutableMap.<FailableRunnable<MetaException>, String>builder()
-            .put(txnHandler::removeDuplicateCompletedTxnComponents,
-                    "Cleaning duplicate COMPLETED_TXN_COMPONENTS entries")
-            .put(txnHandler::purgeCompactionHistory, "Cleaning obsolete compaction history entries")
-            .build();
+    ImmutableMap.Builder<FailableRunnable<MetaException>, String> taskBuilder =
+        ImmutableMap.<FailableRunnable<MetaException>, String>builder()
+            .put(txnHandler::purgeCompactionHistory, "Cleaning obsolete compaction history entries");
+    if (MetastoreConf.getBoolVar(getConf(), MetastoreConf.ConfVars.METASTORE_SUPPORT_ACID)) {
+      taskBuilder.put(txnHandler::removeDuplicateCompletedTxnComponents,
+          "Cleaning duplicate COMPLETED_TXN_COMPONENTS entries");
+    }
+    tasks = taskBuilder.build();
   }
 
   @Override

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

@@ -0,0 +1,125 @@
+/*
+ * 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.hadoop.hive.metastore;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hive.metastore.annotation.MetastoreUnitTest;
+import org.apache.hadoop.hive.metastore.api.MetaException;
+import org.apache.hadoop.hive.metastore.api.Table;
+import org.apache.hadoop.hive.metastore.client.builder.DatabaseBuilder;
+import org.apache.hadoop.hive.metastore.client.builder.TableBuilder;
+import org.apache.hadoop.hive.metastore.conf.MetastoreConf;
+import org.apache.hadoop.hive.metastore.conf.MetastoreConf.ConfVars;
+import org.apache.hadoop.util.StringUtils;
+import org.apache.thrift.TException;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.jupiter.api.Assertions;
+
+import static org.apache.hadoop.hive.metastore.api.hive_metastoreConstants.TABLE_IS_TRANSACTIONAL;
+
+/**
+ * Tests to verify metastore without acid support.
+ */
+@Category(MetastoreUnitTest.class)
+public class TestNoAcidSupport {
+  private static Configuration conf;
+  private static HiveMetaStoreClient client;
+  private static final String DB_NAME = "TestNoAcidSupport";
+  private static final String TABLE_NAME = "t";
+
+  @BeforeClass
+  public static void beforeTests() throws Exception {
+    conf = MetastoreConf.newMetastoreConf();
+    MetastoreConf.setBoolVar(conf, ConfVars.METASTORE_SUPPORT_ACID, false);
+    client = new HiveMetaStoreClient(conf);
+    client.dropDatabase(DB_NAME, true, true, true);
+    new DatabaseBuilder().setName(DB_NAME).create(client, conf);
+  }
+
+  @AfterClass
+  public static void afterTests() throws Exception {
+    try {
+      client.dropDatabase(DB_NAME, true, true, true);
+      client.close();
+    } catch (Throwable e) {
+      System.err.println(StringUtils.stringifyException(e));
+      throw e;
+    }
+  }
+
+  @After
+  public void afterTest() throws TException {
+    client.dropTable(DB_NAME, TABLE_NAME);
+  }
+
+  @Test
+  public void testCreateAcidTable() {
+    Exception exception = Assertions.assertThrows(MetaException.class, () -> {
+      new TableBuilder().setDbName(DB_NAME).setTableName(TABLE_NAME)
+          .addCol("i", ColumnType.INT_TYPE_NAME)
+          .setType(TableType.MANAGED_TABLE.name())
+          .addTableParam(TABLE_IS_TRANSACTIONAL, "true")
+          .create(client, conf);
+    });
+    Assertions.assertTrue(exception.getMessage().contains("ACID tables are not permitted when the " +
+        ConfVars.METASTORE_SUPPORT_ACID.getHiveName() + " property is set to false"));
+  }
+
+  @Test
+  public void testCreateManagedTableChangeToExternal() throws Exception {
+    new TableBuilder().setDbName(DB_NAME).setTableName(TABLE_NAME)
+        .addCol("i", ColumnType.INT_TYPE_NAME)
+        .setType(TableType.MANAGED_TABLE.name())
+        .create(client, conf);
+    Table t = client.getTable(DB_NAME, TABLE_NAME);
+    Assertions.assertEquals(TableType.EXTERNAL_TABLE.name(), t.getTableType());
+    Assertions.assertTrue(Boolean.parseBoolean(t.getParameters().get(HiveMetaHook.EXTERNAL)));
+  }
+
+  @Test
+  public void testCreateExternalTable() throws Exception {
+    new TableBuilder().setDbName(DB_NAME).setTableName(TABLE_NAME)
+        .addCol("i", ColumnType.INT_TYPE_NAME)
+        .setType(TableType.EXTERNAL_TABLE.name())
+        .create(client, conf);
+    Table t = client.getTable(DB_NAME, TABLE_NAME);
+    Assertions.assertEquals(TableType.EXTERNAL_TABLE.name(), t.getTableType());
+    Assertions.assertTrue(Boolean.parseBoolean(t.getParameters().get(HiveMetaHook.EXTERNAL)));
+  }
+
+  @Test
+  public void testAlterToManagedTable() throws Exception {
+    new TableBuilder().setDbName(DB_NAME).setTableName(TABLE_NAME)
+        .addCol("i", ColumnType.INT_TYPE_NAME)
+        .setType(TableType.EXTERNAL_TABLE.name())
+        .create(client, conf);
+    Table t = client.getTable(DB_NAME, TABLE_NAME);
+    Assertions.assertEquals(TableType.EXTERNAL_TABLE.name(), t.getTableType());
+    t.setTableType(TableType.MANAGED_TABLE.name());
+    t.getParameters().put(TABLE_IS_TRANSACTIONAL, "true");
+    Exception exception = Assertions.assertThrows(MetaException.class, () -> {
+      client.alter_table(DB_NAME, TABLE_NAME, t);
+    });
+    Assertions.assertTrue(exception.getMessage().contains("ACID tables are not permitted when the " +
+        ConfVars.METASTORE_SUPPORT_ACID.getHiveName() + " property is set to false"));
+  }
+}