HIVE-29403: Move the truncate table out of HMSHandler

Author: dengzhhu653

standalone-metastore/metastore-common/src/main/java/org/apache/hadoop/hive/metastore/utils/JavaUtils.java

@@ -128,6 +128,18 @@ public static <T> void setField(T req, String methodName, Class[] argsCls, Objec
     }
   }
 
+  public static <T> T getField(Object payload, String methodName) {
+    try {
+      Method method = payload.getClass().getDeclaredMethod(methodName);
+      method.setAccessible(true);
+      return (T) method.invoke(payload);
+    } catch (Exception e) {
+      LOG.error("Unable to invoke the method: {} of the instance: {}, message: {}",
+          methodName, payload, e.getMessage());
+      throw new RuntimeException(e);
+    }
+  }
+
   /**
    * Utility method for ACID to normalize logging info.  Matches
    * org.apache.hadoop.hive.metastore.api.LockRequest#toString

standalone-metastore/metastore-server/pom.xml

@@ -23,6 +23,7 @@
   <name>Hive Metastore Server</name>
   <properties>
     <standalone.metastore.path.to.root>..</standalone.metastore.path.to.root>
+    <reflections.version>0.10.2</reflections.version>
   </properties>
   <dependencies>
     <dependency>
@@ -70,6 +71,17 @@
         </exclusion>
       </exclusions>
     </dependency>
+    <dependency>
+      <groupId>org.reflections</groupId>
+      <artifactId>reflections</artifactId>
+      <version>${reflections.version}</version>
+      <exclusions>
+        <exclusion>
+          <groupId>com.google.code.findbugs</groupId>
+          <artifactId>annotations</artifactId>
+        </exclusion>
+      </exclusions>
+    </dependency>
     <dependency>
       <groupId>com.fasterxml.jackson.core</groupId>
       <artifactId>jackson-databind</artifactId>

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

@@ -68,10 +68,10 @@
 import java.util.LinkedList;
 import java.util.Optional;
 
-import static org.apache.hadoop.hive.metastore.HMSHandler.addTruncateBaseFile;
 import static org.apache.hadoop.hive.metastore.HiveMetaHook.ALTERLOCATION;
 import static org.apache.hadoop.hive.metastore.HiveMetaHook.ALTER_TABLE_OPERATION_TYPE;
 import static org.apache.hadoop.hive.metastore.HiveMetaStoreClient.RENAME_PARTITION_MAKE_COPY;
+import static org.apache.hadoop.hive.metastore.handler.TruncateTableHandler.addTruncateBaseFile;
 import static org.apache.hadoop.hive.metastore.utils.MetaStoreServerUtils.findStaleColumns;
 import static org.apache.hadoop.hive.metastore.utils.MetaStoreServerUtils.isDbReplicationTarget;
 import static org.apache.hadoop.hive.metastore.utils.MetaStoreUtils.getDefaultCatalog;

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

@@ -23,7 +23,9 @@
 import com.google.common.util.concurrent.MoreExecutors;
 
 import java.io.IOException;
+import java.lang.reflect.Modifier;
 import java.util.Map;
+import java.util.Set;
 import java.util.UUID;
 import java.util.concurrent.CancellationException;
 import java.util.concurrent.ConcurrentHashMap;
@@ -35,29 +37,30 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
 
+import org.apache.commons.lang3.StringUtils;
 import org.apache.hadoop.hive.metastore.IHMSHandler;
-import org.apache.hadoop.hive.metastore.api.AddPartitionsRequest;
 import org.apache.hadoop.hive.metastore.api.AsyncOperationResp;
-import org.apache.hadoop.hive.metastore.api.DropDatabaseRequest;
-import org.apache.hadoop.hive.metastore.api.DropPartitionsRequest;
-import org.apache.hadoop.hive.metastore.api.DropTableRequest;
 import org.apache.hadoop.hive.metastore.api.MetaException;
 import org.apache.hadoop.hive.metastore.conf.MetastoreConf;
 import org.apache.hadoop.hive.metastore.metrics.Metrics;
 import org.apache.hadoop.hive.metastore.metrics.MetricsConstants;
 import org.apache.thrift.TBase;
 import org.apache.thrift.TException;
+import org.reflections.Reflections;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import static org.apache.hadoop.hive.metastore.ExceptionHandler.handleException;
 import static org.apache.hadoop.hive.metastore.conf.MetastoreConf.ConfVars.HIVE_IN_TEST;
+import static org.apache.hadoop.hive.metastore.utils.JavaUtils.getField;
+import static org.apache.hadoop.hive.metastore.utils.JavaUtils.newInstance;
 
-public abstract class AbstractOperationHandler<T extends TBase, A extends AbstractOperationHandler.Result> {
-
-  private static final Logger LOG = LoggerFactory.getLogger(AbstractOperationHandler.class);
-  private static final Map<String, AbstractOperationHandler> OPID_TO_HANDLER = new ConcurrentHashMap<>();
+public abstract class AbstractRequestHandler<T extends TBase, A extends AbstractRequestHandler.Result> {
+  private static final Logger LOG = LoggerFactory.getLogger(AbstractRequestHandler.class);
+  private static final Map<String, AbstractRequestHandler> OPID_TO_HANDLER = new ConcurrentHashMap<>();
+  private static final AtomicLong ID = new AtomicLong(0);
   private static final ScheduledExecutorService OPID_CLEANER = Executors.newScheduledThreadPool(1, r -> {
     Thread thread = new Thread(r);
     thread.setDaemon(true);
@@ -67,33 +70,31 @@ public abstract class AbstractOperationHandler<T extends TBase, A extends Abstra
 
   private static final Map<Class<? extends TBase>, HandlerFactory> REQ_FACTORIES = new ConcurrentHashMap<>();
   static {
-    REQ_FACTORIES.put(DropTableRequest.class, (base, request) -> {
-      DropTableRequest req = (DropTableRequest) request;
-      AbstractOperationHandler opHandler = ofCache(req.getId(), req.isCancel());
-      if (opHandler == null) {
-        opHandler = new DropTableHandler(base, req);
+    Set<Class<? extends AbstractRequestHandler>> handlerClasses =
+        new Reflections("org.apache.hadoop.hive.metastore.handler").getSubTypesOf(AbstractRequestHandler.class);
+    for (Class<? extends AbstractRequestHandler> clz : handlerClasses) {
+      if (Modifier.isAbstract(clz.getModifiers())) {
+        continue;
       }
-      return opHandler;
-    });
-
-    REQ_FACTORIES.put(DropDatabaseRequest.class, (base, request) -> {
-      DropDatabaseRequest req = (DropDatabaseRequest) request;
-      AbstractOperationHandler opHandler = ofCache(req.getId(), req.isCancel());
-      if (opHandler == null) {
-        opHandler = new DropDatabaseHandler(base, req);
+      RequestHandler handler = clz.getAnnotation(RequestHandler.class);
+      Class<? extends TBase> requestBody;
+      if (handler == null || (requestBody = handler.requestBody()) == null) {
+        continue;
       }
-      return opHandler;
-    });
-
-    REQ_FACTORIES.put(DropPartitionsRequest.class, (base, request) -> {
-      DropPartitionsRequest req = (DropPartitionsRequest) request;
-      return new DropPartitionsHandler(base, req);
-    });
-
-    REQ_FACTORIES.put(AddPartitionsRequest.class, (base, request) -> {
-      AddPartitionsRequest req = (AddPartitionsRequest) request;
-      return new AddPartitionsHandler(base, req);
-    });
+      validateHandler(clz, handler);
+      REQ_FACTORIES.put(requestBody, (base, request) -> {
+        AbstractRequestHandler opHandler = null;
+        if (handler.supportAsync()) {
+          opHandler =
+              ofCache(getField(request, handler.id()), getField(request, handler.cancel()));
+        }
+        if (opHandler == null) {
+          opHandler =
+              newInstance(clz, new Class[]{IHMSHandler.class, requestBody}, new Object[]{base, request});
+        }
+        return opHandler;
+      });
+    }
   }
 
   private Result result;
@@ -107,19 +108,19 @@ public abstract class AbstractOperationHandler<T extends TBase, A extends Abstra
   protected final String id;
   private long timeout;
 
-  private AbstractOperationHandler(String id) {
+  private AbstractRequestHandler(String id) {
     this.id = id;
   }
 
-  AbstractOperationHandler(IHMSHandler handler, boolean async, T request) {
-    this.id = UUID.randomUUID().toString();
+  AbstractRequestHandler(IHMSHandler handler, boolean async, T request) {
+    this.id = UUID.randomUUID() + "-" + ID.incrementAndGet();
     this.handler = handler;
     this.request = request;
     this.async = async;
     this.timeout = MetastoreConf.getBoolVar(handler.getConf(), HIVE_IN_TEST) ? 10 : 5000;
     final Timer.Context timerContext;
-    if (getHandlerAlias() != null) {
-      Timer timer = Metrics.getOrCreateTimer(MetricsConstants.API_PREFIX + getHandlerAlias());
+    if (StringUtils.isNotEmpty(getMetricAlias())) {
+      Timer timer = Metrics.getOrCreateTimer(MetricsConstants.API_PREFIX + getMetricAlias());
       timerContext = timer != null ? timer.time() : null;
     } else {
       timerContext = null;
@@ -159,9 +160,9 @@ private AbstractOperationHandler(String id) {
     this.executor.shutdown();
   }
 
-  private static <T extends TBase, A extends Result> AbstractOperationHandler<T, A>
+  private static <T extends TBase, A extends Result> AbstractRequestHandler<T, A>
       ofCache(String opId, boolean shouldCancel) throws TException {
-    AbstractOperationHandler<T, A> opHandler = null;
+    AbstractRequestHandler<T, A> opHandler = null;
     if (opId != null) {
       opHandler = OPID_TO_HANDLER.get(opId);
       if (opHandler == null && !shouldCancel) {
@@ -171,7 +172,7 @@ private AbstractOperationHandler(String id) {
         if (opHandler != null) {
           opHandler.cancelOperation();
         } else {
-          opHandler = new AbstractOperationHandler<>(opId) {
+          opHandler = new AbstractRequestHandler<>(opId) {
             @Override
             public OperationStatus getOperationStatus() throws TException {
               OperationStatus resp = new OperationStatus(opId);
@@ -198,7 +199,7 @@ public String getProgress() {
     return opHandler;
   }
 
-  public static <T extends AbstractOperationHandler> T offer(IHMSHandler handler, TBase req)
+  public static <T extends AbstractRequestHandler> T offer(IHMSHandler handler, TBase req)
       throws TException, IOException {
     HandlerFactory factory = REQ_FACTORIES.get(req.getClass());
     if (factory != null) {
@@ -242,37 +243,22 @@ public OperationStatus getOperationStatus() throws TException {
   }
 
   public static class OperationStatus {
-    private final String id;
-    private String message;
-    private boolean finished;
+    final String id;
+    String message;
+    boolean finished;
     OperationStatus(String id) {
       this.id = id;
     }
-
-    public String getMessage() {
-      return message;
-    }
-
     public void setMessage(String message) {
       this.message = message;
     }
-
-    public String getId() {
-      return id;
-    }
-
-    public boolean isFinished() {
-      return finished;
-    }
-
     public void setFinished(boolean finished) {
       this.finished = finished;
     }
-
     public AsyncOperationResp toAsyncOperationResp() {
-      AsyncOperationResp resp = new AsyncOperationResp(getId());
-      resp.setFinished(isFinished());
-      resp.setMessage(getMessage());
+      AsyncOperationResp resp = new AsyncOperationResp(id);
+      resp.setFinished(finished);
+      resp.setMessage(message);
       return resp;
     }
   }
@@ -294,7 +280,7 @@ public void cancelOperation() {
    */
   public final A getResult() throws TException {
     OperationStatus resp = getOperationStatus();
-    if (!resp.isFinished()) {
+    if (!resp.finished) {
       throw new IllegalStateException("Result is un-available as " +
           getMessagePrefix() + " is still running");
     }
@@ -335,15 +321,16 @@ protected void beforeExecute() throws TException, IOException {
 
   public boolean success() throws TException {
     OperationStatus status = getOperationStatus();
-    return status.isFinished() && result != null && result.success();
+    return status.finished && result != null && result.success();
   }
 
   /**
    * Get the alias of this handler for metrics.
-   * @return the alias, null if no need to measure the operation.
+   * @return the alias, null or empty if no need to measure the operation.
    */
-  protected String getHandlerAlias() {
-    return null;
+  private String getMetricAlias() {
+    RequestHandler handler = getClass().getAnnotation(RequestHandler.class);
+    return handler != null ? handler.metricAlias() : null;
   }
 
   public void checkInterrupted() throws MetaException {
@@ -366,8 +353,8 @@ public static boolean containsOp(String opId) {
     return OPID_TO_HANDLER.containsKey(opId);
   }
 
-  public interface HandlerFactory {
-    AbstractOperationHandler create(IHMSHandler base, TBase req) throws TException, IOException;
+  interface HandlerFactory {
+    AbstractRequestHandler create(IHMSHandler base, TBase req) throws TException, IOException;
   }
 
   public interface Result {
@@ -386,4 +373,19 @@ default Result shrinkIfNecessary() {
       return this;
     }
   }
+
+  private static void validateHandler(Class<? extends AbstractRequestHandler> clz,
+      RequestHandler handler) {
+    try {
+      Class<? extends TBase> requestBody = handler.requestBody();
+      // Check the constructor
+      clz.getDeclaredConstructor(IHMSHandler.class, requestBody);
+      if (handler.supportAsync()) {
+        requestBody.getMethod(handler.id());
+        requestBody.getMethod(handler.cancel());
+      }
+    } catch (Exception e) {
+      throw new RuntimeException(clz + " is not a satisfied handler as it's declared to be", e);
+    }
+  }
 }

…re/handler/AbstractOperationHandler.java …tore/handler/AbstractRequestHandler.javastandalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/handler/AbstractOperationHandler.java renamed to standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/handler/AbstractRequestHandler.java standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/handler/AbstractOperationHandler.java renamed to standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/handler/AbstractRequestHandler.java

@@ -76,8 +76,9 @@
 import static org.apache.hadoop.hive.metastore.utils.MetaStoreUtils.getDefaultCatalog;
 import static org.apache.hadoop.hive.metastore.utils.StringUtils.normalizeIdentifier;
 
+@RequestHandler(requestBody = AddPartitionsRequest.class)
 public class AddPartitionsHandler
-    extends AbstractOperationHandler<AddPartitionsRequest, AddPartitionsHandler.AddPartitionsResult> {
+    extends AbstractRequestHandler<AddPartitionsRequest, AddPartitionsHandler.AddPartitionsResult> {
   private TableName tableName;
   private Table table;
   private Database db;

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

@@ -67,8 +67,9 @@
 import static org.apache.hadoop.hive.metastore.utils.MetaStoreServerUtils.isDbReplicationTarget;
 import static org.apache.hadoop.hive.metastore.utils.StringUtils.normalizeIdentifier;
 
+@RequestHandler(requestBody = DropDatabaseRequest.class, supportAsync = true, metricAlias = "drop_database_req")
 public class DropDatabaseHandler
-    extends AbstractOperationHandler<DropDatabaseRequest, DropDatabaseHandler.DropDatabaseResult> {
+    extends AbstractRequestHandler<DropDatabaseRequest, DropDatabaseHandler.DropDatabaseResult> {
 
   private static final Logger LOG = LoggerFactory.getLogger(DropDatabaseHandler.class);
   private String name;
@@ -143,7 +144,7 @@ public DropDatabaseResult execute() throws TException, IOException {
         dropRequest.setDeleteData(false);
         dropRequest.setDropPartitions(true);
         dropRequest.setAsyncDrop(false);
-        DropTableHandler dropTable = AbstractOperationHandler.offer(handler, dropRequest);
+        DropTableHandler dropTable = AbstractRequestHandler.offer(handler, dropRequest);
         if (tableDataShouldBeDeleted
             && dropTable.success()) {
           DropTableHandler.DropTableResult dropTableResult = dropTable.getResult();
@@ -404,11 +405,6 @@ public Result shrinkIfNecessary() {
     }
   }
 
-  @Override
-  protected String getHandlerAlias() {
-    return "drop_database_req";
-  }
-
   @Override
   protected void afterExecute(DropDatabaseResult result) throws MetaException, IOException {
     try {

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

@@ -60,15 +60,16 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import static org.apache.hadoop.hive.metastore.HMSHandler.addTruncateBaseFile;
+import static org.apache.hadoop.hive.metastore.handler.TruncateTableHandler.addTruncateBaseFile;
 import static org.apache.hadoop.hive.metastore.utils.MetaStoreServerUtils.checkTableDataShouldBeDeleted;
 import static org.apache.hadoop.hive.metastore.utils.MetaStoreServerUtils.getWriteId;
 import static org.apache.hadoop.hive.metastore.utils.MetaStoreServerUtils.isMustPurge;
 import static org.apache.hadoop.hive.metastore.utils.MetaStoreUtils.getDefaultCatalog;
 import static org.apache.hadoop.hive.metastore.utils.StringUtils.normalizeIdentifier;
 
+@RequestHandler(requestBody = DropPartitionsRequest.class)
 public class DropPartitionsHandler
-    extends AbstractOperationHandler<DropPartitionsRequest, DropPartitionsHandler.DropPartitionsResult> {
+    extends AbstractRequestHandler<DropPartitionsRequest, DropPartitionsHandler.DropPartitionsResult> {
   private static final Logger LOG = LoggerFactory.getLogger(DropPartitionsHandler.class);
   private TableName tableName;
   private Table table;