[flink] branch master updated: [FLINK-12217][table] OperationTreeBuilder.map() should perform ExpressionResolver.resolve()

[dw...@apache.org]

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

dwysakowicz 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 7fba616  [FLINK-12217][table] OperationTreeBuilder.map() should perform ExpressionResolver.resolve()
7fba616 is described below

commit 7fba616857a6a71c6a818c008e59af94bc933375
Author: hequn8128 <ch...@gmail.com>
AuthorDate: Wed Apr 17 12:29:10 2019 +0800

    [FLINK-12217][table] OperationTreeBuilder.map() should perform ExpressionResolver.resolve()
---
 .../flink/table/operations/OperationTreeBuilder.scala       |  7 +++++--
 .../stream/table/stringexpr/CalcStringExpressionTest.scala  | 13 +++++++++++++
 2 files changed, 18 insertions(+), 2 deletions(-)

diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/operations/OperationTreeBuilder.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/operations/OperationTreeBuilder.scala
index 1579eb4..c6fec01 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/operations/OperationTreeBuilder.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/operations/OperationTreeBuilder.scala
@@ -348,12 +348,15 @@ class OperationTreeBuilder(private val tableEnv: TableEnvironment) {
 
   def map(mapFunction: Expression, child: TableOperation): TableOperation = {
 
-    if (!isScalarFunction(mapFunction)) {
+    val resolver = resolverFor(tableCatalog, functionCatalog, child).build()
+    val resolvedMapFunction = resolveSingleExpression(mapFunction, resolver)
+
+    if (!isScalarFunction(resolvedMapFunction)) {
       throw new ValidationException("Only ScalarFunction can be used in the map operator.")
     }
 
     val expandedFields = new CallExpression(BuiltInFunctionDefinitions.FLATTEN,
-      List(mapFunction).asJava)
+      List(resolvedMapFunction).asJava)
     project(Collections.singletonList(expandedFields), child)
   }
 
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/stream/table/stringexpr/CalcStringExpressionTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/stream/table/stringexpr/CalcStringExpressionTest.scala
index 0fd5528..999c81b 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/stream/table/stringexpr/CalcStringExpressionTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/stream/table/stringexpr/CalcStringExpressionTest.scala
@@ -21,6 +21,7 @@ package org.apache.flink.table.api.stream.table.stringexpr
 import org.apache.flink.api.scala._
 import org.apache.flink.table.api.scala._
 import org.apache.flink.table.expressions.Literal
+import org.apache.flink.table.expressions.utils.Func23
 import org.apache.flink.table.utils.TableTestBase
 import org.junit.Test
 
@@ -167,4 +168,16 @@ class CalcStringExpressionTest extends TableTestBase {
 
     verifyTableEquals(t1, t2)
   }
+
+  @Test
+  def testMap(): Unit = {
+    val util = streamTestUtil()
+    val t = util.addTable[(Int, Long, String)]("Table3",'a, 'b, 'c)
+    util.tableEnv.registerFunction("func", Func23)
+
+    val t1 = t.map("func(a, b, c)")
+    val t2 = t.map(Func23('a, 'b, 'c))
+
+    verifyTableEquals(t1, t2)
+  }
 }