[flink] branch release-1.10 updated: [FLINK-17152][table] Fix FunctionDefinitionUtil generate wrong resultType and acc type for AggregateFunctionDefinition

[jark]

This is an automated email from the ASF dual-hosted git repository.

jark pushed a commit to branch release-1.10
in repository https://gitbox.apache.org/repos/asf/flink.git


The following commit(s) were added to refs/heads/release-1.10 by this push:
     new cc8faa4  [FLINK-17152][table] Fix FunctionDefinitionUtil generate 
wrong resultType and acc type for AggregateFunctionDefinition
cc8faa4 is described below

commit cc8faa4fdc7ba6b2761b4c4154a0d8b661880914
Author: Terry Wang <zjuwa...@foxmail.com>
AuthorDate: Thu Apr 16 11:58:34 2020 +0800

    [FLINK-17152][table] Fix FunctionDefinitionUtil generate wrong resultType 
and acc type for AggregateFunctionDefinition
    
    This closes #11748
---
 .../flink/table/functions/FunctionDefinitionUtil.java    |  8 ++++----
 .../table/functions/FunctionDefinitionUtilTest.java      | 16 +++++++++++-----
 2 files changed, 15 insertions(+), 9 deletions(-)

diff --git 
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/functions/FunctionDefinitionUtil.java
 
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/functions/FunctionDefinitionUtil.java
index 374b138..1719ace 100644
--- 
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/functions/FunctionDefinitionUtil.java
+++ 
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/functions/FunctionDefinitionUtil.java
@@ -53,8 +53,8 @@ public class FunctionDefinitionUtil {
                        return new AggregateFunctionDefinition(
                                name,
                                a,
-                               
UserDefinedFunctionHelper.getAccumulatorTypeOfAggregateFunction(a),
-                               
UserDefinedFunctionHelper.getReturnTypeOfAggregateFunction(a)
+                               
UserDefinedFunctionHelper.getReturnTypeOfAggregateFunction(a),
+                               
UserDefinedFunctionHelper.getAccumulatorTypeOfAggregateFunction(a)
                        );
                } else if (udf instanceof TableAggregateFunction) {
                        TableAggregateFunction a = (TableAggregateFunction) udf;
@@ -62,8 +62,8 @@ public class FunctionDefinitionUtil {
                        return new TableAggregateFunctionDefinition(
                                name,
                                a,
-                               
UserDefinedFunctionHelper.getAccumulatorTypeOfAggregateFunction(a),
-                               
UserDefinedFunctionHelper.getReturnTypeOfAggregateFunction(a)
+                               
UserDefinedFunctionHelper.getReturnTypeOfAggregateFunction(a),
+                               
UserDefinedFunctionHelper.getAccumulatorTypeOfAggregateFunction(a)
                        );
                } else {
                        throw new UnsupportedOperationException(
diff --git 
a/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/functions/FunctionDefinitionUtilTest.java
 
b/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/functions/FunctionDefinitionUtilTest.java
index 691734c..83497a8 100644
--- 
a/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/functions/FunctionDefinitionUtilTest.java
+++ 
b/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/functions/FunctionDefinitionUtilTest.java
@@ -19,9 +19,11 @@
 package org.apache.flink.table.functions;
 
 import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeinfo.Types;
 
 import org.junit.Test;
 
+import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;
 
 /**
@@ -71,6 +73,8 @@ public class FunctionDefinitionUtilTest {
 
                assertTrue(((AggregateFunctionDefinition) 
fd2).getAggregateFunction()
                                instanceof TestAggFunctionWithoutResultType);
+               assertEquals(((AggregateFunctionDefinition) 
fd2).getResultTypeInfo(), Types.LONG);
+               assertEquals(((AggregateFunctionDefinition) 
fd2).getAccumulatorTypeInfo(), Types.STRING);
 
        }
 
@@ -90,6 +94,8 @@ public class FunctionDefinitionUtilTest {
 
                assertTrue(((TableAggregateFunctionDefinition) 
fd2).getTableAggregateFunction()
                                instanceof 
TestTableAggFunctionWithoutResultType);
+               assertEquals(((TableAggregateFunctionDefinition) 
fd2).getResultTypeInfo(), Types.LONG);
+               assertEquals(((TableAggregateFunctionDefinition) 
fd2).getAccumulatorTypeInfo(), Types.STRING);
        }
 
        /**
@@ -143,14 +149,14 @@ public class FunctionDefinitionUtilTest {
        /**
         * Test function.
         */
-       public static class TestAggFunctionWithoutResultType extends 
AggregateFunction<Long, Long> {
+       public static class TestAggFunctionWithoutResultType extends 
AggregateFunction<Long, String> {
                @Override
-               public Long createAccumulator() {
+               public String createAccumulator() {
                        return null;
                }
 
                @Override
-               public Long getValue(Long accumulator) {
+               public Long getValue(String accumulator) {
                        return null;
                }
        }
@@ -178,9 +184,9 @@ public class FunctionDefinitionUtilTest {
        /**
         * Test function.
         */
-       public static class TestTableAggFunctionWithoutResultType extends 
TableAggregateFunction<Long, Long> {
+       public static class TestTableAggFunctionWithoutResultType extends 
TableAggregateFunction<Long, String> {
                @Override
-               public Long createAccumulator() {
+               public String createAccumulator() {
                        return null;
                }
        }