HIVE-25043: Addendum: Support custom UDF in Vectorized mode (#5703)

Author: ryukobayashi

common/src/java/org/apache/hadoop/hive/conf/HiveConf.java

@@ -4702,9 +4702,9 @@ public static enum ConfVars {
         "The default value is true."),
 
     HIVE_VECTOR_ADAPTOR_CUSTOM_UDF_WHITELIST("hive.vectorized.adaptor.custom.udf.whitelist", "",
-        "Custom UDF allowed when hive.vectorized.adaptor.usage.mode is chosen.\n" +
-        "Specify classes separated by commas:\n" +
-        "package.FooClass,package.BarClass"),
+        "A comma-separated list of custom UDFs allowed to operate in vectorized mode " +
+        "when hive.vectorized.adaptor.usage.mode is set to chosen.\n" +
+        "Only Generic UDFs are supported for whitelisting; ensure that each custom UDF class extends GenericUDF"),
 
     HIVE_VECTORIZATION_PTF_MAX_MEMORY_BUFFERING_BATCH_COUNT("hive.vectorized.ptf.max.memory.buffering.batch.count", 25,
         "Maximum number of vectorized row batches to buffer in memory for PTF\n" +

data/conf/llap-udfs.lst

@@ -0,0 +1 @@
+org.apache.hadoop.hive.ql.udf.generic.GenericUDFCustomDateSub

itests/src/test/resources/testconfiguration.properties

@@ -71,6 +71,7 @@ minillap.query.files=\
   cte_2.q,\
   cte_4.q,\
   cttl.q,\
+  custom_udf_vectorization.q,\
   dynamic_partition_pruning_2.q,\
   dynamic_semijoin_user_level.q,\
   dynpart_cast.q,\

itests/util/src/main/java/org/apache/hadoop/hive/ql/udf/generic/GenericUDFCustomDateSub.java

@@ -0,0 +1,55 @@
+/*
+ * 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.ql.udf.generic;
+
+import org.apache.hadoop.hive.ql.exec.Description;
+import org.apache.hadoop.hive.ql.exec.vector.VectorizedExpressions;
+import org.apache.hadoop.hive.ql.exec.vector.expressions.VectorUDFDateSubColCol;
+import org.apache.hadoop.hive.ql.exec.vector.expressions.VectorUDFDateSubColScalar;
+import org.apache.hadoop.hive.ql.exec.vector.expressions.VectorUDFDateSubScalarCol;
+
+/**
+ * UDFDateSub.
+ *
+ * Subtract a number of days to the date. The time part of the string will be
+ * ignored.
+ *
+ * NOTE: This is a subset of what MySQL offers as:
+ * http://dev.mysql.com/doc/refman
+ * /5.1/en/date-and-time-functions.html#function_date-sub
+ *
+ */
+@Description(name = "date_sub",
+    value = "_FUNC_(start_date, num_days) - Returns the date that is num_days before start_date.",
+    extended = "start_date is a string in the format 'yyyy-MM-dd HH:mm:ss' or"
+        + " 'yyyy-MM-dd'. num_days is a number. The time part of start_date is "
+        + "ignored.\n"
+        + "Example:\n "
+        + "  > SELECT _FUNC_('2009-07-30', 1) FROM src LIMIT 1;\n"
+        + "  '2009-07-29'")
+@VectorizedExpressions({VectorUDFDateSubColScalar.class, VectorUDFDateSubScalarCol.class, VectorUDFDateSubColCol.class})
+public class GenericUDFCustomDateSub extends GenericUDFDateAdd {
+  public GenericUDFCustomDateSub() {
+    this.signModifier = -1;
+  }
+
+  @Override
+  public String getDisplayString(String[] children) {
+    return getStandardDisplayString("date_sub", children);
+  }
+}

ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorizationContext.java

@@ -277,18 +277,6 @@ public static HiveVectorIfStmtMode getHiveConfValue(HiveConf hiveConf) {
 
   private HiveVectorIfStmtMode hiveVectorIfStmtMode;
 
-  private Set<String> allowedCustomUDFs;
-
-  private Set<String> getAllowedCustomUDFs(HiveConf hiveConf) {
-    String udfs = HiveConf.getVar(hiveConf,
-        HiveConf.ConfVars.HIVE_VECTOR_ADAPTOR_CUSTOM_UDF_WHITELIST);
-    if (udfs != null && !udfs.isEmpty()) {
-      return new HashSet<>(Arrays.asList(udfs.split(",")));
-    }
-
-    return new HashSet<>();
-  }
-
   //when set to true use the overflow checked vector expressions
   private boolean useCheckedVectorExpressions;
 
@@ -310,7 +298,6 @@ private void setHiveConfVars(HiveConf hiveConf) {
     adaptorSuppressEvaluateExceptions =
         HiveConf.getBoolVar(
             hiveConf, HiveConf.ConfVars.HIVE_VECTORIZED_ADAPTOR_SUPPRESS_EVALUATE_EXCEPTIONS);
-    this.allowedCustomUDFs = getAllowedCustomUDFs(hiveConf);
   }
 
   private void copyHiveConfVars(VectorizationContext vContextEnvironment) {
@@ -1050,7 +1037,7 @@ public VectorExpression getVectorExpression(ExprNodeDesc exprDesc, VectorExpress
                 "Could not vectorize expression (mode = " + mode.name() + "): " + exprDesc.toString()
                   + " because hive.vectorized.adaptor.usage.mode=none");
           case CHOSEN:
-            if (isNonVectorizedPathUDF(expr, mode, allowedCustomUDFs)) {
+            if (isNonVectorizedPathUDF(expr, mode)) {
               ve = getCustomUDFExpression(expr, mode);
             } else {
               throw new HiveException(
@@ -1460,7 +1447,7 @@ public static GenericUDF getGenericUDFForCast(TypeInfo castType) throws HiveExce
    * may be implemented in the future with an optimized VectorExpression.
    */
   private static boolean isNonVectorizedPathUDF(ExprNodeGenericFuncDesc expr,
-      VectorExpressionDescriptor.Mode mode, Set<String> allowCustomUDFs) {
+      VectorExpressionDescriptor.Mode mode) {
     GenericUDF gudf = expr.getGenericUDF();
     if (gudf instanceof GenericUDFBridge) {
       GenericUDFBridge bridge = (GenericUDFBridge) gudf;
@@ -1498,8 +1485,6 @@ private static boolean isNonVectorizedPathUDF(ExprNodeGenericFuncDesc expr,
       return true;
     } else if (gudf instanceof GenericUDFConcat && (mode == VectorExpressionDescriptor.Mode.PROJECTION)) {
       return true;
-    } else if (allowCustomUDFs.contains(gudf.getClass().getName())) {
-      return true;
     }
     return false;
   }

ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/Vectorizer.java

@@ -178,6 +178,7 @@
 import org.apache.hadoop.hive.ql.plan.ptf.WindowFrameDef;
 import org.apache.hadoop.hive.ql.plan.ptf.WindowFunctionDef;
 import org.apache.hadoop.hive.ql.plan.ptf.WindowTableFunctionDef;
+import org.apache.hadoop.hive.ql.session.SessionState;
 import org.apache.hadoop.hive.ql.udf.UDFAcos;
 import org.apache.hadoop.hive.ql.udf.UDFAsin;
 import org.apache.hadoop.hive.ql.udf.UDFAtan;
@@ -550,6 +551,9 @@ public Vectorizer() {
 
     // For conditional expressions
     supportedGenericUDFs.add(GenericUDFIf.class);
+
+    // Add user custom UDFs
+    addCustomUDFs(SessionState.getSessionConf());
   }
 
   private class VectorTaskColumnInfo {
@@ -2411,6 +2415,40 @@ private boolean getOnlyStructObjectInspectors(ReduceWork reduceWork,
     }
   }
 
+  private void addCustomUDFs(HiveConf hiveConf) {
+    if (hiveConf == null) {
+      return;
+    }
+
+    if (HiveVectorAdaptorUsageMode.CHOSEN !=
+        HiveVectorAdaptorUsageMode.getHiveConfValue(hiveConf)) {
+      return;
+    }
+
+    String[] udfs =
+        HiveConf.getTrimmedStringsVar(hiveConf,
+            HiveConf.ConfVars.HIVE_VECTOR_ADAPTOR_CUSTOM_UDF_WHITELIST);
+    if (udfs == null) {
+      return;
+    }
+
+    ClassLoader loader = Utilities.getSessionSpecifiedClassLoader();
+
+    for (String udf : udfs) {
+      try {
+        Class<?> cls = Class.forName(udf, true, loader);
+        if (GenericUDF.class.isAssignableFrom(cls)) {
+          supportedGenericUDFs.add(cls);
+          LOG.info("Registered custom UDF: {}", udf);
+        } else {
+          LOG.warn("{} must inherit from the GenericUDF", udf);
+        }
+      } catch (ClassNotFoundException e) {
+        LOG.warn("Failed to register custom UDF: {}", udf, e);
+      }
+    }
+  }
+
   @Override
   public PhysicalContext resolve(PhysicalContext physicalContext) throws SemanticException {
 

ql/src/test/queries/clientpositive/custom_udf_vectorization.q

@@ -0,0 +1,24 @@
+--! qt:dataset:src
+set hive.mapred.mode=nonstrict;
+set hive.explain.user=false;
+set hive.vectorized.execution.enabled=true;
+set hive.vectorized.adaptor.usage.mode=chosen;
+set hive.llap.execution.mode=auto;
+
+CREATE TEMPORARY FUNCTION CDATE_SUB AS 'org.apache.hadoop.hive.ql.udf.generic.GenericUDFCustomDateSub';
+
+set hive.vectorized.adaptor.custom.udf.whitelist=org.apache.hadoop.hive.ql.udf.generic.GenericUDFCustomDateSub;
+
+EXPLAIN VECTORIZATION
+SELECT CDATE_SUB('2000-01-01', 1) FROM src
+UNION ALL
+SELECT CDATE_SUB('2000-01-01', CAST(key as int)) FROM src
+UNION ALL
+SELECT CDATE_SUB('2000-01-01', 1) FROM src WHERE key IS NOT NULL
+UNION ALL
+SELECT CDATE_SUB('2000-01-01', CAST(key as int)) FROM src WHERE key Is NOT NULL
+UNION ALL
+SELECT CDATE_SUB('2000-01-01', 1) FROM src WHERE key <> '' AND value <> ''
+UNION ALL
+SELECT CDATE_SUB('2000-01-01', CAST(key as int)) FROM src WHERE key <> '' AND value <> '';
+