You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by lz...@apache.org on 2020/11/03 07:07:43 UTC
[flink] branch release-1.11 updated:
[FLINK-16268][table-planner-blink] Failed to run rank over window with Hive
built-in functions
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch release-1.11
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.11 by this push:
new c677334 [FLINK-16268][table-planner-blink] Failed to run rank over window with Hive built-in functions
c677334 is described below
commit c677334727bc388fd9fd051c7aac0cf6e595d82c
Author: Rui Li <li...@apache.org>
AuthorDate: Tue Nov 3 15:06:49 2020 +0800
[FLINK-16268][table-planner-blink] Failed to run rank over window with Hive built-in functions
This closes #13857
---
.../flink/table/module/hive/HiveModuleTest.java | 19 +++++++++++++++++++
.../expressions/PlannerTypeInferenceUtilImpl.java | 21 +++++++++++++++++++++
.../planner/functions/utils/HiveFunctionUtils.java | 2 +-
.../functions/utils/UserDefinedFunctionUtils.scala | 6 +++++-
4 files changed, 46 insertions(+), 2 deletions(-)
diff --git a/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/table/module/hive/HiveModuleTest.java b/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/table/module/hive/HiveModuleTest.java
index 343d818..be8a75c 100644
--- a/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/table/module/hive/HiveModuleTest.java
+++ b/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/table/module/hive/HiveModuleTest.java
@@ -25,6 +25,7 @@ import org.apache.flink.table.functions.FunctionDefinition;
import org.apache.flink.table.functions.ScalarFunction;
import org.apache.flink.table.functions.ScalarFunctionDefinition;
import org.apache.flink.table.functions.hive.HiveSimpleUDF;
+import org.apache.flink.table.module.CoreModule;
import org.apache.flink.table.types.DataType;
import org.apache.flink.types.Row;
@@ -185,4 +186,22 @@ public class HiveModuleTest {
results = Lists.newArrayList(tableEnv.sqlQuery("select length('')").execute().collect());
assertEquals("[0]", results.toString());
}
+
+ @Test
+ // tests to verify we have set arguments for hive udf before trying to get result type
+ public void testHiveUDFSetArguments() throws Exception {
+ TableEnvironment tableEnv = HiveTestUtils.createTableEnvWithBlinkPlannerBatchMode();
+
+ tableEnv.unloadModule("core");
+ tableEnv.loadModule("hive", new HiveModule());
+ tableEnv.loadModule("core", CoreModule.INSTANCE);
+
+ String path = getClass().getResource("/csv/test.csv").toURI().toString();
+ tableEnv.executeSql(String.format(
+ "create table src(x int,y int) with ('connector'='filesystem','format'='csv','path'='%s')", path));
+
+ Lists.newArrayList(tableEnv.executeSql("select x from src where y is not null limit 10").collect());
+ Lists.newArrayList(tableEnv.executeSql("select count(distinct if(y is null, 0, y)) from src where x=-1 limit 1").collect());
+ Lists.newArrayList(tableEnv.executeSql("select x, rank() over (partition by x order by y) from src").collect());
+ }
}
diff --git a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/expressions/PlannerTypeInferenceUtilImpl.java b/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/expressions/PlannerTypeInferenceUtilImpl.java
index 4640377..d813ccb 100644
--- a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/expressions/PlannerTypeInferenceUtilImpl.java
+++ b/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/expressions/PlannerTypeInferenceUtilImpl.java
@@ -25,11 +25,15 @@ import org.apache.flink.table.delegation.PlannerTypeInferenceUtil;
import org.apache.flink.table.expressions.Expression;
import org.apache.flink.table.expressions.ResolvedExpression;
import org.apache.flink.table.expressions.UnresolvedCallExpression;
+import org.apache.flink.table.expressions.ValueLiteralExpression;
+import org.apache.flink.table.functions.ScalarFunction;
+import org.apache.flink.table.planner.functions.utils.HiveFunctionUtils;
import org.apache.flink.table.planner.typeutils.TypeCoercion;
import org.apache.flink.table.planner.validate.ValidationFailure;
import org.apache.flink.table.planner.validate.ValidationResult;
import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.inference.TypeInferenceUtil;
+import org.apache.flink.table.types.logical.LogicalType;
import java.util.ArrayList;
import java.util.List;
@@ -74,6 +78,23 @@ public final class PlannerTypeInferenceUtilImpl implements PlannerTypeInferenceU
.map(ResolvedExpression::getOutputDataType)
.collect(Collectors.toList());
+ if (plannerCall instanceof PlannerScalarFunctionCall) {
+ ScalarFunction scalarFunction = ((PlannerScalarFunctionCall) plannerCall).scalarFunction();
+ // need to set arg types for Hive functions
+ if (HiveFunctionUtils.isHiveFunc(scalarFunction)) {
+ LogicalType[] logicalTypes = expectedArgumentTypes.stream()
+ .map(DataType::getLogicalType).toArray(LogicalType[]::new);
+ Object[] constArgs = new Object[logicalTypes.length];
+ for (int i = 0; i < constArgs.length; i++) {
+ if (resolvedArgs.get(i) instanceof ValueLiteralExpression) {
+ ValueLiteralExpression literalExpression = (ValueLiteralExpression) resolvedArgs.get(i);
+ constArgs[i] = literalExpression.getValueAs(Object.class).orElse(null);
+ }
+ }
+ HiveFunctionUtils.invokeSetArgs(scalarFunction, constArgs, logicalTypes);
+ }
+ }
+
return new TypeInferenceUtil.Result(
expectedArgumentTypes,
null,
diff --git a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/functions/utils/HiveFunctionUtils.java b/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/functions/utils/HiveFunctionUtils.java
index 13a82cb..7939f47 100644
--- a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/functions/utils/HiveFunctionUtils.java
+++ b/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/functions/utils/HiveFunctionUtils.java
@@ -51,7 +51,7 @@ public class HiveFunctionUtils {
}
- static Serializable invokeSetArgs(
+ public static Serializable invokeSetArgs(
Serializable function, Object[] constantArguments, LogicalType[] argTypes) {
try {
// See hive HiveFunction
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/functions/utils/UserDefinedFunctionUtils.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/functions/utils/UserDefinedFunctionUtils.scala
index 471a39b..026ddc2 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/functions/utils/UserDefinedFunctionUtils.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/functions/utils/UserDefinedFunctionUtils.scala
@@ -467,7 +467,11 @@ object UserDefinedFunctionUtils {
displayName: String,
function: ScalarFunction,
typeFactory: FlinkTypeFactory): SqlFunction = {
- new ScalarSqlFunction(identifier, displayName, function, typeFactory)
+ if (HiveFunctionUtils.isHiveFunc(function)) {
+ new HiveScalarSqlFunction(identifier, function, typeFactory)
+ } else {
+ new ScalarSqlFunction(identifier, displayName, function, typeFactory)
+ }
}
/**