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
commit abf3091e9da92660490e600b7605f3e2d2d53ad4
Author: Dawid Wysakowicz <dwysakow...@apache.org>
AuthorDate: Fri Jul 3 16:30:34 2020 +0200
[FLINK-18419] Make user ClassLoader available in TableEnvironment
---
.../table/client/gateway/local/ExecutionContext.java | 9 ++++++---
.../bridge/java/internal/StreamTableEnvironmentImpl.java | 16 +++++++++++++---
.../java/internal/StreamTableEnvironmentImplTest.java | 3 ++-
.../flink/table/api/internal/TableEnvironmentImpl.java | 8 ++++++--
.../apache/flink/table/utils/TableEnvironmentMock.java | 10 +++++++++-
.../scala/internal/StreamTableEnvironmentImpl.scala | 9 ++++++---
.../scala/internal/StreamTableEnvironmentImplTest.scala | 3 ++-
.../apache/flink/table/planner/utils/TableTestBase.scala | 9 ++++++---
.../table/api/stream/StreamTableEnvironmentTest.scala | 3 ++-
.../flink/table/api/stream/sql/AggregateTest.scala | 3 ++-
.../org/apache/flink/table/utils/TableTestBase.scala | 6 ++++--
11 files changed, 58 insertions(+), 21 deletions(-)
diff --git
a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/ExecutionContext.java
b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/ExecutionContext.java
index 0a444eb2..60e7736 100644
---
a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/ExecutionContext.java
+++
b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/ExecutionContext.java
@@ -424,7 +424,8 @@ public class ExecutionContext<ClusterID> {
Executor executor,
CatalogManager catalogManager,
ModuleManager moduleManager,
- FunctionCatalog functionCatalog) {
+ FunctionCatalog functionCatalog,
+ ClassLoader userClassLoader) {
final Map<String, String> plannerProperties =
settings.toPlannerProperties();
final Planner planner =
ComponentFactoryService.find(PlannerFactory.class, plannerProperties)
@@ -438,7 +439,8 @@ public class ExecutionContext<ClusterID> {
env,
planner,
executor,
- settings.isStreamingMode());
+ settings.isStreamingMode(),
+ userClassLoader);
}
private static Executor lookupExecutor(
@@ -599,7 +601,8 @@ public class ExecutionContext<ClusterID> {
executor,
catalogManager,
moduleManager,
- functionCatalog);
+ functionCatalog,
+ classLoader);
} else if (environment.getExecution().isBatchPlanner()) {
streamExecEnv = null;
execEnv =
ExecutionEnvironment.getExecutionEnvironment();
diff --git
a/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/table/api/bridge/java/internal/StreamTableEnvironmentImpl.java
b/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/table/api/bridge/java/internal/StreamTableEnvironmentImpl.java
index 15877ad..05c2f20 100644
---
a/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/table/api/bridge/java/internal/StreamTableEnvironmentImpl.java
+++
b/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/table/api/bridge/java/internal/StreamTableEnvironmentImpl.java
@@ -89,8 +89,17 @@ public final class StreamTableEnvironmentImpl extends
TableEnvironmentImpl imple
StreamExecutionEnvironment executionEnvironment,
Planner planner,
Executor executor,
- boolean isStreamingMode) {
- super(catalogManager, moduleManager, tableConfig, executor,
functionCatalog, planner, isStreamingMode);
+ boolean isStreamingMode,
+ ClassLoader userClassLoader) {
+ super(
+ catalogManager,
+ moduleManager,
+ tableConfig,
+ executor,
+ functionCatalog,
+ planner,
+ isStreamingMode,
+ userClassLoader);
this.executionEnvironment = executionEnvironment;
}
@@ -137,7 +146,8 @@ public final class StreamTableEnvironmentImpl extends
TableEnvironmentImpl imple
executionEnvironment,
planner,
executor,
- settings.isStreamingMode()
+ settings.isStreamingMode(),
+ classLoader
);
}
diff --git
a/flink-table/flink-table-api-java-bridge/src/test/java/org/apache/flink/table/api/bridge/java/internal/StreamTableEnvironmentImplTest.java
b/flink-table/flink-table-api-java-bridge/src/test/java/org/apache/flink/table/api/bridge/java/internal/StreamTableEnvironmentImplTest.java
index 954fc45..7fcbf80 100644
---
a/flink-table/flink-table-api-java-bridge/src/test/java/org/apache/flink/table/api/bridge/java/internal/StreamTableEnvironmentImplTest.java
+++
b/flink-table/flink-table-api-java-bridge/src/test/java/org/apache/flink/table/api/bridge/java/internal/StreamTableEnvironmentImplTest.java
@@ -101,7 +101,8 @@ public class StreamTableEnvironmentImplTest {
env,
new TestPlanner(elements.getTransformation()),
new ExecutorMock(),
- true
+ true,
+ this.getClass().getClassLoader()
);
}
diff --git
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java
index 05c578b..954edb5 100644
---
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java
+++
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java
@@ -164,6 +164,7 @@ public class TableEnvironmentImpl implements
TableEnvironmentInternal {
protected final Planner planner;
protected final Parser parser;
private final boolean isStreamingMode;
+ private final ClassLoader userClassLoader;
private static final String UNSUPPORTED_QUERY_IN_SQL_UPDATE_MSG =
"Unsupported SQL query! sqlUpdate() only accepts a
single SQL statement of type " +
"INSERT, CREATE TABLE, DROP TABLE, ALTER TABLE, USE
CATALOG, USE [CATALOG.]DATABASE, " +
@@ -197,7 +198,8 @@ public class TableEnvironmentImpl implements
TableEnvironmentInternal {
Executor executor,
FunctionCatalog functionCatalog,
Planner planner,
- boolean isStreamingMode) {
+ boolean isStreamingMode,
+ ClassLoader userClassLoader) {
this.catalogManager = catalogManager;
this.catalogManager.setCatalogTableSchemaResolver(
new
CatalogTableSchemaResolver(planner.getParser(), isStreamingMode));
@@ -210,6 +212,7 @@ public class TableEnvironmentImpl implements
TableEnvironmentInternal {
this.planner = planner;
this.parser = planner.getParser();
this.isStreamingMode = isStreamingMode;
+ this.userClassLoader = userClassLoader;
this.operationTreeBuilder = OperationTreeBuilder.create(
tableConfig,
functionCatalog.asLookup(parser::parseIdentifier),
@@ -272,7 +275,8 @@ public class TableEnvironmentImpl implements
TableEnvironmentInternal {
executor,
functionCatalog,
planner,
- settings.isStreamingMode()
+ settings.isStreamingMode(),
+ classLoader
);
}
diff --git
a/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/utils/TableEnvironmentMock.java
b/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/utils/TableEnvironmentMock.java
index dcbec07..5d0f021 100644
---
a/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/utils/TableEnvironmentMock.java
+++
b/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/utils/TableEnvironmentMock.java
@@ -46,7 +46,15 @@ public class TableEnvironmentMock extends
TableEnvironmentImpl {
FunctionCatalog functionCatalog,
PlannerMock planner,
boolean isStreamingMode) {
- super(catalogManager, moduleManager, tableConfig, executor,
functionCatalog, planner, isStreamingMode);
+ super(
+ catalogManager,
+ moduleManager,
+ tableConfig,
+ executor,
+ functionCatalog,
+ planner,
+ isStreamingMode,
+ TableEnvironmentMock.class.getClassLoader());
this.catalogManager = catalogManager;
this.executor = executor;
diff --git
a/flink-table/flink-table-api-scala-bridge/src/main/scala/org/apache/flink/table/api/bridge/scala/internal/StreamTableEnvironmentImpl.scala
b/flink-table/flink-table-api-scala-bridge/src/main/scala/org/apache/flink/table/api/bridge/scala/internal/StreamTableEnvironmentImpl.scala
index 7c99e37..317cb6a 100644
---
a/flink-table/flink-table-api-scala-bridge/src/main/scala/org/apache/flink/table/api/bridge/scala/internal/StreamTableEnvironmentImpl.scala
+++
b/flink-table/flink-table-api-scala-bridge/src/main/scala/org/apache/flink/table/api/bridge/scala/internal/StreamTableEnvironmentImpl.scala
@@ -58,7 +58,8 @@ class StreamTableEnvironmentImpl (
scalaExecutionEnvironment: StreamExecutionEnvironment,
planner: Planner,
executor: Executor,
- isStreaming: Boolean)
+ isStreaming: Boolean,
+ userClassLoader: ClassLoader)
extends TableEnvironmentImpl(
catalogManager,
moduleManager,
@@ -66,7 +67,8 @@ class StreamTableEnvironmentImpl (
executor,
functionCatalog,
planner,
- isStreaming)
+ isStreaming,
+ userClassLoader)
with org.apache.flink.table.api.bridge.scala.StreamTableEnvironment {
override def fromDataStream[T](dataStream: DataStream[T]): Table = {
@@ -299,7 +301,8 @@ object StreamTableEnvironmentImpl {
executionEnvironment,
planner,
executor,
- settings.isStreamingMode
+ settings.isStreamingMode,
+ classLoader
)
}
diff --git
a/flink-table/flink-table-api-scala-bridge/src/test/scala/org/apache/flink/table/api/bridge/scala/internal/StreamTableEnvironmentImplTest.scala
b/flink-table/flink-table-api-scala-bridge/src/test/scala/org/apache/flink/table/api/bridge/scala/internal/StreamTableEnvironmentImplTest.scala
index 0f84d9b..7f67ebb 100644
---
a/flink-table/flink-table-api-scala-bridge/src/test/scala/org/apache/flink/table/api/bridge/scala/internal/StreamTableEnvironmentImplTest.scala
+++
b/flink-table/flink-table-api-scala-bridge/src/test/scala/org/apache/flink/table/api/bridge/scala/internal/StreamTableEnvironmentImplTest.scala
@@ -91,7 +91,8 @@ class StreamTableEnvironmentImplTest {
env,
new TestPlanner(elements.javaStream.getTransformation),
new ExecutorMock,
- true)
+ true,
+ this.getClass.getClassLoader)
}
private class TestPlanner(transformation: Transformation[_]) extends
PlannerMock {
diff --git
a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/utils/TableTestBase.scala
b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/utils/TableTestBase.scala
index 05b54c7..c942911 100644
---
a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/utils/TableTestBase.scala
+++
b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/utils/TableTestBase.scala
@@ -959,7 +959,8 @@ class TestingTableEnvironment private(
executor: Executor,
functionCatalog: FunctionCatalog,
planner: PlannerBase,
- isStreamingMode: Boolean)
+ isStreamingMode: Boolean,
+ userClassLoader: ClassLoader)
extends TableEnvironmentImpl(
catalogManager,
moduleManager,
@@ -967,7 +968,8 @@ class TestingTableEnvironment private(
executor,
functionCatalog,
planner,
- isStreamingMode) {
+ isStreamingMode,
+ userClassLoader) {
// just for testing, remove this method while
// `<T, ACC> void registerFunction(String name, AggregateFunction<T, ACC>
aggregateFunction);`
@@ -1118,7 +1120,8 @@ object TestingTableEnvironment {
executor,
functionCatalog,
planner,
- settings.isStreamingMode)
+ settings.isStreamingMode,
+ classLoader)
}
}
diff --git
a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/stream/StreamTableEnvironmentTest.scala
b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/stream/StreamTableEnvironmentTest.scala
index 5ef1158..406e6b4 100644
---
a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/stream/StreamTableEnvironmentTest.scala
+++
b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/stream/StreamTableEnvironmentTest.scala
@@ -425,7 +425,8 @@ class StreamTableEnvironmentTest extends TableTestBase {
jStreamExecEnv,
streamPlanner,
executor,
- true)
+ true,
+ Thread.currentThread().getContextClassLoader)
val sType = new TupleTypeInfo(Types.LONG, Types.INT, Types.STRING,
Types.INT, Types.LONG)
.asInstanceOf[TupleTypeInfo[JTuple5[JLong, JInt, String, JInt, JLong]]]
diff --git
a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/stream/sql/AggregateTest.scala
b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/stream/sql/AggregateTest.scala
index f7da2d6..835a26f 100644
---
a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/stream/sql/AggregateTest.scala
+++
b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/stream/sql/AggregateTest.scala
@@ -79,7 +79,8 @@ class AggregateTest extends TableTestBase {
Mockito.mock(classOf[StreamExecutionEnvironment]),
new PlannerMock,
Mockito.mock(classOf[Executor]),
- true
+ true,
+ Thread.currentThread().getContextClassLoader
)
tablEnv.registerFunction("udag", new MyAgg)
diff --git
a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/utils/TableTestBase.scala
b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/utils/TableTestBase.scala
index 44e0bb9..562a240 100644
---
a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/utils/TableTestBase.scala
+++
b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/utils/TableTestBase.scala
@@ -347,7 +347,8 @@ case class StreamTableTestUtil(
javaEnv,
streamPlanner,
executor,
- true)
+ true,
+ Thread.currentThread().getContextClassLoader)
val env = new StreamExecutionEnvironment(javaEnv)
val tableEnv = new ScalaStreamTableEnvironmentImpl(
@@ -358,7 +359,8 @@ case class StreamTableTestUtil(
env,
streamPlanner,
executor,
- true)
+ true,
+ Thread.currentThread().getContextClassLoader)
def addTable[T: TypeInformation](
name: String,
