You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ku...@apache.org on 2019/06/25 15:08:00 UTC
[flink] branch master updated: [FLINK-12975][table-planner-blink]
UserDefinedFunctionUtils should distinguish overload any parameters methods
This is an automated email from the ASF dual-hosted git repository.
kurt 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 97d2876 [FLINK-12975][table-planner-blink] UserDefinedFunctionUtils should distinguish overload any parameters methods
97d2876 is described below
commit 97d28761add07a1c3569254302a1705e8128f91c
Author: Jingsong Lee <lz...@aliyun.com>
AuthorDate: Tue Jun 25 23:07:50 2019 +0800
[FLINK-12975][table-planner-blink] UserDefinedFunctionUtils should distinguish overload any parameters methods
This closes #8869
---
.../functions/utils/UserDefinedFunctionUtils.scala | 36 +++++++++++++---------
.../table/runtime/stream/sql/CorrelateITCase.scala | 29 +++++++++++++++++
2 files changed, 50 insertions(+), 15 deletions(-)
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/functions/utils/UserDefinedFunctionUtils.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/functions/utils/UserDefinedFunctionUtils.scala
index a231d20..082d17a 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/functions/utils/UserDefinedFunctionUtils.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/functions/utils/UserDefinedFunctionUtils.scala
@@ -19,15 +19,6 @@
package org.apache.flink.table.functions.utils
-import java.lang.reflect.{Method, Modifier}
-import java.lang.{Integer => JInt, Long => JLong}
-import java.sql.{Date, Time, Timestamp}
-
-import com.google.common.primitives.Primitives
-import org.apache.calcite.rel.`type`.{RelDataType, RelDataTypeFactory}
-import org.apache.calcite.rex.{RexLiteral, RexNode}
-import org.apache.calcite.sql.`type`.SqlTypeName
-import org.apache.calcite.sql.{SqlFunction, SqlOperatorBinding}
import org.apache.flink.api.common.functions.InvalidTypesException
import org.apache.flink.api.common.typeinfo.TypeInformation
import org.apache.flink.api.java.typeutils._
@@ -37,15 +28,26 @@ import org.apache.flink.table.dataformat.{BaseRow, BinaryString, Decimal}
import org.apache.flink.table.functions._
import org.apache.flink.table.plan.schema.DeferredTypeFlinkTableFunction
import org.apache.flink.table.types.ClassDataTypeConverter.fromClassToDataType
-import org.apache.flink.table.types.ClassLogicalTypeConverter.getInternalClassForType
+import org.apache.flink.table.types.ClassLogicalTypeConverter.{getDefaultExternalClassForType, getInternalClassForType}
+import org.apache.flink.table.types.DataType
import org.apache.flink.table.types.LogicalTypeDataTypeConverter.{fromDataTypeToLogicalType, fromLogicalTypeToDataType}
import org.apache.flink.table.types.TypeInfoLogicalTypeConverter.fromTypeInfoToLogicalType
import org.apache.flink.table.types.logical.{LogicalType, LogicalTypeRoot, RowType}
import org.apache.flink.table.types.utils.TypeConversions.fromLegacyInfoToDataType
-import org.apache.flink.table.types.{ClassLogicalTypeConverter, DataType}
+import org.apache.flink.table.typeutils.TypeCheckUtils.isAny
import org.apache.flink.types.Row
import org.apache.flink.util.InstantiationUtil
+import com.google.common.primitives.Primitives
+import org.apache.calcite.rel.`type`.{RelDataType, RelDataTypeFactory}
+import org.apache.calcite.rex.{RexLiteral, RexNode}
+import org.apache.calcite.sql.`type`.SqlTypeName
+import org.apache.calcite.sql.{SqlFunction, SqlOperatorBinding}
+
+import java.lang.reflect.{Method, Modifier}
+import java.lang.{Integer => JInt, Long => JLong}
+import java.sql.{Date, Time, Timestamp}
+
import scala.collection.JavaConversions._
import scala.collection.mutable
import scala.language.postfixOps
@@ -694,7 +696,7 @@ object UserDefinedFunctionUtils {
if (t == null) {
null
} else {
- ClassLogicalTypeConverter.getDefaultExternalClassForType(t)
+ getDefaultExternalClassForType(t)
}
}.toArray
@@ -729,9 +731,13 @@ object UserDefinedFunctionUtils {
internal: LogicalType,
parameterType: DataType): Boolean = {
val paraInternalType = fromDataTypeToLogicalType(parameterType)
- // There is a special equal to GenericType. We need rewrite type extract to BaseRow etc...
- paraInternalType == internal ||
- getInternalClassForType(internal) == getInternalClassForType(paraInternalType)
+ if (isAny(internal) && isAny(paraInternalType)) {
+ getDefaultExternalClassForType(internal) == getDefaultExternalClassForType(paraInternalType)
+ } else {
+ // There is a special equal to GenericType. We need rewrite type extract to BaseRow etc...
+ paraInternalType == internal ||
+ getInternalClassForType(internal) == getInternalClassForType(paraInternalType)
+ }
}
def getOperandType(callBinding: SqlOperatorBinding): Seq[LogicalType] = {
diff --git a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/runtime/stream/sql/CorrelateITCase.scala b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/runtime/stream/sql/CorrelateITCase.scala
index be8cd26..fa5ca18 100644
--- a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/runtime/stream/sql/CorrelateITCase.scala
+++ b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/runtime/stream/sql/CorrelateITCase.scala
@@ -17,11 +17,15 @@
*/
package org.apache.flink.table.runtime.stream.sql
+import java.lang.{Boolean => JBoolean}
+
import org.apache.flink.api.scala._
+import org.apache.flink.table.api.Types
import org.apache.flink.table.api.scala._
import org.apache.flink.table.runtime.utils.JavaUserDefinedScalarFunctions.UdfWithOpen
import org.apache.flink.table.runtime.utils.JavaUserDefinedTableFunctions.StringSplit
import org.apache.flink.table.runtime.utils.{StreamingTestBase, TestSinkUtil, TestingAppendSink, TestingAppendTableSink}
+import org.apache.flink.table.util.{RF, TableFunc7}
import org.apache.flink.types.Row
import org.junit.Assert.assertEquals
@@ -147,6 +151,31 @@ class CorrelateITCase extends StreamingTestBase {
(sink1.getAppendResults ++ sink2.getAppendResults).sorted)
}
+ @Test
+ def testMultipleEvals(): Unit = {
+ val row = Row.of(
+ 12.asInstanceOf[Integer],
+ true.asInstanceOf[JBoolean],
+ Row.of(1.asInstanceOf[Integer], 2.asInstanceOf[Integer], 3.asInstanceOf[Integer])
+ )
+
+ val rowType = Types.ROW(Types.INT, Types.BOOLEAN, Types.ROW(Types.INT, Types.INT, Types.INT))
+ val in = env.fromElements(row, row)(rowType).toTable(tEnv, 'a, 'b, 'c)
+
+ val sink = new TestingAppendSink
+
+ tEnv.registerTable("MyTable", in)
+ tEnv.registerFunction("rfFunc", new RF)
+ tEnv.registerFunction("tfFunc", new TableFunc7)
+ tEnv.sqlQuery(
+ "SELECT rfFunc(a) as d, e FROM MyTable, LATERAL TABLE(tfFunc(rfFunc(a))) as T(e)")
+ .toAppendStream[Row].addSink(sink)
+
+ env.execute()
+
+ assertEquals(List(), sink.getAppendResults.sorted)
+ }
+
@Ignore // TODO DATE_ADD
@Test
def testReUsePerRecord(): Unit = {