You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ja...@apache.org on 2022/08/26 11:58:11 UTC
[flink] branch master updated: [FLINK-29035][table-planner] Fix bug of ExpressionReducer does not work with jar resources
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 0ade193d393 [FLINK-29035][table-planner] Fix bug of ExpressionReducer does not work with jar resources
0ade193d393 is described below
commit 0ade193d39326dd5b84334348a4b6ce76c4a915a
Author: Ron <ld...@163.com>
AuthorDate: Fri Aug 26 19:57:59 2022 +0800
[FLINK-29035][table-planner] Fix bug of ExpressionReducer does not work with jar resources
This closes #20635
---
.../table/planner/codegen/ExpressionReducer.scala | 2 +-
.../planner/runtime/batch/sql/FunctionITCase.java | 64 ----------------------
.../planner/runtime/stream/sql/FunctionITCase.java | 16 ++++++
3 files changed, 17 insertions(+), 65 deletions(-)
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/ExpressionReducer.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/ExpressionReducer.scala
index 91a21b5f4e3..fca0edbba5e 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/ExpressionReducer.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/ExpressionReducer.scala
@@ -94,7 +94,7 @@ class ExpressionReducer(
EMPTY_ROW_TYPE
)
- val function = generatedFunction.newInstance(Thread.currentThread().getContextClassLoader)
+ val function = generatedFunction.newInstance(classLoader)
val richMapFunction = function match {
case r: RichMapFunction[GenericRowData, GenericRowData] => r
case _ =>
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/batch/sql/FunctionITCase.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/batch/sql/FunctionITCase.java
index 59c95aa496f..7d886fe4fc6 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/batch/sql/FunctionITCase.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/batch/sql/FunctionITCase.java
@@ -18,20 +18,16 @@
package org.apache.flink.table.planner.runtime.batch.sql;
-import org.apache.flink.core.fs.Path;
import org.apache.flink.table.api.Table;
-import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.planner.factories.utils.TestCollectionTableFactory;
import org.apache.flink.table.planner.runtime.utils.BatchTestBase;
import org.apache.flink.types.Row;
-import org.apache.flink.util.CollectionUtil;
import org.apache.flink.util.UserClassLoaderJarTestUtils;
import org.junit.Before;
import org.junit.Test;
import java.util.Arrays;
-import java.util.Collections;
import java.util.List;
import java.util.Random;
import java.util.UUID;
@@ -88,32 +84,6 @@ public class FunctionITCase extends BatchTestBase {
assertThat(functions).doesNotContain("f11");
}
- @Test
- public void testCreateCatalogFunctionByUsingJar() {
- String ddl =
- String.format(
- "CREATE FUNCTION default_database.f11 AS '%s' USING JAR '%s'",
- udfClassName, jarPath);
- tEnv().executeSql(ddl);
- assertThat(Arrays.asList(tEnv().listFunctions())).contains("f11");
-
- tEnv().executeSql("DROP FUNCTION default_database.f11");
- assertThat(Arrays.asList(tEnv().listFunctions())).doesNotContain("f11");
- }
-
- @Test
- public void testCreateTemporaryCatalogFunctionByUsingJar() {
- String ddl =
- String.format(
- "CREATE TEMPORARY FUNCTION default_database.f12 AS '%s' USING JAR '%s'",
- udfClassName, jarPath);
- tEnv().executeSql(ddl);
- assertThat(Arrays.asList(tEnv().listFunctions())).contains("f12");
-
- tEnv().executeSql("DROP TEMPORARY FUNCTION default_database.f12");
- assertThat(Arrays.asList(tEnv().listFunctions())).doesNotContain("f12");
- }
-
@Test
public void testUserDefinedTemporarySystemFunctionByUsingJar() throws Exception {
String functionDDL =
@@ -125,40 +95,6 @@ public class FunctionITCase extends BatchTestBase {
testUserDefinedFunctionByUsingJar(functionDDL, dropFunctionDDL);
}
- @Test
- public void testUserDefinedRegularCatalogFunctionByUsingJar() throws Exception {
- String functionDDL =
- String.format(
- "create function lowerUdf as '%s' using jar '%s'", udfClassName, jarPath);
-
- String dropFunctionDDL = "drop function lowerUdf";
- testUserDefinedFunctionByUsingJar(functionDDL, dropFunctionDDL);
- }
-
- @Test
- public void testUserDefinedTemporaryCatalogFunctionByUsingJar() throws Exception {
- String functionDDL =
- String.format(
- "create temporary function lowerUdf as '%s' using jar '%s'",
- udfClassName, jarPath);
-
- String dropFunctionDDL = "drop temporary function lowerUdf";
- testUserDefinedFunctionByUsingJar(functionDDL, dropFunctionDDL);
- }
-
- @Test
- public void testUsingAddJar() throws Exception {
- tEnv().executeSql(String.format("ADD JAR '%s'", jarPath));
-
- TableResult tableResult = tEnv().executeSql("SHOW JARS");
- assertThat(CollectionUtil.iteratorToList(tableResult.collect()))
- .isEqualTo(Collections.singletonList(Row.of(new Path(jarPath).getPath())));
-
- testUserDefinedFunctionByUsingJar(
- String.format("create function lowerUdf as '%s' LANGUAGE JAVA", udfClassName),
- "drop function lowerUdf");
- }
-
private void testUserDefinedFunctionByUsingJar(String createFunctionDDL, String dropFunctionDDL)
throws Exception {
List<Row> sourceData =
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/sql/FunctionITCase.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/sql/FunctionITCase.java
index 6d6ddc23bb4..f85688c73be 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/sql/FunctionITCase.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/sql/FunctionITCase.java
@@ -474,6 +474,22 @@ public class FunctionITCase extends StreamingTestBase {
tEnv().executeSql(dropFunctionDDL);
}
+ @Test
+ public void testExpressionReducerByUsingJar() {
+ String functionDDL =
+ String.format(
+ "create temporary function lowerUdf as '%s' using jar '%s'",
+ udfClassName, jarPath);
+ tEnv().executeSql(functionDDL);
+
+ TableResult tableResult = tEnv().executeSql("SELECT lowerUdf('HELLO')");
+
+ List<Row> actualRows = CollectionUtil.iteratorToList(tableResult.collect());
+ assertThat(actualRows).isEqualTo(Arrays.asList(Row.of("hello")));
+
+ tEnv().executeSql("drop temporary function lowerUdf");
+ }
+
/** Test udf class. */
public static class TestUDF extends ScalarFunction {