You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by go...@apache.org on 2021/01/19 14:19:52 UTC
[flink] 04/05: [FLINK-20949][table-planner-blink] Introduce
BatchPhysicalLegacySink,
and make BatchExecLegacySink only extended from ExecNode
This is an automated email from the ASF dual-hosted git repository.
godfrey pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 45bf1b228cbdfc82f8b74b15a1c17db82ad513fb
Author: godfreyhe <go...@163.com>
AuthorDate: Thu Jan 14 11:28:32 2021 +0800
[FLINK-20949][table-planner-blink] Introduce BatchPhysicalLegacySink, and make BatchExecLegacySink only extended from ExecNode
This closes #14637
---
.../plan/nodes/exec/batch/BatchExecLegacySink.java | 81 ++++++++++
.../plan/nodes/exec/utils/ExecNodePlanDumper.java | 4 +-
.../nodes/physical/batch/BatchExecLegacySink.scala | 171 ---------------------
.../physical/batch/BatchPhysicalLegacySink.scala | 67 ++++++++
.../BatchCommonSubGraphBasedOptimizer.scala | 4 +-
.../planner/plan/rules/FlinkBatchRuleSets.scala | 2 +-
...ule.scala => BatchPhysicalLegacySinkRule.scala} | 14 +-
7 files changed, 158 insertions(+), 185 deletions(-)
diff --git a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/batch/BatchExecLegacySink.java b/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/batch/BatchExecLegacySink.java
new file mode 100644
index 0000000..e249378
--- /dev/null
+++ b/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/batch/BatchExecLegacySink.java
@@ -0,0 +1,81 @@
+/*
+ * 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.planner.plan.nodes.exec.batch;
+
+import org.apache.flink.table.api.TableException;
+import org.apache.flink.table.planner.plan.nodes.exec.ExecEdge;
+import org.apache.flink.table.planner.plan.nodes.exec.ExecNode;
+import org.apache.flink.table.planner.plan.nodes.exec.common.CommonExecLegacySink;
+import org.apache.flink.table.runtime.types.ClassLogicalTypeConverter;
+import org.apache.flink.table.sinks.TableSink;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.RowType;
+
+import javax.annotation.Nullable;
+
+import java.lang.reflect.Modifier;
+
+/**
+ * Batch {@link ExecNode} to to write data into an external sink defined by a {@link TableSink}.
+ *
+ * @param <T> The return type of the {@link TableSink}.
+ */
+public class BatchExecLegacySink<T> extends CommonExecLegacySink<T>
+ implements BatchExecNode<Object> {
+
+ public BatchExecLegacySink(
+ TableSink<T> tableSink,
+ @Nullable String[] upsertKeys,
+ ExecEdge inputEdge,
+ LogicalType outputType,
+ String description) {
+ super(tableSink, upsertKeys, false, false, inputEdge, outputType, description);
+ }
+
+ @Override
+ protected RowType checkAndConvertInputTypeIfNeeded(RowType inputRowType) {
+ final DataType resultDataType = tableSink.getConsumedDataType();
+ validateType(resultDataType);
+ return inputRowType;
+ }
+
+ /**
+ * Validate if class represented by the typeInfo is static and globally accessible.
+ *
+ * @param dataType type to check
+ * @throws TableException if type does not meet these criteria
+ */
+ private void validateType(DataType dataType) {
+ Class<?> clazz = dataType.getConversionClass();
+ if (clazz == null) {
+ clazz =
+ ClassLogicalTypeConverter.getDefaultExternalClassForType(
+ dataType.getLogicalType());
+ }
+ if (clazz.isMemberClass() && !Modifier.isStatic(clazz.getModifiers())
+ || !Modifier.isPublic(clazz.getModifiers())
+ || clazz.getCanonicalName() == null) {
+ throw new TableException(
+ String.format(
+ "Class '%s' described in type information '%s' must be static and globally accessible.",
+ clazz, dataType));
+ }
+ }
+}
diff --git a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/utils/ExecNodePlanDumper.java b/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/utils/ExecNodePlanDumper.java
index 2e2c79b..6040ff8 100644
--- a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/utils/ExecNodePlanDumper.java
+++ b/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/utils/ExecNodePlanDumper.java
@@ -18,7 +18,6 @@
package org.apache.flink.table.planner.plan.nodes.exec.utils;
-import org.apache.flink.table.planner.plan.nodes.calcite.LegacySink;
import org.apache.flink.table.planner.plan.nodes.exec.ExecNode;
import org.apache.flink.table.planner.plan.nodes.exec.common.CommonExecLegacySink;
import org.apache.flink.table.planner.plan.nodes.exec.common.CommonExecSink;
@@ -159,8 +158,7 @@ public class ExecNodePlanDumper {
int reuseId = reuseInfo.getReuseId(node);
boolean isReuseNode = reuseId >= 0;
- if (node instanceof LegacySink
- || node instanceof CommonExecLegacySink
+ if (node instanceof CommonExecLegacySink
|| node instanceof CommonExecSink
|| (isReuseNode && isFirstVisit)) {
if (isReuseNode) {
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchExecLegacySink.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchExecLegacySink.scala
deleted file mode 100644
index a90d9c3..0000000
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchExecLegacySink.scala
+++ /dev/null
@@ -1,171 +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.planner.plan.nodes.physical.batch
-
-import org.apache.flink.api.dag.Transformation
-import org.apache.flink.streaming.api.datastream.DataStream
-import org.apache.flink.table.api.TableException
-import org.apache.flink.table.data.RowData
-import org.apache.flink.table.planner.codegen.SinkCodeGenerator._
-import org.apache.flink.table.planner.codegen.{CodeGenUtils, CodeGeneratorContext}
-import org.apache.flink.table.planner.delegation.BatchPlanner
-import org.apache.flink.table.planner.plan.nodes.calcite.LegacySink
-import org.apache.flink.table.planner.plan.nodes.exec.batch.BatchExecNode
-import org.apache.flink.table.planner.plan.nodes.exec.utils.ExecNodeUtil
-import org.apache.flink.table.planner.plan.nodes.exec.{ExecEdge, LegacyBatchExecNode}
-import org.apache.flink.table.planner.plan.utils.UpdatingPlanChecker
-import org.apache.flink.table.planner.sinks.{DataStreamTableSink, TableSinkUtils}
-import org.apache.flink.table.runtime.types.ClassLogicalTypeConverter
-import org.apache.flink.table.runtime.typeutils.InternalTypeInfo
-import org.apache.flink.table.sinks.{RetractStreamTableSink, StreamTableSink, TableSink, UpsertStreamTableSink}
-import org.apache.flink.table.types.DataType
-
-import org.apache.calcite.plan.{RelOptCluster, RelTraitSet}
-import org.apache.calcite.rel.RelNode
-
-import java.lang.reflect.Modifier
-import java.util
-
-import scala.collection.JavaConversions._
-
-/**
- * Batch physical RelNode to to write data into an external sink defined by a [[TableSink]].
- */
-class BatchExecLegacySink[T](
- cluster: RelOptCluster,
- traitSet: RelTraitSet,
- inputRel: RelNode,
- sink: TableSink[T],
- sinkName: String)
- extends LegacySink(cluster, traitSet, inputRel, sink, sinkName)
- with BatchPhysicalRel
- with LegacyBatchExecNode[Any] {
-
- override def copy(traitSet: RelTraitSet, inputs: util.List[RelNode]): RelNode = {
- new BatchExecLegacySink(cluster, traitSet, inputs.get(0), sink, sinkName)
- }
-
- //~ ExecNode methods -----------------------------------------------------------
-
- // the input records will not trigger any output of a sink because it has no output,
- // so it's dam behavior is BLOCKING
- override def getInputEdges: util.List[ExecEdge] = List(
- ExecEdge.builder()
- .damBehavior(ExecEdge.DamBehavior.BLOCKING)
- .build())
-
- override protected def translateToPlanInternal(
- planner: BatchPlanner): Transformation[Any] = {
- val resultTransformation = sink match {
- case streamTableSink: StreamTableSink[T] =>
- val transformation = streamTableSink match {
- case _: RetractStreamTableSink[T] =>
- translateToTransformation(withChangeFlag = true, planner)
-
- case upsertSink: UpsertStreamTableSink[T] =>
- upsertSink.setIsAppendOnly(true)
- upsertSink.setKeyFields(
- UpdatingPlanChecker.getUniqueKeyForUpsertSink(this, upsertSink).orNull)
- translateToTransformation(withChangeFlag = true, planner)
-
- case _ =>
- translateToTransformation(withChangeFlag = false, planner)
- }
- val boundedStream = new DataStream(planner.getExecEnv, transformation)
- val dsSink = streamTableSink.consumeDataStream(boundedStream)
- if (dsSink == null) {
- throw new TableException("The StreamTableSink#consumeDataStream(DataStream) must be " +
- "implemented and return the sink transformation DataStreamSink. " +
- s"However, ${sink.getClass.getCanonicalName} doesn't implement this method.")
- }
- dsSink.getTransformation
- case dsTableSink: DataStreamTableSink[T] =>
- // In case of table to bounded stream through Batchplannerironment#toBoundedStream, we
- // insert a DataStreamTableSink then wrap it as a LogicalSink, there is no real batch table
- // sink, so we do not need to invoke TableSink#emitBoundedStream and set resource, just a
- // translation to Transformation is ok.
- translateToTransformation(withChangeFlag = dsTableSink.withChangeFlag, planner)
-
- case _ =>
- throw new TableException(s"Only Support StreamTableSink! " +
- s"However ${sink.getClass.getCanonicalName} is not a StreamTableSink.")
- }
- resultTransformation.asInstanceOf[Transformation[Any]]
- }
-
- private def translateToTransformation(
- withChangeFlag: Boolean,
- planner: BatchPlanner): Transformation[T] = {
- val config = planner.getTableConfig
- val resultDataType = sink.getConsumedDataType
- validateType(resultDataType)
- val inputNode = getInputNodes.get(0)
- inputNode match {
- // Sink's input must be LegacyBatchExecNode[RowData] or BatchExecNode[RowData] now.
- case _: LegacyBatchExecNode[RowData] | _: BatchExecNode[RowData] =>
- val plan = inputNode.translateToPlan(planner).asInstanceOf[Transformation[T]]
- if (CodeGenUtils.isInternalClass(resultDataType)) {
- plan
- } else {
- val inputRowType = plan.getOutputType.asInstanceOf[InternalTypeInfo[RowData]].toRowType
- val physicalOutputType = TableSinkUtils.inferSinkPhysicalDataType(
- sink.getConsumedDataType,
- inputRowType,
- withChangeFlag)
- val outputTypeInfo = deriveSinkOutputTypeInfo[T](sink, physicalOutputType, withChangeFlag)
- val converterOperator = generateRowConverterOperator[T](
- CodeGeneratorContext(config),
- inputRowType,
- sink,
- physicalOutputType,
- withChangeFlag,
- "SinkConversion")
- ExecNodeUtil.createOneInputTransformation(
- plan,
- s"SinkConversionTo${resultDataType.getConversionClass.getSimpleName}",
- converterOperator,
- outputTypeInfo,
- plan.getParallelism,
- 0)
- }
- case _ =>
- throw new TableException("Cannot generate BoundedStream due to an invalid logical plan. " +
- "This is a bug and should not happen. Please file an issue.")
- }
- }
-
- /**
- * Validate if class represented by the typeInfo is static and globally accessible
- * @param dataType type to check
- * @throws TableException if type does not meet these criteria
- */
- private def validateType(dataType: DataType): Unit = {
- var clazz = dataType.getConversionClass
- if (clazz == null) {
- clazz = ClassLogicalTypeConverter.getDefaultExternalClassForType(dataType.getLogicalType)
- }
- if ((clazz.isMemberClass && !Modifier.isStatic(clazz.getModifiers)) ||
- !Modifier.isPublic(clazz.getModifiers) ||
- clazz.getCanonicalName == null) {
- throw new TableException(
- s"Class '$clazz' described in type information '$dataType' must be " +
- s"static and globally accessible.")
- }
- }
-}
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchPhysicalLegacySink.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchPhysicalLegacySink.scala
new file mode 100644
index 0000000..a5d9f28
--- /dev/null
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchPhysicalLegacySink.scala
@@ -0,0 +1,67 @@
+/*
+ * 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.planner.plan.nodes.physical.batch
+
+import org.apache.flink.table.planner.plan.nodes.calcite.LegacySink
+import org.apache.flink.table.planner.plan.nodes.exec.batch.BatchExecLegacySink
+import org.apache.flink.table.planner.plan.nodes.exec.{ExecEdge, ExecNode}
+import org.apache.flink.table.planner.plan.utils.UpdatingPlanChecker
+import org.apache.flink.table.runtime.types.LogicalTypeDataTypeConverter.fromDataTypeToLogicalType
+import org.apache.flink.table.sinks.{TableSink, UpsertStreamTableSink}
+
+import org.apache.calcite.plan.{RelOptCluster, RelTraitSet}
+import org.apache.calcite.rel.RelNode
+
+import java.util
+
+/**
+ * Batch physical RelNode to to write data into an external sink defined by a [[TableSink]].
+ *
+ * @tparam T The return type of the [[TableSink]].
+ */
+class BatchPhysicalLegacySink[T](
+ cluster: RelOptCluster,
+ traitSet: RelTraitSet,
+ inputRel: RelNode,
+ sink: TableSink[T],
+ sinkName: String)
+ extends LegacySink(cluster, traitSet, inputRel, sink, sinkName)
+ with BatchPhysicalRel {
+
+ override def copy(traitSet: RelTraitSet, inputs: util.List[RelNode]): RelNode = {
+ new BatchPhysicalLegacySink(cluster, traitSet, inputs.get(0), sink, sinkName)
+ }
+
+ override def translateToExecNode(): ExecNode[_] = {
+ val upsertKeys = sink match {
+ case upsertSink: UpsertStreamTableSink[T] =>
+ UpdatingPlanChecker.getUniqueKeyForUpsertSink(this, upsertSink)
+ case _ => Option.empty[Array[String]]
+ }
+ new BatchExecLegacySink[T](
+ sink,
+ upsertKeys.orNull,
+ // the input records will not trigger any output of a sink because it has no output,
+ // so it's dam behavior is BLOCKING
+ ExecEdge.builder().damBehavior(ExecEdge.DamBehavior.BLOCKING).build(),
+ fromDataTypeToLogicalType(sink.getConsumedDataType),
+ getRelDetailedDescription
+ )
+ }
+}
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/optimize/BatchCommonSubGraphBasedOptimizer.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/optimize/BatchCommonSubGraphBasedOptimizer.scala
index 20fd8b2..9610e9b 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/optimize/BatchCommonSubGraphBasedOptimizer.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/optimize/BatchCommonSubGraphBasedOptimizer.scala
@@ -22,7 +22,7 @@ import org.apache.flink.table.api.TableConfig
import org.apache.flink.table.catalog.{CatalogManager, FunctionCatalog}
import org.apache.flink.table.planner.calcite.{FlinkContext, SqlExprToRexConverterFactory}
import org.apache.flink.table.planner.delegation.BatchPlanner
-import org.apache.flink.table.planner.plan.nodes.physical.batch.BatchExecLegacySink
+import org.apache.flink.table.planner.plan.nodes.physical.batch.BatchPhysicalLegacySink
import org.apache.flink.table.planner.plan.optimize.program.{BatchOptimizeContext, FlinkBatchProgram}
import org.apache.flink.table.planner.plan.schema.IntermediateRelTable
import org.apache.flink.table.planner.utils.TableConfigUtils
@@ -57,7 +57,7 @@ class BatchCommonSubGraphBasedOptimizer(planner: BatchPlanner)
val optimizedTree = optimizeTree(originTree)
optimizedTree match {
- case _: BatchExecLegacySink[_] => // ignore
+ case _: BatchPhysicalLegacySink[_] => // ignore
case _ =>
val name = createUniqueIntermediateRelTableName
val intermediateRelTable = new IntermediateRelTable(Collections.singletonList(name),
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/rules/FlinkBatchRuleSets.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/rules/FlinkBatchRuleSets.scala
index 6b43ffe..2448e41 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/rules/FlinkBatchRuleSets.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/rules/FlinkBatchRuleSets.scala
@@ -435,7 +435,7 @@ object FlinkBatchRuleSets {
BatchPhysicalPythonCorrelateRule.INSTANCE,
// sink
BatchPhysicalSinkRule.INSTANCE,
- BatchExecLegacySinkRule.INSTANCE
+ BatchPhysicalLegacySinkRule.INSTANCE
)
/**
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/batch/BatchExecLegacySinkRule.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/batch/BatchPhysicalLegacySinkRule.scala
similarity index 93%
rename from flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/batch/BatchExecLegacySinkRule.scala
rename to flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/batch/BatchPhysicalLegacySinkRule.scala
index e4e84cf..59bd020 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/batch/BatchExecLegacySinkRule.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/batch/BatchPhysicalLegacySinkRule.scala
@@ -23,7 +23,7 @@ import org.apache.flink.table.filesystem.FileSystemOptions
import org.apache.flink.table.planner.plan.`trait`.FlinkRelDistribution
import org.apache.flink.table.planner.plan.nodes.FlinkConventions
import org.apache.flink.table.planner.plan.nodes.logical.FlinkLogicalLegacySink
-import org.apache.flink.table.planner.plan.nodes.physical.batch.BatchExecLegacySink
+import org.apache.flink.table.planner.plan.nodes.physical.batch.BatchPhysicalLegacySink
import org.apache.flink.table.planner.plan.utils.FlinkRelOptUtil
import org.apache.flink.table.sinks.PartitionableTableSink
@@ -33,11 +33,11 @@ import org.apache.calcite.rel.{RelCollations, RelNode}
import scala.collection.JavaConversions._
-class BatchExecLegacySinkRule extends ConverterRule(
+class BatchPhysicalLegacySinkRule extends ConverterRule(
classOf[FlinkLogicalLegacySink],
FlinkConventions.LOGICAL,
FlinkConventions.BATCH_PHYSICAL,
- "BatchExecLegacySinkRule") {
+ "BatchPhysicalLegacySinkRule") {
def convert(rel: RelNode): RelNode = {
val sinkNode = rel.asInstanceOf[FlinkLogicalLegacySink]
@@ -78,7 +78,7 @@ class BatchExecLegacySinkRule extends ConverterRule(
val newInput = RelOptRule.convert(sinkNode.getInput, requiredTraitSet)
- new BatchExecLegacySink(
+ new BatchPhysicalLegacySink(
rel.getCluster,
newTrait,
newInput,
@@ -87,8 +87,6 @@ class BatchExecLegacySinkRule extends ConverterRule(
}
}
-object BatchExecLegacySinkRule {
-
- val INSTANCE: RelOptRule = new BatchExecLegacySinkRule
-
+object BatchPhysicalLegacySinkRule {
+ val INSTANCE: RelOptRule = new BatchPhysicalLegacySinkRule
}