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

Author: Venu Reddy

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

@@ -73,7 +73,7 @@ public class MetastoreConf {
   static final String HIVE_ALTER_HANDLE_CLASS =
       "org.apache.hadoop.hive.metastore.HiveAlterHandler";
   @VisibleForTesting
-  static final String MATERIALZIATIONS_REBUILD_LOCK_CLEANER_TASK_CLASS =
+  public static final String MATERIALZIATIONS_REBUILD_LOCK_CLEANER_TASK_CLASS =
       "org.apache.hadoop.hive.metastore.MaterializationsRebuildLockCleanerTask";
   @VisibleForTesting
   static final String METASTORE_TASK_THREAD_CLASS =
@@ -86,24 +86,24 @@ public class MetastoreConf {
   @VisibleForTesting
   static final String EVENT_CLEANER_TASK_CLASS =
       "org.apache.hadoop.hive.metastore.events.EventCleanerTask";
-  static final String ACID_METRICS_TASK_CLASS =
+  public static final String ACID_METRICS_TASK_CLASS =
       "org.apache.hadoop.hive.metastore.metrics.AcidMetricService";
-  static final String ACID_METRICS_LOGGER_CLASS =
+  public static final String ACID_METRICS_LOGGER_CLASS =
       "org.apache.hadoop.hive.metastore.metrics.AcidMetricLogger";
   @VisibleForTesting
   static final String METASTORE_DELEGATION_MANAGER_CLASS =
       "org.apache.hadoop.hive.metastore.security.MetastoreDelegationTokenManager";
   @VisibleForTesting
-  static final String ACID_HOUSEKEEPER_SERVICE_CLASS =
+  public static final String ACID_HOUSEKEEPER_SERVICE_CLASS =
       "org.apache.hadoop.hive.metastore.txn.service.AcidHouseKeeperService";
   @VisibleForTesting
   static final String COMPACTION_HOUSEKEEPER_SERVICE_CLASS =
       "org.apache.hadoop.hive.metastore.txn.service.CompactionHouseKeeperService";
   @VisibleForTesting
-  static final String ACID_TXN_CLEANER_SERVICE_CLASS =
+  public static final String ACID_TXN_CLEANER_SERVICE_CLASS =
       "org.apache.hadoop.hive.metastore.txn.service.AcidTxnCleanerService";
   @VisibleForTesting
-  static final String ACID_OPEN_TXNS_COUNTER_SERVICE_CLASS =
+  public static final String ACID_OPEN_TXNS_COUNTER_SERVICE_CLASS =
       "org.apache.hadoop.hive.metastore.txn.service.AcidOpenTxnsCounterService";
   @VisibleForTesting
   static final String ICEBERG_TABLE_SNAPSHOT_EXPIRY_SERVICE_CLASS =
@@ -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,
@@ -1987,6 +1990,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/HiveMetaStore.java

@@ -700,6 +700,15 @@ public static void startMetaStore(int port, HadoopThriftAuthBridge bridge,
     boolean directSqlEnabled = MetastoreConf.getBoolVar(conf, ConfVars.TRY_DIRECT_SQL);
     LOG.info("Direct SQL optimization = {}",  directSqlEnabled);
 
+    if (!MetastoreConf.getBoolVar(conf, ConfVars.METASTORE_SUPPORT_ACID)) {
+      if (MetastoreConf.getBoolVar(conf, MetastoreConf.ConfVars.COMPACTOR_CLEANER_ON)) {
+        LOG.warn("Compactor Cleaner is turned On. But, automatic compaction cleaner will not be triggered " +
+                "when the {} property is set to false. Hence, Cleaner is turned Off",
+            ConfVars.METASTORE_SUPPORT_ACID.getHiveName());
+        MetastoreConf.setBoolVar(conf, MetastoreConf.ConfVars.COMPACTOR_CLEANER_ON,false);
+      }
+    }
+
     if (startMetaStoreThreads) {
       Lock metaStoreThreadsLock = new ReentrantLock();
       Condition startCondition = metaStoreThreadsLock.newCondition();

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

@@ -615,6 +615,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())) {
@@ -629,6 +649,7 @@ public Table transformCreateTable(Table table, List<String> processorCapabilitie
     if (params == null) {
       params = new HashMap<>();
     }
+    transformToExternalIfAcidNotSupported(newTable);
     String tableType = newTable.getTableType();
     String txnal = null;
     String txn_properties = null;
@@ -682,8 +703,7 @@ public Table transformCreateTable(Table table, List<String> processorCapabilitie
           throw new MetaException("Processor has no capabilities, cannot create an ACID table.");
         }
 
-
-        newTable = validateTablePaths(table);
+        validateTablePaths(newTable);
         if (isInsertAcid) { // MICRO_MANAGED Tables
           if (processorCapabilities.contains(HIVEMANAGEDINSERTWRITE)) {
             LOG.debug("Processor has required capabilities to be able to create INSERT-only tables");
@@ -704,7 +724,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;
@@ -744,7 +764,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/HouseKeepingTasks.java

@@ -20,7 +20,7 @@
 
 import com.cronutils.utils.VisibleForTesting;
 import com.google.common.util.concurrent.ThreadFactoryBuilder;
-
+import com.google.common.collect.ImmutableSet;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hive.metastore.HiveMetaStore;
 import org.apache.hadoop.hive.metastore.MetastoreTaskThread;
@@ -69,6 +69,12 @@ public List<MetastoreTaskThread> getRemoteOnlyTasks() throws Exception {
 
     Collection<String> taskNames =
         MetastoreConf.getStringCollection(configuration, MetastoreConf.ConfVars.TASK_THREADS_REMOTE_ONLY);
+    if (!MetastoreConf.getBoolVar(configuration, MetastoreConf.ConfVars.METASTORE_SUPPORT_ACID)) {
+      taskNames.removeAll(ImmutableSet.of(MetastoreConf.ACID_OPEN_TXNS_COUNTER_SERVICE_CLASS,
+          MetastoreConf.ACID_HOUSEKEEPER_SERVICE_CLASS,
+          MetastoreConf.ACID_TXN_CLEANER_SERVICE_CLASS,
+          MetastoreConf.MATERIALZIATIONS_REBUILD_LOCK_CLEANER_TASK_CLASS));
+    }
     for (String taskName : taskNames) {
       if (CompactionHouseKeeperService.class.getName().equals(taskName) && !isCompactorEnabled) {
         continue;
@@ -85,6 +91,10 @@ public List<MetastoreTaskThread> getAlwaysTasks() throws Exception {
     List<MetastoreTaskThread> alwaysTasks = new ArrayList<>();
     Collection<String> taskNames =
         MetastoreConf.getStringCollection(configuration, MetastoreConf.ConfVars.TASK_THREADS_ALWAYS);
+    if (!MetastoreConf.getBoolVar(configuration, MetastoreConf.ConfVars.METASTORE_SUPPORT_ACID)) {
+      taskNames.removeAll(ImmutableSet.of(MetastoreConf.ACID_METRICS_TASK_CLASS,
+          MetastoreConf.ACID_METRICS_LOGGER_CLASS));
+    }
     for (String taskName : taskNames) {
       MetastoreTaskThread task =
           JavaUtils.newInstance(JavaUtils.getClass(taskName, MetastoreTaskThread.class));

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 dbName = "TestNoAcidSupport";
+  private String tableName = "t1";
+
+  @BeforeClass
+  public static void beforeTests() throws Exception {
+    conf = MetastoreConf.newMetastoreConf();
+    MetastoreConf.setBoolVar(conf, ConfVars.METASTORE_SUPPORT_ACID, false);
+    client = new HiveMetaStoreClient(conf);
+    client.dropDatabase(dbName, true, true, true);
+    new DatabaseBuilder().setName(dbName).create(client, conf);
+  }
+
+  @AfterClass
+  public static void afterTests() throws Exception {
+    try {
+      client.dropDatabase(dbName, true, true, true);
+      client.close();
+    } catch (Throwable e) {
+      System.err.println(StringUtils.stringifyException(e));
+      throw e;
+    }
+  }
+
+  @After
+  public void afterTest() throws TException {
+    client.dropTable(dbName, tableName);
+  }
+
+  @Test
+  public void testCreateAcidTable() {
+    Exception exception = Assertions.assertThrows(MetaException.class, () -> {
+      new TableBuilder().setDbName(dbName).setTableName(tableName)
+          .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(dbName).setTableName(tableName)
+        .addCol("i", ColumnType.INT_TYPE_NAME)
+        .setType(TableType.MANAGED_TABLE.name())
+        .create(client, conf);
+    Table t1 = client.getTable(dbName, tableName);
+    Assertions.assertEquals(TableType.EXTERNAL_TABLE.name(), t1.getTableType());
+    Assertions.assertTrue(Boolean.parseBoolean(t1.getParameters().get(HiveMetaHook.EXTERNAL)));
+  }
+
+  @Test
+  public void testCreateExternalTable() throws Exception {
+    new TableBuilder().setDbName(dbName).setTableName(tableName)
+        .addCol("i", ColumnType.INT_TYPE_NAME)
+        .setType(TableType.EXTERNAL_TABLE.name())
+        .create(client, conf);
+    Table t1 = client.getTable(dbName, tableName);
+    Assertions.assertEquals( TableType.EXTERNAL_TABLE.name(), t1.getTableType());
+    Assertions.assertTrue(Boolean.parseBoolean(t1.getParameters().get(HiveMetaHook.EXTERNAL)));
+  }
+
+  @Test
+  public void testAlterToManagedTable() throws Exception {
+    new TableBuilder().setDbName(dbName).setTableName(tableName)
+        .addCol("i", ColumnType.INT_TYPE_NAME)
+        .setType(TableType.EXTERNAL_TABLE.name())
+        .create(client, conf);
+    Table t1 = client.getTable(dbName, tableName);
+    Assertions.assertEquals( TableType.EXTERNAL_TABLE.name(), t1.getTableType());
+    t1.setTableType(TableType.MANAGED_TABLE.name());
+    t1.getParameters().put(TABLE_IS_TRANSACTIONAL, "true");
+    Exception exception = Assertions.assertThrows(MetaException.class, () -> {
+      client.alter_table(dbName, tableName, t1);
+    });
+    Assertions.assertTrue(exception.getMessage()
+        .contains("ACID tables are not permitted when the \"hive.metastore.support.acid\" property is set to false"));
+  }
+}