This is an automated email from the ASF dual-hosted git repository.
jark pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new f9bd7d2f553 [FLINK-28296][hive] Support the Hive UDAF which implement
GenericUDAFResolver interface (#20102)
f9bd7d2f553 is described below
commit f9bd7d2f553174c2ae25da63e3d5be156649cf27
Author: yuxia Luo <[email protected]>
AuthorDate: Fri Aug 5 21:16:51 2022 +0800
[FLINK-28296][hive] Support the Hive UDAF which implement
GenericUDAFResolver interface (#20102)
---
.../factories/HiveFunctionDefinitionFactory.java | 14 ++++++-
.../table/functions/hive/HiveGenericUDAF.java | 33 ++++++++--------
.../table/functions/hive/HiveGenericUDAFTest.java | 45 +++++++++++++++++++---
3 files changed, 69 insertions(+), 23 deletions(-)
diff --git
a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/factories/HiveFunctionDefinitionFactory.java
b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/factories/HiveFunctionDefinitionFactory.java
index d3cb9f39763..7c193573d80 100644
---
a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/factories/HiveFunctionDefinitionFactory.java
+++
b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/factories/HiveFunctionDefinitionFactory.java
@@ -35,6 +35,7 @@ import org.apache.flink.table.functions.hive.HiveSimpleUDF;
import org.apache.hadoop.hive.ql.exec.UDAF;
import org.apache.hadoop.hive.ql.exec.UDF;
+import org.apache.hadoop.hive.ql.udf.generic.GenericUDAFResolver;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDAFResolver2;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDF;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDTF;
@@ -120,19 +121,28 @@ public class HiveFunctionDefinitionFactory implements
FunctionDefinitionFactory
LOG.info("Transforming Hive function '{}' into a HiveGenericUDTF",
name);
return new HiveGenericUDTF(new HiveFunctionWrapper<>(functionClz),
hiveShim);
} else if (GenericUDAFResolver2.class.isAssignableFrom(functionClz)
+ || GenericUDAFResolver.class.isAssignableFrom(functionClz)
|| UDAF.class.isAssignableFrom(functionClz)) {
if (GenericUDAFResolver2.class.isAssignableFrom(functionClz)) {
LOG.info(
"Transforming Hive function '{}' into a
HiveGenericUDAF without UDAF bridging",
name);
- return new HiveGenericUDAF(new
HiveFunctionWrapper<>(functionClz), false, hiveShim);
+ return new HiveGenericUDAF(
+ new HiveFunctionWrapper<>(functionClz), false, true,
hiveShim);
+ } else if
(GenericUDAFResolver.class.isAssignableFrom(functionClz)) {
+ LOG.info(
+ "Transforming Hive function '{}' into a
HiveGenericUDAF without UDAF bridging",
+ name);
+ return new HiveGenericUDAF(
+ new HiveFunctionWrapper<>(functionClz), false, false,
hiveShim);
} else {
LOG.info(
"Transforming Hive function '{}' into a
HiveGenericUDAF with UDAF bridging",
name);
- return new HiveGenericUDAF(new
HiveFunctionWrapper<>(functionClz), true, hiveShim);
+ return new HiveGenericUDAF(
+ new HiveFunctionWrapper<>(functionClz), true, false,
hiveShim);
}
} else {
throw new IllegalArgumentException(
diff --git
a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/functions/hive/HiveGenericUDAF.java
b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/functions/hive/HiveGenericUDAF.java
index c15b5d85255..e396038077e 100644
---
a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/functions/hive/HiveGenericUDAF.java
+++
b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/functions/hive/HiveGenericUDAF.java
@@ -42,6 +42,7 @@ import
org.apache.hadoop.hive.ql.udf.generic.GenericUDAFBridge;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDAFEvaluator;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDAFResolver;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDAFResolver2;
+import org.apache.hadoop.hive.ql.udf.generic.SimpleGenericUDAFParameterInfo;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
import java.util.Arrays;
@@ -59,6 +60,8 @@ public class HiveGenericUDAF
// Flag that indicates whether a bridge between GenericUDAF and UDAF is
required.
// Old UDAF can be used with the GenericUDAF infrastructure through
bridging.
private final boolean isUDAFBridgeRequired;
+ // Flag that indicates whether the GenericUDAF implement
GenericUDAFResolver2
+ private final boolean isUDAFResolve2;
private final HiveShim hiveShim;
private HiveFunctionArguments arguments;
@@ -70,17 +73,14 @@ public class HiveGenericUDAF
private transient boolean allIdentityConverter;
private transient boolean initialized;
- public HiveGenericUDAF(
- HiveFunctionWrapper<GenericUDAFResolver> funcWrapper, HiveShim
hiveShim) {
- this(funcWrapper, false, hiveShim);
- }
-
public HiveGenericUDAF(
HiveFunctionWrapper<GenericUDAFResolver> funcWrapper,
boolean isUDAFBridgeRequired,
+ boolean isUDAFResolve2,
HiveShim hiveShim) {
this.hiveFunctionWrapper = funcWrapper;
this.isUDAFBridgeRequired = isUDAFBridgeRequired;
+ this.isUDAFResolve2 = isUDAFResolve2;
this.hiveShim = hiveShim;
}
@@ -125,15 +125,7 @@ public class HiveGenericUDAF
public GenericUDAFEvaluator createEvaluator(ObjectInspector[]
inputInspectors)
throws SemanticException {
- GenericUDAFResolver2 resolver;
-
- if (isUDAFBridgeRequired) {
- resolver = new GenericUDAFBridge((UDAF)
hiveFunctionWrapper.createFunction());
- } else {
- resolver = (GenericUDAFResolver2)
hiveFunctionWrapper.createFunction();
- }
-
- return resolver.getEvaluator(
+ SimpleGenericUDAFParameterInfo parameterInfo =
hiveShim.createUDAFParameterInfo(
inputInspectors,
// The flag to indicate if the UDAF invocation was
from the windowing
@@ -153,7 +145,18 @@ public class HiveGenericUDAF
// function implementation
// is not expected to ensure the wildcard handling of
the target relation.
// That is handled by the framework.
- Boolean.FALSE));
+ Boolean.FALSE);
+
+ if (isUDAFBridgeRequired) {
+ return new GenericUDAFBridge((UDAF)
hiveFunctionWrapper.createFunction())
+ .getEvaluator(parameterInfo);
+ } else if (isUDAFResolve2) {
+ return ((GenericUDAFResolver2)
hiveFunctionWrapper.createFunction())
+ .getEvaluator(parameterInfo);
+ } else {
+ // otherwise, it's instance of GenericUDAFResolver
+ return
hiveFunctionWrapper.createFunction().getEvaluator(parameterInfo.getParameters());
+ }
}
/**
diff --git
a/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/table/functions/hive/HiveGenericUDAFTest.java
b/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/table/functions/hive/HiveGenericUDAFTest.java
index 771d85cef9e..49ca87f3484 100644
---
a/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/table/functions/hive/HiveGenericUDAFTest.java
+++
b/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/table/functions/hive/HiveGenericUDAFTest.java
@@ -22,7 +22,9 @@ import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.catalog.hive.client.HiveShimLoader;
import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.inference.utils.CallContextMock;
+import org.apache.flink.types.Row;
+import org.apache.hadoop.hive.ql.udf.generic.GenericUDAFContextNGrams;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDAFCount;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDAFEvaluator;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDAFMin;
@@ -47,7 +49,7 @@ public class HiveGenericUDAFTest {
DataType[] argTypes = new DataType[] {DataTypes.BIGINT()};
- HiveGenericUDAF udf = init(GenericUDAFMin.class, constantArgs,
argTypes);
+ HiveGenericUDAF udf = init(GenericUDAFMin.class, constantArgs,
argTypes, false, true);
GenericUDAFEvaluator.AggregationBuffer acc = udf.createAccumulator();
@@ -66,7 +68,7 @@ public class HiveGenericUDAFTest {
DataType[] argTypes = new DataType[] {DataTypes.DOUBLE()};
- HiveGenericUDAF udf = init(GenericUDAFSum.class, constantArgs,
argTypes);
+ HiveGenericUDAF udf = init(GenericUDAFSum.class, constantArgs,
argTypes, false, true);
GenericUDAFEvaluator.AggregationBuffer acc = udf.createAccumulator();
@@ -82,7 +84,7 @@ public class HiveGenericUDAFTest {
argTypes = new DataType[] {DataTypes.DECIMAL(5, 3)};
- udf = init(GenericUDAFSum.class, constantArgs, argTypes);
+ udf = init(GenericUDAFSum.class, constantArgs, argTypes, false, true);
acc = udf.createAccumulator();
@@ -101,7 +103,7 @@ public class HiveGenericUDAFTest {
DataType[] argTypes = new DataType[] {DataTypes.DOUBLE()};
- HiveGenericUDAF udf = init(GenericUDAFCount.class, constantArgs,
argTypes);
+ HiveGenericUDAF udf = init(GenericUDAFCount.class, constantArgs,
argTypes, false, true);
GenericUDAFEvaluator.AggregationBuffer acc = udf.createAccumulator();
@@ -114,8 +116,36 @@ public class HiveGenericUDAFTest {
assertThat(udf.getValue(acc)).isEqualTo(3L);
}
+ @Test
+ public void testUDAFResolver() throws Exception {
+ // test the Hive's UDAF that implement deprecated GenericUDAFResolver
interface
+ Object[] constantArgs = new Object[] {null, null, null};
+ DataType[] argTypes =
+ new DataType[] {
+ DataTypes.ARRAY(DataTypes.STRING()),
+ DataTypes.ARRAY(DataTypes.STRING()),
+ DataTypes.INT()
+ };
+
+ HiveGenericUDAF udaf =
+ init(GenericUDAFContextNGrams.class, constantArgs, argTypes,
false, false);
+
+ GenericUDAFEvaluator.AggregationBuffer acc = udaf.createAccumulator();
+
+ udaf.accumulate(
+ acc, new Object[] {"what", "i", "think"}, new Object[]
{"what", "i", null}, 1);
+
+ udaf.merge(acc, Collections.emptyList());
+ assertThat(Arrays.toString((Row[])
udaf.getValue(acc))).isEqualTo("[+I[[think], 1.0]]");
+ }
+
private static HiveGenericUDAF init(
- Class<?> hiveUdfClass, Object[] constantArgs, DataType[] argTypes)
throws Exception {
+ Class<?> hiveUdfClass,
+ Object[] constantArgs,
+ DataType[] argTypes,
+ boolean isUDAFBridgeRequired,
+ boolean isUDAFResolve2)
+ throws Exception {
HiveFunctionWrapper<GenericUDAFResolver> wrapper = new
HiveFunctionWrapper<>(hiveUdfClass);
CallContextMock callContext = new CallContextMock();
@@ -127,7 +157,10 @@ public class HiveGenericUDAFTest {
HiveGenericUDAF udf =
new HiveGenericUDAF(
- wrapper,
HiveShimLoader.loadHiveShim(HiveShimLoader.getHiveVersion()));
+ wrapper,
+ isUDAFBridgeRequired,
+ isUDAFResolve2,
+
HiveShimLoader.loadHiveShim(HiveShimLoader.getHiveVersion()));
udf.setArguments(callContext);
udf.inferReturnType();
