You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by pn...@apache.org on 2019/05/09 12:34:53 UTC
[flink] 02/02: [FLINK-11974][table-planner-blink] Use
StreamOperatorFactory in Blink runtime
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 7b3678f41276c2be69975762f9331744ad373cae
Author: JingsongLi <lz...@aliyun.com>
AuthorDate: Mon May 6 17:07:32 2019 +0800
[FLINK-11974][table-planner-blink] Use StreamOperatorFactory in Blink runtime
---
.../flink/table/codegen/CalcCodeGenerator.scala | 6 +-
.../table/codegen/CorrelateCodeGenerator.scala | 6 +-
.../flink/table/codegen/ExpandCodeGenerator.scala | 8 +-
.../table/codegen/LongHashJoinGenerator.scala | 6 +-
.../table/codegen/OperatorCodeGenerator.scala | 18 ++-
.../flink/table/codegen/SinkCodeGenerator.scala | 7 +-
.../batch/BatchExecHashAggregateBase.scala | 4 +-
.../batch/BatchExecHashWindowAggregateBase.scala | 4 +-
.../batch/BatchExecSortAggregateBase.scala | 4 +-
.../batch/BatchExecSortWindowAggregateBase.scala | 4 +-
.../apache/flink/table/plan/util/ScanUtil.scala | 4 +-
.../table/codegen/LongHashJoinGeneratorTest.java | 3 +-
.../codegen/agg/batch/AggWithoutKeysTest.scala | 7 +-
.../table/codegen/agg/batch/BatchAggTestBase.scala | 14 +-
.../agg/batch/HashAggCodeGeneratorTest.scala | 7 +-
.../agg/batch/SortAggCodeGeneratorTest.scala | 7 +-
.../table/runtime/batch/sql/join/JoinITCase.scala | 12 +-
.../flink/table/generated/GeneratedClass.java | 10 ++
.../table/runtime/CodeGenOperatorFactory.java | 62 ++++++++
.../table/runtime/OneInputOperatorWrapper.java | 155 ------------------
.../table/runtime/TwoInputOperatorWrapper.java | 175 ---------------------
.../runtime/join/Int2HashJoinOperatorTest.java | 30 ++--
22 files changed, 147 insertions(+), 406 deletions(-)
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/codegen/CalcCodeGenerator.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/codegen/CalcCodeGenerator.scala
index e4c364b..fec7397 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/codegen/CalcCodeGenerator.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/codegen/CalcCodeGenerator.scala
@@ -21,7 +21,7 @@ import org.apache.flink.streaming.api.transformations.StreamTransformation
import org.apache.flink.table.`type`.{RowType, TypeConverters}
import org.apache.flink.table.api.{TableConfig, TableException}
import org.apache.flink.table.dataformat.{BaseRow, BoxedWrapperRow}
-import org.apache.flink.table.runtime.OneInputOperatorWrapper
+import org.apache.flink.table.runtime.CodeGenOperatorFactory
import org.apache.calcite.plan.RelOptCluster
import org.apache.calcite.rex._
@@ -39,7 +39,7 @@ object CalcCodeGenerator {
calcProgram: RexProgram,
condition: Option[RexNode],
retainHeader: Boolean = false,
- opName: String): OneInputOperatorWrapper[BaseRow, BaseRow] = {
+ opName: String): CodeGenOperatorFactory[BaseRow] = {
val inputType = TypeConverters.createInternalTypeFromTypeInfo(
inputTransform.getOutputType).asInstanceOf[RowType]
// filter out time attributes
@@ -66,7 +66,7 @@ object CalcCodeGenerator {
inputTerm = inputTerm,
lazyInputUnboxingCode = true)
- new OneInputOperatorWrapper(genOperator)
+ new CodeGenOperatorFactory(genOperator)
}
private[flink] def generateProcessCode(
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/codegen/CorrelateCodeGenerator.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/codegen/CorrelateCodeGenerator.scala
index f9a7d7c..19c356c 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/codegen/CorrelateCodeGenerator.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/codegen/CorrelateCodeGenerator.scala
@@ -34,7 +34,7 @@ import org.apache.flink.table.generated.GeneratedCollector
import org.apache.flink.table.plan.nodes.logical.FlinkLogicalTableFunctionScan
import org.apache.flink.table.plan.schema.FlinkTableFunction
import org.apache.flink.table.plan.util.RelExplainUtil
-import org.apache.flink.table.runtime.OneInputOperatorWrapper
+import org.apache.flink.table.runtime.CodeGenOperatorFactory
import org.apache.flink.table.runtime.collector.TableFunctionCollector
import org.apache.flink.table.runtime.util.StreamRecordCollector
@@ -157,7 +157,7 @@ object CorrelateCodeGenerator {
ruleDescription: String,
functionClass: Class[T],
udtfCollector: GeneratedCollector[TableFunctionCollector[_]],
- retainHeader: Boolean = true): OneInputOperatorWrapper[BaseRow, BaseRow] = {
+ retainHeader: Boolean = true): CodeGenOperatorFactory[BaseRow] = {
ctx.references ++= collectorCtx.references
val exprGenerator = new ExprCodeGenerator(ctx, false)
.bindInput(inputType)
@@ -274,7 +274,7 @@ object CorrelateCodeGenerator {
"",
inputType,
config)
- new OneInputOperatorWrapper(genOperator)
+ new CodeGenOperatorFactory(genOperator)
}
private def generateProjectResultExpr(
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/codegen/ExpandCodeGenerator.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/codegen/ExpandCodeGenerator.scala
index c1727ad..7cd8ee5 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/codegen/ExpandCodeGenerator.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/codegen/ExpandCodeGenerator.scala
@@ -20,8 +20,8 @@ package org.apache.flink.table.codegen
import org.apache.flink.table.`type`.RowType
import org.apache.flink.table.api.TableConfig
-import org.apache.flink.table.dataformat.{BaseRow, BoxedWrapperRow, GenericRow}
-import org.apache.flink.table.runtime.OneInputOperatorWrapper
+import org.apache.flink.table.dataformat.{BaseRow, BoxedWrapperRow}
+import org.apache.flink.table.runtime.CodeGenOperatorFactory
import org.apache.calcite.rex.RexNode
@@ -37,7 +37,7 @@ object ExpandCodeGenerator {
config: TableConfig,
projects: java.util.List[java.util.List[RexNode]],
retainHeader: Boolean = false,
- opName: String): OneInputOperatorWrapper[BaseRow, BaseRow] = {
+ opName: String): CodeGenOperatorFactory[BaseRow] = {
val inputTerm = CodeGenUtils.DEFAULT_INPUT1_TERM
val exprGenerator = new ExprCodeGenerator(ctx, false)
@@ -71,7 +71,7 @@ object ExpandCodeGenerator {
inputTerm = inputTerm,
lazyInputUnboxingCode = false)
- new OneInputOperatorWrapper(genOperator)
+ new CodeGenOperatorFactory(genOperator)
}
}
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/codegen/LongHashJoinGenerator.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/codegen/LongHashJoinGenerator.scala
index e69bf11..7defe5d 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/codegen/LongHashJoinGenerator.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/codegen/LongHashJoinGenerator.scala
@@ -25,7 +25,7 @@ import org.apache.flink.table.codegen.CodeGenUtils.{BASE_ROW, BINARY_ROW, baseRo
import org.apache.flink.table.codegen.OperatorCodeGenerator.generateCollect
import org.apache.flink.table.dataformat.{BaseRow, JoinedRow}
import org.apache.flink.table.generated.{GeneratedJoinCondition, GeneratedProjection}
-import org.apache.flink.table.runtime.TwoInputOperatorWrapper
+import org.apache.flink.table.runtime.CodeGenOperatorFactory
import org.apache.flink.table.runtime.hashtable.{LongHashPartition, LongHybridHashTable}
import org.apache.flink.table.runtime.join.HashJoinType
import org.apache.flink.table.typeutils.BinaryRowSerializer
@@ -105,7 +105,7 @@ object LongHashJoinGenerator {
buildRowSize: Int,
buildRowCount: Long,
reverseJoinFunction: Boolean,
- condFunc: GeneratedJoinCondition): TwoInputOperatorWrapper[BaseRow, BaseRow, BaseRow] = {
+ condFunc: GeneratedJoinCondition): CodeGenOperatorFactory[BaseRow] = {
val buildSer = new BinaryRowSerializer(buildType.getArity)
val probeSer = new BinaryRowSerializer(probeType.getArity)
@@ -336,6 +336,6 @@ object LongHashJoinGenerator {
buildType,
probeType)
- new TwoInputOperatorWrapper[BaseRow, BaseRow, BaseRow](genOp)
+ new CodeGenOperatorFactory[BaseRow](genOp)
}
}
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/codegen/OperatorCodeGenerator.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/codegen/OperatorCodeGenerator.scala
index 76414ca..c406f97 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/codegen/OperatorCodeGenerator.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/codegen/OperatorCodeGenerator.scala
@@ -17,8 +17,10 @@
*/
package org.apache.flink.table.codegen
-import org.apache.flink.streaming.api.operators.{OneInputStreamOperator, StreamOperator, TwoInputStreamOperator}
+import org.apache.flink.streaming.api.graph.StreamConfig
+import org.apache.flink.streaming.api.operators.{OneInputStreamOperator, Output, StreamOperator, TwoInputStreamOperator}
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord
+import org.apache.flink.streaming.runtime.tasks.StreamTask
import org.apache.flink.table.`type`.InternalType
import org.apache.flink.table.api.TableConfig
import org.apache.flink.table.codegen.CodeGenUtils._
@@ -63,9 +65,14 @@ object OperatorCodeGenerator extends Logging {
private final Object[] references;
${ctx.reuseMemberCode()}
- public $operatorName(Object[] references) throws Exception {
+ public $operatorName(
+ Object[] references,
+ ${classOf[StreamTask[_, _]].getCanonicalName} task,
+ ${classOf[StreamConfig].getCanonicalName} config,
+ ${classOf[Output[_]].getCanonicalName} output) throws Exception {
this.references = references;
${ctx.reuseInitCode()}
+ this.setup(task, config, output);
}
@Override
@@ -144,9 +151,14 @@ object OperatorCodeGenerator extends Logging {
private final Object[] references;
${ctx.reuseMemberCode()}
- public $operatorName(Object[] references) throws Exception {
+ public $operatorName(
+ Object[] references,
+ ${classOf[StreamTask[_, _]].getCanonicalName} task,
+ ${classOf[StreamConfig].getCanonicalName} config,
+ ${classOf[Output[_]].getCanonicalName} output) throws Exception {
this.references = references;
${ctx.reuseInitCode()}
+ this.setup(task, config, output);
}
@Override
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/codegen/SinkCodeGenerator.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/codegen/SinkCodeGenerator.scala
index fcd1180..5ec3a38 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/codegen/SinkCodeGenerator.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/codegen/SinkCodeGenerator.scala
@@ -33,7 +33,7 @@ import org.apache.flink.table.codegen.CodeGenUtils.genToExternal
import org.apache.flink.table.codegen.OperatorCodeGenerator.generateCollect
import org.apache.flink.table.dataformat.util.BaseRowUtil
import org.apache.flink.table.dataformat.{BaseRow, GenericRow}
-import org.apache.flink.table.runtime.OneInputOperatorWrapper
+import org.apache.flink.table.runtime.CodeGenOperatorFactory
import org.apache.flink.table.sinks.{DataStreamTableSink, TableSink}
import org.apache.flink.table.typeutils.{BaseRowTypeInfo, TimeIndicatorTypeInfo}
import org.apache.flink.types.Row
@@ -63,7 +63,7 @@ object SinkCodeGenerator {
rowtimeField: Option[Int],
withChangeFlag: Boolean,
resultType: TypeInformation[_],
- sink: TableSink[_]): (OneInputOperatorWrapper[BaseRow, OUT], TypeInformation[OUT]) = {
+ sink: TableSink[_]): (CodeGenOperatorFactory[OUT], TypeInformation[OUT]) = {
val requestedTypeInfo = if (withChangeFlag) {
resultType match {
@@ -182,8 +182,7 @@ object SinkCodeGenerator {
endInputCode,
createInternalTypeFromTypeInfo(inputTypeInfo),
config)
- val operator = new OneInputOperatorWrapper[BaseRow, OUT](generated)
- (operator, outputTypeInfo.asInstanceOf[TypeInformation[OUT]])
+ (new CodeGenOperatorFactory[OUT](generated), outputTypeInfo.asInstanceOf[TypeInformation[OUT]])
}
private def checkRowConverterValid[OUT](
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/nodes/physical/batch/BatchExecHashAggregateBase.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/nodes/physical/batch/BatchExecHashAggregateBase.scala
index 1502299..9fbc1d7 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/nodes/physical/batch/BatchExecHashAggregateBase.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/nodes/physical/batch/BatchExecHashAggregateBase.scala
@@ -29,7 +29,7 @@ import org.apache.flink.table.plan.cost.FlinkCostFactory
import org.apache.flink.table.plan.nodes.exec.{BatchExecNode, ExecNode}
import org.apache.flink.table.plan.util.AggregateUtil.transformToBatchAggregateInfoList
import org.apache.flink.table.plan.util.FlinkRelMdUtil
-import org.apache.flink.table.runtime.OneInputOperatorWrapper
+import org.apache.flink.table.runtime.CodeGenOperatorFactory
import org.apache.calcite.plan.{RelOptCluster, RelOptCost, RelOptPlanner, RelTraitSet}
import org.apache.calcite.rel.RelNode
@@ -140,7 +140,7 @@ abstract class BatchExecHashAggregateBase(
ctx, relBuilder, aggInfos, inputType, outputType, grouping, auxGrouping, isMerge, isFinal
).genWithKeys(reservedManagedMem, maxManagedMem)
}
- val operator = new OneInputOperatorWrapper[BaseRow, BaseRow](generatedOperator)
+ val operator = new CodeGenOperatorFactory[BaseRow](generatedOperator)
new OneInputTransformation(
input,
getOperatorName,
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/nodes/physical/batch/BatchExecHashWindowAggregateBase.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/nodes/physical/batch/BatchExecHashWindowAggregateBase.scala
index 8c9fac4..dcd9026 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/nodes/physical/batch/BatchExecHashWindowAggregateBase.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/nodes/physical/batch/BatchExecHashWindowAggregateBase.scala
@@ -31,7 +31,7 @@ import org.apache.flink.table.plan.logical.LogicalWindow
import org.apache.flink.table.plan.nodes.exec.{BatchExecNode, ExecNode}
import org.apache.flink.table.plan.util.AggregateUtil.transformToBatchAggregateInfoList
import org.apache.flink.table.plan.util.FlinkRelMdUtil
-import org.apache.flink.table.runtime.OneInputOperatorWrapper
+import org.apache.flink.table.runtime.CodeGenOperatorFactory
import org.apache.flink.table.runtime.aggregate.BytesHashMap
import org.apache.calcite.plan.{RelOptCluster, RelOptCost, RelOptPlanner, RelTraitSet}
@@ -141,7 +141,7 @@ abstract class BatchExecHashWindowAggregateBase(
aggInfos, inputRowType, grouping, auxGrouping, enableAssignPane, isMerge, isFinal).gen(
inputType, outputType, groupBufferLimitSize, reservedManagedMem, maxManagedMem, 0,
windowSize, slideSize)
- val operator = new OneInputOperatorWrapper[BaseRow, BaseRow](generatedOperator)
+ val operator = new CodeGenOperatorFactory[BaseRow](generatedOperator)
new OneInputTransformation(
input,
getOperatorName,
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/nodes/physical/batch/BatchExecSortAggregateBase.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/nodes/physical/batch/BatchExecSortAggregateBase.scala
index 679ea11..bd7d535 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/nodes/physical/batch/BatchExecSortAggregateBase.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/nodes/physical/batch/BatchExecSortAggregateBase.scala
@@ -27,7 +27,7 @@ import org.apache.flink.table.functions.UserDefinedFunction
import org.apache.flink.table.plan.cost.{FlinkCost, FlinkCostFactory}
import org.apache.flink.table.plan.nodes.exec.{BatchExecNode, ExecNode}
import org.apache.flink.table.plan.util.AggregateUtil.transformToBatchAggregateInfoList
-import org.apache.flink.table.runtime.OneInputOperatorWrapper
+import org.apache.flink.table.runtime.CodeGenOperatorFactory
import org.apache.calcite.plan.{RelOptCluster, RelOptCost, RelOptPlanner, RelTraitSet}
import org.apache.calcite.rel.RelNode
@@ -119,7 +119,7 @@ abstract class BatchExecSortAggregateBase(
SortAggCodeGenerator.genWithKeys(
ctx, relBuilder, aggInfos, inputType, outputType, grouping, auxGrouping, isMerge, isFinal)
}
- val operator = new OneInputOperatorWrapper[BaseRow, BaseRow](generatedOperator)
+ val operator = new CodeGenOperatorFactory[BaseRow](generatedOperator)
new OneInputTransformation(
input,
getOperatorName,
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/nodes/physical/batch/BatchExecSortWindowAggregateBase.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/nodes/physical/batch/BatchExecSortWindowAggregateBase.scala
index fb7b562..88c986d 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/nodes/physical/batch/BatchExecSortWindowAggregateBase.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/nodes/physical/batch/BatchExecSortWindowAggregateBase.scala
@@ -30,7 +30,7 @@ import org.apache.flink.table.plan.cost.{FlinkCost, FlinkCostFactory}
import org.apache.flink.table.plan.logical.LogicalWindow
import org.apache.flink.table.plan.nodes.exec.{BatchExecNode, ExecNode}
import org.apache.flink.table.plan.util.AggregateUtil.transformToBatchAggregateInfoList
-import org.apache.flink.table.runtime.OneInputOperatorWrapper
+import org.apache.flink.table.runtime.CodeGenOperatorFactory
import org.apache.calcite.plan.{RelOptCluster, RelOptCost, RelOptPlanner, RelTraitSet}
import org.apache.calcite.rel.RelNode
@@ -133,7 +133,7 @@ abstract class BatchExecSortWindowAggregateBase(
} else {
generator.genWithKeys()
}
- val operator = new OneInputOperatorWrapper[BaseRow, BaseRow](generatedOperator)
+ val operator = new CodeGenOperatorFactory[BaseRow](generatedOperator)
new OneInputTransformation(
input,
getOperatorName,
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/util/ScanUtil.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/util/ScanUtil.scala
index 1a59022..f38604b 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/util/ScanUtil.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/util/ScanUtil.scala
@@ -28,7 +28,7 @@ import org.apache.flink.table.codegen.CodeGenUtils.{DEFAULT_INPUT1_TERM, GENERIC
import org.apache.flink.table.codegen.OperatorCodeGenerator.generateCollect
import org.apache.flink.table.codegen.{CodeGenUtils, CodeGeneratorContext, ExprCodeGenerator, OperatorCodeGenerator}
import org.apache.flink.table.dataformat.{BaseRow, GenericRow}
-import org.apache.flink.table.runtime.OneInputOperatorWrapper
+import org.apache.flink.table.runtime.CodeGenOperatorFactory
import org.apache.flink.table.typeutils.TimeIndicatorTypeInfo
import org.apache.calcite.rel.`type`.RelDataType
@@ -112,7 +112,7 @@ object ScanUtil {
config,
converter = inputTermConverter)
- val substituteStreamOperator = new OneInputOperatorWrapper[Any, BaseRow](generatedOperator)
+ val substituteStreamOperator = new CodeGenOperatorFactory[BaseRow](generatedOperator)
new OneInputTransformation(
input,
diff --git a/flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/codegen/LongHashJoinGeneratorTest.java b/flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/codegen/LongHashJoinGeneratorTest.java
index 8b87f6d..84ff59c 100644
--- a/flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/codegen/LongHashJoinGeneratorTest.java
+++ b/flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/codegen/LongHashJoinGeneratorTest.java
@@ -18,7 +18,6 @@
package org.apache.flink.table.codegen;
-import org.apache.flink.streaming.api.operators.StreamOperator;
import org.apache.flink.table.api.TableConfig;
import org.apache.flink.table.dataformat.BaseRow;
import org.apache.flink.table.generated.GeneratedJoinCondition;
@@ -37,7 +36,7 @@ import org.junit.Test;
public class LongHashJoinGeneratorTest extends Int2HashJoinOperatorTest {
@Override
- public StreamOperator newOperator(long memorySize, HashJoinType type, boolean reverseJoinFunction) {
+ public Object newOperator(long memorySize, HashJoinType type, boolean reverseJoinFunction) {
RowType keyType = new RowType(InternalTypes.INT);
Assert.assertTrue(LongHashJoinGenerator.support(type, keyType, new boolean[] {true}));
return LongHashJoinGenerator.gen(
diff --git a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/codegen/agg/batch/AggWithoutKeysTest.scala b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/codegen/agg/batch/AggWithoutKeysTest.scala
index 2069af9..f31a5c2 100644
--- a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/codegen/agg/batch/AggWithoutKeysTest.scala
+++ b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/codegen/agg/batch/AggWithoutKeysTest.scala
@@ -18,11 +18,10 @@
package org.apache.flink.table.codegen.agg.batch
-import org.apache.flink.streaming.api.operators.OneInputStreamOperator
import org.apache.flink.table.`type`.{InternalType, InternalTypes, RowType}
import org.apache.flink.table.`type`.TypeConverters.createInternalTypeFromTypeInfo
import org.apache.flink.table.dataformat.BaseRow
-import org.apache.flink.table.runtime.OneInputOperatorWrapper
+import org.apache.flink.table.runtime.CodeGenOperatorFactory
import org.junit.Test
@@ -90,7 +89,7 @@ class AggWithoutKeysTest extends BatchAggTestBase {
}
private def getOperatorWithoutKey(isMerge: Boolean, isFinal: Boolean)
- : (OneInputStreamOperator[BaseRow, BaseRow], RowType, RowType) = {
+ : (CodeGenOperatorFactory[BaseRow], RowType, RowType) = {
val (iType, oType) = if (isMerge && isFinal) {
(localOutputType, globalOutputType)
} else if (!isMerge && isFinal) {
@@ -100,6 +99,6 @@ class AggWithoutKeysTest extends BatchAggTestBase {
}
val genOp = AggWithoutKeysCodeGenerator.genWithoutKeys(
ctx, relBuilder, aggInfoList, iType, oType, isMerge, isFinal, "Without")
- (new OneInputOperatorWrapper[BaseRow, BaseRow](genOp), iType, oType)
+ (new CodeGenOperatorFactory[BaseRow](genOp), iType, oType)
}
}
diff --git a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/codegen/agg/batch/BatchAggTestBase.scala b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/codegen/agg/batch/BatchAggTestBase.scala
index 976bb6a..b286396 100644
--- a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/codegen/agg/batch/BatchAggTestBase.scala
+++ b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/codegen/agg/batch/BatchAggTestBase.scala
@@ -20,13 +20,12 @@ package org.apache.flink.table.codegen.agg.batch
import org.apache.flink.runtime.execution.Environment
import org.apache.flink.runtime.jobgraph.OperatorID
-import org.apache.flink.streaming.api.operators.OneInputStreamOperator
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord
-import org.apache.flink.streaming.runtime.tasks.{OneInputStreamTask, OneInputStreamTaskTestHarness, OperatorChain}
+import org.apache.flink.streaming.runtime.tasks.{OneInputStreamTask, OneInputStreamTaskTestHarness}
import org.apache.flink.table.`type`.{InternalType, InternalTypes, RowType}
import org.apache.flink.table.codegen.agg.AggTestBase
import org.apache.flink.table.dataformat.{BaseRow, BinaryString, GenericRow}
-import org.apache.flink.table.runtime.OneInputOperatorWrapper
+import org.apache.flink.table.runtime.CodeGenOperatorFactory
import org.apache.flink.table.util.BaseRowTestUtil
import org.junit.Assert
@@ -63,7 +62,7 @@ abstract class BatchAggTestBase extends AggTestBase {
}
def testOperator(
- args: (OneInputStreamOperator[BaseRow, BaseRow], RowType, RowType),
+ args: (CodeGenOperatorFactory[BaseRow], RowType, RowType),
input: Array[BaseRow], expectedOutput: Array[GenericRow]): Unit = {
val testHarness = new OneInputStreamTaskTestHarness[BaseRow, BaseRow](
new function.Function[Environment, OneInputStreamTask[BaseRow, BaseRow]] {
@@ -73,7 +72,7 @@ abstract class BatchAggTestBase extends AggTestBase {
testHarness.setupOutputForSingletonOperatorChain()
val streamConfig = testHarness.getStreamConfig
- streamConfig.setStreamOperator(args._1)
+ streamConfig.setStreamOperatorFactory(args._1)
streamConfig.setOperatorID(new OperatorID)
testHarness.invoke()
@@ -85,11 +84,6 @@ abstract class BatchAggTestBase extends AggTestBase {
testHarness.waitForInputProcessing()
- val op = testHarness.getTask.getStreamStatusMaintainer.asInstanceOf[OperatorChain[_, _]]
- .getHeadOperator.asInstanceOf[OneInputOperatorWrapper[_, _]].getOperator
- // TODO close will invoke endInput
- // op.getClass.getMethod("endInput").invoke(op)
-
testHarness.endInput()
testHarness.waitForTaskCompletion()
diff --git a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/codegen/agg/batch/HashAggCodeGeneratorTest.scala b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/codegen/agg/batch/HashAggCodeGeneratorTest.scala
index bb1e693..b591982 100644
--- a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/codegen/agg/batch/HashAggCodeGeneratorTest.scala
+++ b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/codegen/agg/batch/HashAggCodeGeneratorTest.scala
@@ -19,12 +19,11 @@
package org.apache.flink.table.codegen.agg.batch
import org.apache.flink.api.common.typeinfo.Types
-import org.apache.flink.streaming.api.operators.OneInputStreamOperator
import org.apache.flink.table.`type`.{InternalType, InternalTypes, RowType}
import org.apache.flink.table.dataformat.BaseRow
import org.apache.flink.table.functions.aggfunctions.AvgAggFunction.IntegralAvgAggFunction
import org.apache.flink.table.plan.util.{AggregateInfo, AggregateInfoList}
-import org.apache.flink.table.runtime.OneInputOperatorWrapper
+import org.apache.flink.table.runtime.CodeGenOperatorFactory
import org.apache.calcite.rel.core.AggregateCall
import org.junit.Test
@@ -113,7 +112,7 @@ class HashAggCodeGeneratorTest extends BatchAggTestBase {
}
private def getOperatorWithKey(isMerge: Boolean, isFinal: Boolean)
- : (OneInputStreamOperator[BaseRow, BaseRow], RowType, RowType) = {
+ : (CodeGenOperatorFactory[BaseRow], RowType, RowType) = {
val (iType, oType) = if (isMerge && isFinal) {
(localOutputType, globalOutputType)
} else if (!isMerge && isFinal) {
@@ -125,7 +124,7 @@ class HashAggCodeGeneratorTest extends BatchAggTestBase {
val generator = new HashAggCodeGenerator(
ctx, relBuilder, aggInfoList, iType, oType, Array(0), auxGrouping, isMerge, isFinal)
val genOp = generator.genWithKeys(100 * 32 * 1024, 0)
- (new OneInputOperatorWrapper[BaseRow, BaseRow](genOp), iType, oType)
+ (new CodeGenOperatorFactory[BaseRow](genOp), iType, oType)
}
diff --git a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/codegen/agg/batch/SortAggCodeGeneratorTest.scala b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/codegen/agg/batch/SortAggCodeGeneratorTest.scala
index ecf1d7c..45e3f9b 100644
--- a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/codegen/agg/batch/SortAggCodeGeneratorTest.scala
+++ b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/codegen/agg/batch/SortAggCodeGeneratorTest.scala
@@ -18,11 +18,10 @@
package org.apache.flink.table.codegen.agg.batch
-import org.apache.flink.streaming.api.operators.OneInputStreamOperator
import org.apache.flink.table.`type`.TypeConverters.createInternalTypeFromTypeInfo
import org.apache.flink.table.`type`.{InternalType, InternalTypes, RowType}
import org.apache.flink.table.dataformat.BaseRow
-import org.apache.flink.table.runtime.OneInputOperatorWrapper
+import org.apache.flink.table.runtime.CodeGenOperatorFactory
import org.junit.Test
@@ -93,7 +92,7 @@ class SortAggCodeGeneratorTest extends BatchAggTestBase {
}
private def getOperatorWithKey(isMerge: Boolean, isFinal: Boolean)
- : (OneInputStreamOperator[BaseRow, BaseRow], RowType, RowType) = {
+ : (CodeGenOperatorFactory[BaseRow], RowType, RowType) = {
val localOutputType = new RowType(
Array[InternalType](
InternalTypes.STRING, InternalTypes.STRING,
@@ -116,6 +115,6 @@ class SortAggCodeGeneratorTest extends BatchAggTestBase {
val auxGrouping = if (isMerge) Array(1) else Array(4)
val genOp = SortAggCodeGenerator.genWithKeys(
ctx, relBuilder, aggInfoList, iType, oType, Array(0), auxGrouping, isMerge, isFinal)
- (new OneInputOperatorWrapper[BaseRow, BaseRow](genOp), iType, oType)
+ (new CodeGenOperatorFactory[BaseRow](genOp), iType, oType)
}
}
diff --git a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/runtime/batch/sql/join/JoinITCase.scala b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/runtime/batch/sql/join/JoinITCase.scala
index 4d226af..908c0fd 100644
--- a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/runtime/batch/sql/join/JoinITCase.scala
+++ b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/runtime/batch/sql/join/JoinITCase.scala
@@ -23,9 +23,8 @@ import org.apache.flink.api.common.typeinfo.BasicArrayTypeInfo.STRING_ARRAY_TYPE
import org.apache.flink.api.common.typeinfo.BasicTypeInfo.{INT_TYPE_INFO, LONG_TYPE_INFO, STRING_TYPE_INFO}
import org.apache.flink.api.common.typeutils.TypeComparator
import org.apache.flink.api.java.typeutils.{GenericTypeInfo, ObjectArrayTypeInfo, RowTypeInfo}
-import org.apache.flink.streaming.api.operators.SimpleOperatorFactory
import org.apache.flink.table.api.{TableConfigOptions, Types}
-import org.apache.flink.table.runtime.TwoInputOperatorWrapper
+import org.apache.flink.table.runtime.CodeGenOperatorFactory
import org.apache.flink.table.runtime.batch.sql.join.JoinType.{BroadcastHashJoin, HashJoin, JoinType, NestedLoopJoin, SortMergeJoin}
import org.apache.flink.table.runtime.utils.BatchTestBase
import org.apache.flink.table.runtime.utils.BatchTestBase.row
@@ -78,10 +77,11 @@ class JoinITCase() extends BatchTestBase {
var haveTwoOp = false
env.getStreamGraph.getAllOperatorFactory.foreach(o =>
- o.f1.asInstanceOf[SimpleOperatorFactory[_]].getOperator match {
- case two: TwoInputOperatorWrapper[_, _, _] =>
- Assert.assertTrue(two.getGeneratedClass.getCode.contains("LongHashJoinOperator"))
- haveTwoOp = true
+ o.f1 match {
+ case factory: CodeGenOperatorFactory[_] =>
+ if (factory.getGeneratedClass.getCode.contains("LongHashJoinOperator")) {
+ haveTwoOp = true
+ }
case _ =>
}
)
diff --git a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/generated/GeneratedClass.java b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/generated/GeneratedClass.java
index 736fbcd..903943a 100644
--- a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/generated/GeneratedClass.java
+++ b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/generated/GeneratedClass.java
@@ -59,6 +59,16 @@ public abstract class GeneratedClass<T> implements Serializable {
}
}
+ @SuppressWarnings("unchecked")
+ public T newInstance(ClassLoader classLoader, Object... args) {
+ try {
+ return (T) compile(classLoader).getConstructors()[0].newInstance(args);
+ } catch (Exception e) {
+ throw new RuntimeException(
+ "Could not instantiate generated class '" + className + "'", e);
+ }
+ }
+
private Class<?> compile(ClassLoader classLoader) {
if (compiledClass == null) {
// cache the compiled class
diff --git a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/CodeGenOperatorFactory.java b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/CodeGenOperatorFactory.java
new file mode 100644
index 0000000..7f1d6f8
--- /dev/null
+++ b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/CodeGenOperatorFactory.java
@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime;
+
+import org.apache.flink.streaming.api.graph.StreamConfig;
+import org.apache.flink.streaming.api.operators.ChainingStrategy;
+import org.apache.flink.streaming.api.operators.Output;
+import org.apache.flink.streaming.api.operators.StreamOperator;
+import org.apache.flink.streaming.api.operators.StreamOperatorFactory;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.runtime.tasks.StreamTask;
+import org.apache.flink.table.generated.GeneratedClass;
+
+/**
+ * Stream operator factory for code gen operator.
+ */
+public class CodeGenOperatorFactory<OUT> implements StreamOperatorFactory<OUT> {
+
+ private final GeneratedClass<? extends StreamOperator<OUT>> generatedClass;
+ private ChainingStrategy strategy = ChainingStrategy.ALWAYS;
+
+ public CodeGenOperatorFactory(GeneratedClass<? extends StreamOperator<OUT>> generatedClass) {
+ this.generatedClass = generatedClass;
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public <T extends StreamOperator<OUT>> T createStreamOperator(StreamTask<?, ?> containingTask,
+ StreamConfig config, Output<StreamRecord<OUT>> output) {
+ return (T) generatedClass.newInstance(containingTask.getUserCodeClassLoader(),
+ generatedClass.getReferences(), containingTask, config, output);
+ }
+
+ @Override
+ public void setChainingStrategy(ChainingStrategy strategy) {
+ this.strategy = strategy;
+ }
+
+ @Override
+ public ChainingStrategy getChainingStrategy() {
+ return strategy;
+ }
+
+ public GeneratedClass<? extends StreamOperator<OUT>> getGeneratedClass() {
+ return generatedClass;
+ }
+}
diff --git a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/OneInputOperatorWrapper.java b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/OneInputOperatorWrapper.java
deleted file mode 100644
index 15660fa..0000000
--- a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/OneInputOperatorWrapper.java
+++ /dev/null
@@ -1,155 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.table.runtime;
-
-import org.apache.flink.annotation.VisibleForTesting;
-import org.apache.flink.metrics.MetricGroup;
-import org.apache.flink.runtime.checkpoint.CheckpointOptions;
-import org.apache.flink.runtime.jobgraph.OperatorID;
-import org.apache.flink.runtime.state.CheckpointStreamFactory;
-import org.apache.flink.streaming.api.graph.StreamConfig;
-import org.apache.flink.streaming.api.operators.ChainingStrategy;
-import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
-import org.apache.flink.streaming.api.operators.OperatorSnapshotFutures;
-import org.apache.flink.streaming.api.operators.Output;
-import org.apache.flink.streaming.api.operators.SetupableStreamOperator;
-import org.apache.flink.streaming.api.watermark.Watermark;
-import org.apache.flink.streaming.runtime.streamrecord.LatencyMarker;
-import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
-import org.apache.flink.streaming.runtime.tasks.StreamTask;
-import org.apache.flink.table.generated.GeneratedClass;
-
-/**
- * Wrapper for code gen operator.
- * TODO Remove it after FLINK-11974.
- */
-public class OneInputOperatorWrapper<IN, OUT>
- implements OneInputStreamOperator<IN, OUT>, SetupableStreamOperator<OUT> {
-
- private final GeneratedClass<OneInputStreamOperator<IN, OUT>> generatedClass;
-
- private transient OneInputStreamOperator<IN, OUT> operator;
-
- public OneInputOperatorWrapper(GeneratedClass<OneInputStreamOperator<IN, OUT>> generatedClass) {
- this.generatedClass = generatedClass;
- }
-
- @Override
- public void setup(StreamTask<?, ?> containingTask, StreamConfig config,
- Output<StreamRecord<OUT>> output) {
- operator = generatedClass.newInstance(containingTask.getUserCodeClassLoader());
- ((SetupableStreamOperator) operator).setup(containingTask, config, output);
- }
-
- @VisibleForTesting
- public OneInputStreamOperator<IN, OUT> getOperator() {
- return operator;
- }
-
- @Override
- public void open() throws Exception {
- operator.open();
- }
-
- @Override
- public void close() throws Exception {
- operator.close();
- }
-
- @Override
- public void dispose() throws Exception {
- operator.dispose();
- }
-
- @Override
- public void prepareSnapshotPreBarrier(long checkpointId) throws Exception {
- operator.prepareSnapshotPreBarrier(checkpointId);
- }
-
- @Override
- public OperatorSnapshotFutures snapshotState(long checkpointId, long timestamp,
- CheckpointOptions checkpointOptions,
- CheckpointStreamFactory storageLocation) throws Exception {
- return operator.snapshotState(checkpointId, timestamp, checkpointOptions, storageLocation);
- }
-
- @Override
- public void initializeState() throws Exception {
- operator.initializeState();
- }
-
- @Override
- public void setKeyContextElement1(StreamRecord<?> record) throws Exception {
- operator.setKeyContextElement1(record);
- }
-
- @Override
- public void setKeyContextElement2(StreamRecord<?> record) throws Exception {
- operator.setKeyContextElement2(record);
- }
-
- @Override
- public ChainingStrategy getChainingStrategy() {
- return ChainingStrategy.ALWAYS;
- }
-
- @Override
- public void setChainingStrategy(ChainingStrategy strategy) {
- operator.setChainingStrategy(strategy);
- }
-
- @Override
- public MetricGroup getMetricGroup() {
- return operator.getMetricGroup();
- }
-
- @Override
- public OperatorID getOperatorID() {
- return operator.getOperatorID();
- }
-
- @Override
- public void processElement(StreamRecord<IN> element) throws Exception {
- operator.processElement(element);
- }
-
- @Override
- public void processWatermark(Watermark mark) throws Exception {
- operator.processWatermark(mark);
- }
-
- @Override
- public void processLatencyMarker(LatencyMarker latencyMarker) throws Exception {
- operator.processLatencyMarker(latencyMarker);
- }
-
- @Override
- public void notifyCheckpointComplete(long checkpointId) throws Exception {
- operator.notifyCheckpointComplete(checkpointId);
- }
-
- @Override
- public void setCurrentKey(Object key) {
- operator.setCurrentKey(key);
- }
-
- @Override
- public Object getCurrentKey() {
- return operator.getCurrentKey();
- }
-}
diff --git a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/TwoInputOperatorWrapper.java b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/TwoInputOperatorWrapper.java
deleted file mode 100644
index 14ceac3..0000000
--- a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/TwoInputOperatorWrapper.java
+++ /dev/null
@@ -1,175 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.table.runtime;
-
-import org.apache.flink.annotation.VisibleForTesting;
-import org.apache.flink.metrics.MetricGroup;
-import org.apache.flink.runtime.checkpoint.CheckpointOptions;
-import org.apache.flink.runtime.jobgraph.OperatorID;
-import org.apache.flink.runtime.state.CheckpointStreamFactory;
-import org.apache.flink.streaming.api.graph.StreamConfig;
-import org.apache.flink.streaming.api.operators.ChainingStrategy;
-import org.apache.flink.streaming.api.operators.OperatorSnapshotFutures;
-import org.apache.flink.streaming.api.operators.Output;
-import org.apache.flink.streaming.api.operators.SetupableStreamOperator;
-import org.apache.flink.streaming.api.operators.TwoInputStreamOperator;
-import org.apache.flink.streaming.api.watermark.Watermark;
-import org.apache.flink.streaming.runtime.streamrecord.LatencyMarker;
-import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
-import org.apache.flink.streaming.runtime.tasks.StreamTask;
-import org.apache.flink.table.generated.GeneratedClass;
-
-/**
- * Wrapper for code gen operator.
- * TODO Remove it after FLINK-11974.
- */
-public class TwoInputOperatorWrapper<IN1, IN2, OUT>
- implements TwoInputStreamOperator<IN1, IN2, OUT>, SetupableStreamOperator<OUT> {
-
- private final GeneratedClass<TwoInputStreamOperator<IN1, IN2, OUT>> generatedClass;
-
- private transient TwoInputStreamOperator<IN1, IN2, OUT> operator;
-
- public TwoInputOperatorWrapper(GeneratedClass<TwoInputStreamOperator<IN1, IN2, OUT>> generatedClass) {
- this.generatedClass = generatedClass;
- }
-
- @Override
- public void setup(StreamTask<?, ?> containingTask, StreamConfig config,
- Output<StreamRecord<OUT>> output) {
- operator = generatedClass.newInstance(containingTask.getUserCodeClassLoader());
- ((SetupableStreamOperator) operator).setup(containingTask, config, output);
- }
-
- @VisibleForTesting
- public TwoInputStreamOperator<IN1, IN2, OUT> getOperator() {
- return operator;
- }
-
- @VisibleForTesting
- public GeneratedClass<TwoInputStreamOperator<IN1, IN2, OUT>> getGeneratedClass() {
- return generatedClass;
- }
-
- @Override
- public void open() throws Exception {
- operator.open();
- }
-
- @Override
- public void close() throws Exception {
- operator.close();
- }
-
- @Override
- public void dispose() throws Exception {
- operator.dispose();
- }
-
- @Override
- public void prepareSnapshotPreBarrier(long checkpointId) throws Exception {
- operator.prepareSnapshotPreBarrier(checkpointId);
- }
-
- @Override
- public OperatorSnapshotFutures snapshotState(long checkpointId, long timestamp,
- CheckpointOptions checkpointOptions,
- CheckpointStreamFactory storageLocation) throws Exception {
- return operator.snapshotState(checkpointId, timestamp, checkpointOptions, storageLocation);
- }
-
- @Override
- public void initializeState() throws Exception {
- operator.initializeState();
- }
-
- @Override
- public void setKeyContextElement1(StreamRecord<?> record) throws Exception {
- operator.setKeyContextElement1(record);
- }
-
- @Override
- public void setKeyContextElement2(StreamRecord<?> record) throws Exception {
- operator.setKeyContextElement2(record);
- }
-
- @Override
- public ChainingStrategy getChainingStrategy() {
- return ChainingStrategy.ALWAYS;
- }
-
- @Override
- public void setChainingStrategy(ChainingStrategy strategy) {
- operator.setChainingStrategy(strategy);
- }
-
- @Override
- public MetricGroup getMetricGroup() {
- return operator.getMetricGroup();
- }
-
- @Override
- public OperatorID getOperatorID() {
- return operator.getOperatorID();
- }
-
- @Override
- public void processElement1(StreamRecord<IN1> element) throws Exception {
- operator.processElement1(element);
- }
-
- @Override
- public void processElement2(StreamRecord<IN2> element) throws Exception {
- operator.processElement2(element);
- }
-
- @Override
- public void processWatermark1(Watermark mark) throws Exception {
- operator.processWatermark1(mark);
- }
-
- @Override
- public void processWatermark2(Watermark mark) throws Exception {
- operator.processWatermark2(mark);
- }
-
- @Override
- public void processLatencyMarker1(LatencyMarker latencyMarker) throws Exception {
- operator.processLatencyMarker1(latencyMarker);
- }
-
- @Override
- public void processLatencyMarker2(LatencyMarker latencyMarker) throws Exception {
- operator.processLatencyMarker2(latencyMarker);
- }
-
- @Override
- public void notifyCheckpointComplete(long checkpointId) throws Exception {
- operator.notifyCheckpointComplete(checkpointId);
- }
-
- @Override
- public void setCurrentKey(Object key) {
- operator.setCurrentKey(key);
- }
-
- @Override
- public Object getCurrentKey() {
- return operator.getCurrentKey();
- }
-}
diff --git a/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/join/Int2HashJoinOperatorTest.java b/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/join/Int2HashJoinOperatorTest.java
index 604ebe4..fbbea2a 100644
--- a/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/join/Int2HashJoinOperatorTest.java
+++ b/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/join/Int2HashJoinOperatorTest.java
@@ -21,6 +21,7 @@ package org.apache.flink.table.runtime.join;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.runtime.jobgraph.OperatorID;
import org.apache.flink.streaming.api.operators.StreamOperator;
+import org.apache.flink.streaming.api.operators.StreamOperatorFactory;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.streaming.runtime.tasks.OperatorChain;
import org.apache.flink.streaming.runtime.tasks.TwoInputStreamTask;
@@ -33,7 +34,6 @@ import org.apache.flink.table.generated.GeneratedJoinCondition;
import org.apache.flink.table.generated.GeneratedProjection;
import org.apache.flink.table.generated.JoinCondition;
import org.apache.flink.table.generated.Projection;
-import org.apache.flink.table.runtime.TwoInputOperatorWrapper;
import org.apache.flink.table.runtime.util.UniformBinaryRowGenerator;
import org.apache.flink.table.type.InternalTypes;
import org.apache.flink.table.type.RowType;
@@ -181,7 +181,7 @@ public class Int2HashJoinOperatorTest implements Serializable {
MutableObjectIterator<BinaryRow> probeInput = new UniformBinaryRowGenerator(numKeys2, probeValsPerKey, true);
HashJoinType type = HashJoinType.SEMI;
- StreamOperator operator = newOperator(33 * 32 * 1024, type, false);
+ Object operator = newOperator(33 * 32 * 1024, type, false);
joinAndAssert(operator, buildInput, probeInput, 90, 9, 45, true);
}
@@ -196,7 +196,7 @@ public class Int2HashJoinOperatorTest implements Serializable {
MutableObjectIterator<BinaryRow> probeInput = new UniformBinaryRowGenerator(numKeys2, probeValsPerKey, true);
HashJoinType type = HashJoinType.ANTI;
- StreamOperator operator = newOperator(33 * 32 * 1024, type, false);
+ Object operator = newOperator(33 * 32 * 1024, type, false);
joinAndAssert(operator, buildInput, probeInput, 10, 1, 45, true);
}
@@ -211,7 +211,7 @@ public class Int2HashJoinOperatorTest implements Serializable {
MutableObjectIterator<BinaryRow> probeInput = new UniformBinaryRowGenerator(numKeys2, probeValsPerKey, true);
HashJoinType type = HashJoinType.BUILD_LEFT_SEMI;
- StreamOperator operator = newOperator(33 * 32 * 1024, type, false);
+ Object operator = newOperator(33 * 32 * 1024, type, false);
joinAndAssert(operator, buildInput, probeInput, 90, 9, 45, true);
}
@@ -226,7 +226,7 @@ public class Int2HashJoinOperatorTest implements Serializable {
MutableObjectIterator<BinaryRow> probeInput = new UniformBinaryRowGenerator(numKeys2, probeValsPerKey, true);
HashJoinType type = HashJoinType.BUILD_LEFT_ANTI;
- StreamOperator operator = newOperator(33 * 32 * 1024, type, false);
+ Object operator = newOperator(33 * 32 * 1024, type, false);
joinAndAssert(operator, buildInput, probeInput, 10, 1, 45, true);
}
@@ -236,12 +236,12 @@ public class Int2HashJoinOperatorTest implements Serializable {
boolean leftOut, boolean rightOut, boolean buildLeft,
int expectOutSize, int expectOutKeySize, int expectOutVal) throws Exception {
HashJoinType type = HashJoinType.of(buildLeft, leftOut, rightOut);
- StreamOperator operator = newOperator(33 * 32 * 1024, type, !buildLeft);
+ Object operator = newOperator(33 * 32 * 1024, type, !buildLeft);
joinAndAssert(operator, buildInput, probeInput, expectOutSize, expectOutKeySize, expectOutVal, false);
}
static void joinAndAssert(
- StreamOperator operator,
+ Object operator,
MutableObjectIterator<BinaryRow> input1,
MutableObjectIterator<BinaryRow> input2,
int expectOutSize,
@@ -252,7 +252,7 @@ public class Int2HashJoinOperatorTest implements Serializable {
}
static void joinAndAssert(
- StreamOperator operator,
+ Object operator,
MutableObjectIterator<BinaryRow> input1,
MutableObjectIterator<BinaryRow> input2,
int expectOutSize,
@@ -269,7 +269,11 @@ public class Int2HashJoinOperatorTest implements Serializable {
testHarness.memorySize = 36 * 1024 * 1024;
testHarness.getExecutionConfig().enableObjectReuse();
testHarness.setupOutputForSingletonOperatorChain();
- testHarness.getStreamConfig().setStreamOperator(operator);
+ if (operator instanceof StreamOperator) {
+ testHarness.getStreamConfig().setStreamOperator((StreamOperator<?>) operator);
+ } else {
+ testHarness.getStreamConfig().setStreamOperatorFactory((StreamOperatorFactory<?>) operator);
+ }
testHarness.getStreamConfig().setOperatorID(new OperatorID());
testHarness.invoke();
@@ -391,21 +395,15 @@ public class Int2HashJoinOperatorTest implements Serializable {
static void endInput1(TwoInputStreamTaskTestHarness harness) throws Exception {
StreamOperator op = ((OperatorChain) harness.getTask().getStreamStatusMaintainer()).getHeadOperator();
- if (op instanceof TwoInputOperatorWrapper) {
- op = ((TwoInputOperatorWrapper) op).getOperator();
- }
op.getClass().getMethod("endInput1").invoke(op);
}
static void endInput2(TwoInputStreamTaskTestHarness harness) throws Exception {
StreamOperator op = ((OperatorChain) harness.getTask().getStreamStatusMaintainer()).getHeadOperator();
- if (op instanceof TwoInputOperatorWrapper) {
- op = ((TwoInputOperatorWrapper) op).getOperator();
- }
op.getClass().getMethod("endInput2").invoke(op);
}
- public StreamOperator newOperator(long memorySize, HashJoinType type, boolean reverseJoinFunction) {
+ public Object newOperator(long memorySize, HashJoinType type, boolean reverseJoinFunction) {
return HashJoinOperator.newHashJoinOperator(
memorySize, memorySize, 0, type,
new GeneratedJoinCondition("", "", new Object[0]) {