[FLINK-7490] [table] Use correct classloader to compile generated cod…

…e that calls UDAGGs. This closes apache#5018.
InfinitiesLoop · Nov 15, 2017 · 59df4b7 · 59df4b7
1 parent 54eeccf
commit 59df4b7
Show file tree

Hide file tree

Showing 13 changed files with 13 additions and 12 deletions.
diff --git a/...ava/org/apache/flink/api/java/operators/translation/RichCombineToGroupCombineWrapper.java b/...ava/org/apache/flink/api/java/operators/translation/RichCombineToGroupCombineWrapper.java
@@ -42,6 +42,7 @@ public RichCombineToGroupCombineWrapper(F wrappedFunction) {
 
  @Override
  public void open(Configuration config) throws Exception {
+ wrappedFunction.setRuntimeContext(getRuntimeContext());
  wrappedFunction.open(config);
  }
 

diff --git a/...-table/src/main/scala/org/apache/flink/table/runtime/aggregate/AggregateAggFunction.scala b/...-table/src/main/scala/org/apache/flink/table/runtime/aggregate/AggregateAggFunction.scala
@@ -70,7 +70,7 @@ class AggregateAggFunction(genAggregations: GeneratedAggregationsFunction)
  LOG.debug(s"Compiling AggregateHelper: $genAggregations.name \n\n " +
  s"Code:\n$genAggregations.code")
  val clazz = compile(
- getClass.getClassLoader,
+ Thread.currentThread().getContextClassLoader,
  genAggregations.name,
  genAggregations.code)
  LOG.debug("Instantiating AggregateHelper.")

diff --git a/...nk-table/src/main/scala/org/apache/flink/table/runtime/aggregate/DataSetAggFunction.scala b/...nk-table/src/main/scala/org/apache/flink/table/runtime/aggregate/DataSetAggFunction.scala
@@ -46,7 +46,7 @@ class DataSetAggFunction(
  LOG.debug(s"Compiling AggregateHelper: $genAggregations.name \n\n " +
  s"Code:\n$genAggregations.code")
  val clazz = compile(
- getClass.getClassLoader,
+ getRuntimeContext.getUserCodeClassLoader,
  genAggregations.name,
  genAggregations.code)
  LOG.debug("Instantiating AggregateHelper.")

diff --git a/...ble/src/main/scala/org/apache/flink/table/runtime/aggregate/DataSetFinalAggFunction.scala b/...ble/src/main/scala/org/apache/flink/table/runtime/aggregate/DataSetFinalAggFunction.scala
@@ -47,7 +47,7 @@ class DataSetFinalAggFunction(
  LOG.debug(s"Compiling AggregateHelper: $genAggregations.name \n\n " +
  s"Code:\n$genAggregations.code")
  val clazz = compile(
- getClass.getClassLoader,
+ getRuntimeContext.getUserCodeClassLoader,
  genAggregations.name,
  genAggregations.code)
  LOG.debug("Instantiating AggregateHelper.")

diff --git a/...table/src/main/scala/org/apache/flink/table/runtime/aggregate/DataSetPreAggFunction.scala b/...table/src/main/scala/org/apache/flink/table/runtime/aggregate/DataSetPreAggFunction.scala
@@ -48,7 +48,7 @@ class DataSetPreAggFunction(genAggregations: GeneratedAggregationsFunction)
  LOG.debug(s"Compiling AggregateHelper: $genAggregations.name \n\n " +
  s"Code:\n$genAggregations.code")
  val clazz = compile(
- getClass.getClassLoader,
+ getRuntimeContext.getUserCodeClassLoader,
  genAggregations.name,
  genAggregations.code)
  LOG.debug("Instantiating AggregateHelper.")

diff --git a/...org/apache/flink/table/runtime/aggregate/DataSetSessionWindowAggReduceGroupFunction.scala b/...org/apache/flink/table/runtime/aggregate/DataSetSessionWindowAggReduceGroupFunction.scala
@@ -72,7 +72,7 @@ class DataSetSessionWindowAggReduceGroupFunction(
  LOG.debug(s"Compiling AggregateHelper: $genAggregations.name \n\n " +
  s"Code:\n$genAggregations.code")
  val clazz = compile(
- getClass.getClassLoader,
+ getRuntimeContext.getUserCodeClassLoader,
  genAggregations.name,
  genAggregations.code)
  LOG.debug("Instantiating AggregateHelper.")

diff --git a/.../org/apache/flink/table/runtime/aggregate/DataSetSessionWindowAggregatePreProcessor.scala b/.../org/apache/flink/table/runtime/aggregate/DataSetSessionWindowAggregatePreProcessor.scala
@@ -59,7 +59,7 @@ class DataSetSessionWindowAggregatePreProcessor(
  LOG.debug(s"Compiling AggregateHelper: $genAggregations.name \n\n " +
  s"Code:\n$genAggregations.code")
  val clazz = compile(
- getClass.getClassLoader,
+ getRuntimeContext.getUserCodeClassLoader,
  genAggregations.name,
  genAggregations.code)
  LOG.debug("Instantiating AggregateHelper.")

diff --git a/...g/apache/flink/table/runtime/aggregate/DataSetSlideTimeWindowAggReduceGroupFunction.scala b/...g/apache/flink/table/runtime/aggregate/DataSetSlideTimeWindowAggReduceGroupFunction.scala
@@ -68,7 +68,7 @@ class DataSetSlideTimeWindowAggReduceGroupFunction(
  LOG.debug(s"Compiling AggregateHelper: $genAggregations.name \n\n " +
  s"Code:\n$genAggregations.code")
  val clazz = compile(
- getClass.getClassLoader,
+ getRuntimeContext.getUserCodeClassLoader,
  genAggregations.name,
  genAggregations.code)
  LOG.debug("Instantiating AggregateHelper.")

diff --git a/...org/apache/flink/table/runtime/aggregate/DataSetSlideWindowAggReduceCombineFunction.scala b/...org/apache/flink/table/runtime/aggregate/DataSetSlideWindowAggReduceCombineFunction.scala
@@ -67,7 +67,7 @@ class DataSetSlideWindowAggReduceCombineFunction(
  LOG.debug(s"Compiling AggregateHelper: $genPreAggregations.name \n\n " +
  s"Code:\n$genPreAggregations.code")
  val clazz = compile(
- getClass.getClassLoader,
+ getRuntimeContext.getUserCodeClassLoader,
  genPreAggregations.name,
  genPreAggregations.code)
  LOG.debug("Instantiating AggregateHelper.")

diff --git a/...a/org/apache/flink/table/runtime/aggregate/DataSetSlideWindowAggReduceGroupFunction.scala b/...a/org/apache/flink/table/runtime/aggregate/DataSetSlideWindowAggReduceGroupFunction.scala
@@ -63,7 +63,7 @@ class DataSetSlideWindowAggReduceGroupFunction(
  LOG.debug(s"Compiling AggregateHelper: $genAggregations.name \n\n " +
  s"Code:\n$genAggregations.code")
  val clazz = compile(
- getClass.getClassLoader,
+ getRuntimeContext.getUserCodeClassLoader,
  genAggregations.name,
  genAggregations.code)
  LOG.debug("Instantiating AggregateHelper.")

diff --git a/...apache/flink/table/runtime/aggregate/DataSetTumbleCountWindowAggReduceGroupFunction.scala b/...apache/flink/table/runtime/aggregate/DataSetTumbleCountWindowAggReduceGroupFunction.scala
@@ -50,7 +50,7 @@ class DataSetTumbleCountWindowAggReduceGroupFunction(
  LOG.debug(s"Compiling AggregateHelper: $genAggregations.name \n\n " +
  s"Code:\n$genAggregations.code")
  val clazz = compile(
- getClass.getClassLoader,
+ getRuntimeContext.getUserCodeClassLoader,
  genAggregations.name,
  genAggregations.code)
  LOG.debug("Instantiating AggregateHelper.")

diff --git a/...pache/flink/table/runtime/aggregate/DataSetTumbleTimeWindowAggReduceCombineFunction.scala b/...pache/flink/table/runtime/aggregate/DataSetTumbleTimeWindowAggReduceCombineFunction.scala
@@ -66,7 +66,7 @@ class DataSetTumbleTimeWindowAggReduceCombineFunction(
  LOG.debug(s"Compiling AggregateHelper: $genPreAggregations.name \n\n " +
  s"Code:\n$genPreAggregations.code")
  val clazz = compile(
- getClass.getClassLoader,
+ getRuntimeContext.getUserCodeClassLoader,
  genPreAggregations.name,
  genPreAggregations.code)
  LOG.debug("Instantiating AggregateHelper.")

diff --git a/.../apache/flink/table/runtime/aggregate/DataSetTumbleTimeWindowAggReduceGroupFunction.scala b/.../apache/flink/table/runtime/aggregate/DataSetTumbleTimeWindowAggReduceGroupFunction.scala
@@ -62,7 +62,7 @@ class DataSetTumbleTimeWindowAggReduceGroupFunction(
  LOG.debug(s"Compiling AggregateHelper: $genAggregations.name \n\n " +
  s"Code:\n$genAggregations.code")
  val clazz = compile(
- getClass.getClassLoader,
+ getRuntimeContext.getUserCodeClassLoader,
  genAggregations.name,
  genAggregations.code)
  LOG.debug("Instantiating AggregateHelper.")