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:49 UTC
[flink] 01/05: [FLINK-20949][table-planner-blink] Introduce
StreamPhysicalSink, and make StreamExecSink 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 95257a255f0da0a95b31647c6d057914d5748871
Author: godfreyhe <go...@163.com>
AuthorDate: Wed Jan 13 20:51:43 2021 +0800
[FLINK-20949][table-planner-blink] Introduce StreamPhysicalSink, and make StreamExecSink only extended from ExecNode
This closes #14637
---
.../plan/nodes/exec/common/CommonExecSink.java | 202 +++++++++++++++++++++
.../plan/nodes/exec/stream/StreamExecSink.java | 97 ++++++++++
.../plan/nodes/exec/utils/ExecNodePlanDumper.java | 2 +
.../plan/nodes/common/CommonPhysicalSink.scala | 2 +-
.../nodes/physical/stream/StreamExecSink.scala | 102 -----------
.../nodes/physical/stream/StreamPhysicalSink.scala | 68 +++++++
.../FlinkChangelogModeInferenceProgram.scala | 4 +-
.../planner/plan/rules/FlinkStreamRuleSets.scala | 2 +-
...SinkRule.scala => StreamPhysicalSinkRule.scala} | 12 +-
.../flink/table/planner/sinks/TableSinkUtils.scala | 4 +-
10 files changed, 381 insertions(+), 114 deletions(-)
diff --git a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecSink.java b/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecSink.java
new file mode 100644
index 0000000..a04468c
--- /dev/null
+++ b/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecSink.java
@@ -0,0 +1,202 @@
+/*
+ * 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.common;
+
+import org.apache.flink.api.common.io.OutputFormat;
+import org.apache.flink.api.dag.Transformation;
+import org.apache.flink.api.java.typeutils.InputTypeConfigurable;
+import org.apache.flink.runtime.state.KeyGroupRangeAssignment;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.sink.OutputFormatSinkFunction;
+import org.apache.flink.streaming.api.functions.sink.SinkFunction;
+import org.apache.flink.streaming.api.operators.SimpleOperatorFactory;
+import org.apache.flink.streaming.api.transformations.LegacySinkTransformation;
+import org.apache.flink.streaming.api.transformations.PartitionTransformation;
+import org.apache.flink.streaming.runtime.partitioner.KeyGroupStreamPartitioner;
+import org.apache.flink.table.api.TableConfig;
+import org.apache.flink.table.api.TableException;
+import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.api.config.ExecutionConfigOptions;
+import org.apache.flink.table.connector.ChangelogMode;
+import org.apache.flink.table.connector.ParallelismProvider;
+import org.apache.flink.table.connector.sink.DataStreamSinkProvider;
+import org.apache.flink.table.connector.sink.DynamicTableSink;
+import org.apache.flink.table.connector.sink.OutputFormatProvider;
+import org.apache.flink.table.connector.sink.SinkFunctionProvider;
+import org.apache.flink.table.data.RowData;
+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.ExecNodeBase;
+import org.apache.flink.table.planner.plan.utils.KeySelectorUtil;
+import org.apache.flink.table.planner.sinks.TableSinkUtils;
+import org.apache.flink.table.runtime.connector.sink.SinkRuntimeProviderContext;
+import org.apache.flink.table.runtime.keyselector.RowDataKeySelector;
+import org.apache.flink.table.runtime.operators.sink.SinkNotNullEnforcer;
+import org.apache.flink.table.runtime.operators.sink.SinkOperator;
+import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.table.utils.TableSchemaUtils;
+import org.apache.flink.types.RowKind;
+import org.apache.flink.util.Preconditions;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+import java.util.stream.Collectors;
+
+/**
+ * Base {@link ExecNode} to write data to an external sink defined by a {@link DynamicTableSink}.
+ */
+public abstract class CommonExecSink extends ExecNodeBase<Object> {
+ protected final List<String> qualifiedName;
+ private final TableSchema tableSchema;
+ private final DynamicTableSink tableSink;
+ private final ChangelogMode changelogMode;
+ private final boolean isBounded;
+
+ public CommonExecSink(
+ List<String> qualifiedName,
+ TableSchema tableSchema,
+ DynamicTableSink tableSink,
+ ChangelogMode changelogMode,
+ boolean isBounded,
+ ExecEdge inputEdge,
+ LogicalType outputType,
+ String description) {
+ super(Collections.singletonList(inputEdge), outputType, description);
+ this.tableSink = tableSink;
+ this.qualifiedName = qualifiedName;
+ this.tableSchema = tableSchema;
+ this.changelogMode = changelogMode;
+ this.isBounded = isBounded;
+ }
+
+ protected Transformation<Object> createSinkTransformation(
+ StreamExecutionEnvironment env,
+ TableConfig tableConfig,
+ Transformation<RowData> inputTransform,
+ int rowtimeFieldIndex) {
+ final DynamicTableSink.SinkRuntimeProvider runtimeProvider =
+ tableSink.getSinkRuntimeProvider(new SinkRuntimeProviderContext(isBounded));
+
+ final ExecutionConfigOptions.NotNullEnforcer notNullEnforcer =
+ tableConfig
+ .getConfiguration()
+ .get(ExecutionConfigOptions.TABLE_EXEC_SINK_NOT_NULL_ENFORCER);
+ final int[] notNullFieldIndices = TableSinkUtils.getNotNullFieldIndices(tableSchema);
+ final String[] fieldNames =
+ ((RowType) tableSchema.toPhysicalRowDataType().getLogicalType())
+ .getFieldNames()
+ .toArray(new String[0]);
+ final SinkNotNullEnforcer enforcer =
+ new SinkNotNullEnforcer(notNullEnforcer, notNullFieldIndices, fieldNames);
+ final InternalTypeInfo<RowData> inputTypeInfo =
+ InternalTypeInfo.of(getInputNodes().get(0).getOutputType());
+
+ if (runtimeProvider instanceof DataStreamSinkProvider) {
+ if (runtimeProvider instanceof ParallelismProvider) {
+ throw new TableException(
+ "`DataStreamSinkProvider` is not allowed to work with"
+ + " `ParallelismProvider`, "
+ + "please see document of `ParallelismProvider`");
+ } else {
+ final DataStream<RowData> dataStream =
+ new DataStream<>(env, inputTransform).filter(enforcer);
+ final DataStreamSinkProvider provider = (DataStreamSinkProvider) runtimeProvider;
+ return provider.consumeDataStream(dataStream).getTransformation();
+ }
+ } else {
+ Preconditions.checkArgument(
+ runtimeProvider instanceof ParallelismProvider,
+ "runtimeProvider with `ParallelismProvider` implementation is required");
+
+ final SinkFunction<RowData> sinkFunction;
+ if (runtimeProvider instanceof SinkFunctionProvider) {
+ sinkFunction = ((SinkFunctionProvider) runtimeProvider).createSinkFunction();
+ } else if (runtimeProvider instanceof OutputFormatProvider) {
+ OutputFormat<RowData> outputFormat =
+ ((OutputFormatProvider) runtimeProvider).createOutputFormat();
+ sinkFunction = new OutputFormatSinkFunction<>(outputFormat);
+ } else {
+ throw new TableException("This should not happen.");
+ }
+
+ if (sinkFunction instanceof InputTypeConfigurable) {
+ ((InputTypeConfigurable) sinkFunction).setInputType(inputTypeInfo, env.getConfig());
+ }
+
+ final SinkOperator operator =
+ new SinkOperator(env.clean(sinkFunction), rowtimeFieldIndex, enforcer);
+
+ final int inputParallelism = inputTransform.getParallelism();
+ final int parallelism;
+ final Optional<Integer> parallelismOptional =
+ ((ParallelismProvider) runtimeProvider).getParallelism();
+ if (parallelismOptional.isPresent()) {
+ parallelism = parallelismOptional.get();
+ if (parallelism <= 0) {
+ throw new TableException(
+ String.format(
+ "Table: %s configured sink parallelism: "
+ + "%s should not be less than zero or equal to zero",
+ String.join(".", qualifiedName), parallelism));
+ }
+ } else {
+ parallelism = inputParallelism;
+ }
+
+ final int[] primaryKeys = TableSchemaUtils.getPrimaryKeyIndices(tableSchema);
+ final Transformation<RowData> finalInputTransform;
+ if (inputParallelism == parallelism || changelogMode.containsOnly(RowKind.INSERT)) {
+ // if the inputParallelism is equals to the parallelism or insert-only mode, do
+ // nothing.
+ finalInputTransform = inputTransform;
+ } else if (primaryKeys.length == 0) {
+ throw new TableException(
+ String.format(
+ "Table: %s configured sink parallelism is: %s, while the input parallelism is: "
+ + "%s. Since configured parallelism is different from input parallelism and the changelog mode "
+ + "contains [%s], which is not INSERT_ONLY mode, primary key is required but no primary key is found",
+ String.join(".", qualifiedName),
+ parallelism,
+ inputParallelism,
+ changelogMode.getContainedKinds().stream()
+ .map(Enum::toString)
+ .collect(Collectors.joining(","))));
+ } else {
+ // key by before sink
+ final RowDataKeySelector selector =
+ KeySelectorUtil.getRowDataSelector(primaryKeys, inputTypeInfo);
+ final KeyGroupStreamPartitioner<RowData, RowData> partitioner =
+ new KeyGroupStreamPartitioner<>(
+ selector,
+ KeyGroupRangeAssignment.DEFAULT_LOWER_BOUND_MAX_PARALLELISM);
+ finalInputTransform = new PartitionTransformation<>(inputTransform, partitioner);
+ finalInputTransform.setParallelism(parallelism);
+ }
+ return new LegacySinkTransformation<>(
+ finalInputTransform,
+ getDesc(),
+ SimpleOperatorFactory.of(operator),
+ parallelism);
+ }
+ }
+}
diff --git a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecSink.java b/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecSink.java
new file mode 100644
index 0000000..8f14238
--- /dev/null
+++ b/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecSink.java
@@ -0,0 +1,97 @@
+/*
+ * 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.stream;
+
+import org.apache.flink.api.dag.Transformation;
+import org.apache.flink.table.api.TableException;
+import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.connector.ChangelogMode;
+import org.apache.flink.table.connector.sink.DynamicTableSink;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.planner.delegation.PlannerBase;
+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.CommonExecSink;
+import org.apache.flink.table.runtime.typeutils.TypeCheckUtils;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.RowType;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.stream.Collectors;
+
+/**
+ * Stream {@link ExecNode} to to write data into an external sink defined by a {@link
+ * DynamicTableSink}.
+ */
+public class StreamExecSink extends CommonExecSink implements StreamExecNode<Object> {
+
+ public StreamExecSink(
+ List<String> qualifiedName,
+ TableSchema tableSchema,
+ DynamicTableSink tableSink,
+ ChangelogMode inputChangelogMode,
+ ExecEdge inputEdge,
+ LogicalType outputType,
+ String description) {
+ super(
+ qualifiedName,
+ tableSchema,
+ tableSink,
+ tableSink.getChangelogMode(inputChangelogMode),
+ false, // isBounded
+ inputEdge,
+ outputType,
+ description);
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ protected Transformation<Object> translateToPlanInternal(PlannerBase planner) {
+ final ExecNode<RowData> inputNode = (ExecNode<RowData>) getInputNodes().get(0);
+ final Transformation<RowData> inputTransform = inputNode.translateToPlan(planner);
+ final RowType inputRowType = (RowType) inputNode.getOutputType();
+
+ final List<Integer> rowtimeFieldIndices = new ArrayList<>();
+ for (int i = 0; i < inputRowType.getFieldCount(); ++i) {
+ if (TypeCheckUtils.isRowTime(inputRowType.getTypeAt(i))) {
+ rowtimeFieldIndices.add(i);
+ }
+ }
+ final int rowtimeFieldIndex;
+ if (rowtimeFieldIndices.size() > 1) {
+ throw new TableException(
+ String.format(
+ "Found more than one rowtime field: [%s] in the query when insert into '%s'.\n"
+ + "Please select the rowtime field that should be used as event-time timestamp "
+ + "for the DataStream by casting all other fields to TIMESTAMP.",
+ rowtimeFieldIndices.stream()
+ .map(i -> inputRowType.getFieldNames().get(i))
+ .collect(Collectors.joining(", ")),
+ String.join(".", qualifiedName)));
+ } else if (rowtimeFieldIndices.size() == 1) {
+ rowtimeFieldIndex = rowtimeFieldIndices.get(0);
+ } else {
+ rowtimeFieldIndex = -1;
+ }
+
+ return createSinkTransformation(
+ planner.getExecEnv(), planner.getTableConfig(), inputTransform, rowtimeFieldIndex);
+ }
+}
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 03f4239..b39d9df 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
@@ -21,6 +21,7 @@ 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.calcite.Sink;
import org.apache.flink.table.planner.plan.nodes.exec.ExecNode;
+import org.apache.flink.table.planner.plan.nodes.exec.common.CommonExecSink;
import org.apache.flink.table.planner.plan.nodes.exec.visitor.ExecNodeVisitor;
import org.apache.flink.table.planner.plan.nodes.exec.visitor.ExecNodeVisitorImpl;
import org.apache.flink.util.Preconditions;
@@ -159,6 +160,7 @@ public class ExecNodePlanDumper {
int reuseId = reuseInfo.getReuseId(node);
boolean isReuseNode = reuseId >= 0;
if (node instanceof LegacySink
+ || node instanceof CommonExecSink
|| node instanceof Sink
|| (isReuseNode && isFirstVisit)) {
if (isReuseNode) {
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/common/CommonPhysicalSink.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/common/CommonPhysicalSink.scala
index 094be0c..95037a6 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/common/CommonPhysicalSink.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/common/CommonPhysicalSink.scala
@@ -79,7 +79,7 @@ class CommonPhysicalSink (
val notNullEnforcer = tableConfig.getConfiguration
.get(ExecutionConfigOptions.TABLE_EXEC_SINK_NOT_NULL_ENFORCER)
- val notNullFieldIndices = TableSinkUtils.getNotNullFieldIndices(catalogTable)
+ val notNullFieldIndices = TableSinkUtils.getNotNullFieldIndices(catalogTable.getSchema)
val fieldNames = catalogTable.getSchema.toPhysicalRowDataType
.getLogicalType.asInstanceOf[RowType]
.getFieldNames
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamExecSink.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamExecSink.scala
deleted file mode 100644
index df5a58f..0000000
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamExecSink.scala
+++ /dev/null
@@ -1,102 +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.stream
-
-import org.apache.flink.api.dag.Transformation
-import org.apache.flink.table.api.TableException
-import org.apache.flink.table.catalog.{CatalogTable, ObjectIdentifier}
-import org.apache.flink.table.connector.sink.DynamicTableSink
-import org.apache.flink.table.data.RowData
-import org.apache.flink.table.planner.calcite.FlinkTypeFactory
-import org.apache.flink.table.planner.delegation.StreamPlanner
-import org.apache.flink.table.planner.plan.nodes.common.CommonPhysicalSink
-import org.apache.flink.table.planner.plan.nodes.exec.LegacyStreamExecNode
-import org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecNode
-import org.apache.flink.table.planner.plan.utils.ChangelogPlanUtils
-
-import org.apache.calcite.plan.{RelOptCluster, RelTraitSet}
-import org.apache.calcite.rel.RelNode
-
-import java.util
-
-import scala.collection.JavaConversions._
-
-/**
- * Stream physical RelNode to to write data into an external sink defined by a
- * [[DynamicTableSink]].
- */
-class StreamExecSink(
- cluster: RelOptCluster,
- traitSet: RelTraitSet,
- inputRel: RelNode,
- tableIdentifier: ObjectIdentifier,
- catalogTable: CatalogTable,
- tableSink: DynamicTableSink)
- extends CommonPhysicalSink(cluster, traitSet, inputRel, tableIdentifier, catalogTable, tableSink)
- with StreamPhysicalRel
- with LegacyStreamExecNode[Any] {
-
- override def requireWatermark: Boolean = false
-
- override def copy(traitSet: RelTraitSet, inputs: util.List[RelNode]): RelNode = {
- new StreamExecSink(cluster, traitSet, inputs.get(0), tableIdentifier, catalogTable, tableSink)
- }
-
- //~ ExecNode methods -----------------------------------------------------------
-
- override protected def translateToPlanInternal(
- planner: StreamPlanner): Transformation[Any] = {
-
- // get RowData plan
- val inputTransformation = getInputNodes.get(0) match {
- // Sink's input must be LegacyStreamExecNode[RowData] or StreamExecNode[RowData] now.
- case legacyNode: LegacyStreamExecNode[RowData] => legacyNode.translateToPlan(planner)
- case node: StreamExecNode[RowData] => node.translateToPlan(planner)
- case _ =>
- throw new TableException("Cannot generate DataStream due to an invalid logical plan. " +
- "This is a bug and should not happen. Please file an issue.")
- }
- val inputLogicalType = getInput.getRowType
- val rowtimeFields = inputLogicalType.getFieldList.zipWithIndex
- .filter { case (f, _) =>
- FlinkTypeFactory.isRowtimeIndicatorType(f.getType)
- }
- if (rowtimeFields.size > 1) {
- throw new TableException(
- s"Found more than one rowtime field: [${rowtimeFields.map(_._1.getName).mkString(", ")}]" +
- s" in the query when insert into '${tableIdentifier.asSummaryString()}'.\n" +
- s"Please select the rowtime field that should be used as event-time timestamp for the " +
- s"DataStream by casting all other fields to TIMESTAMP.")
- }
-
- val inputChangelogMode = ChangelogPlanUtils.getChangelogMode(
- getInput.asInstanceOf[StreamPhysicalRel]).get
- // tell sink the ChangelogMode of input
- val changelogMode = tableSink.getChangelogMode(inputChangelogMode)
- val rowtimeFieldIndex: Int = rowtimeFields.map(_._2).headOption.getOrElse(-1)
-
- createSinkTransformation(
- planner.getExecEnv,
- inputTransformation,
- planner.getTableConfig,
- rowtimeFieldIndex,
- isBounded = false,
- changelogMode = changelogMode)
- }
-}
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamPhysicalSink.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamPhysicalSink.scala
new file mode 100644
index 0000000..ff0828f
--- /dev/null
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamPhysicalSink.scala
@@ -0,0 +1,68 @@
+/*
+ * 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.stream
+
+import org.apache.flink.table.catalog.{CatalogTable, ObjectIdentifier}
+import org.apache.flink.table.connector.sink.DynamicTableSink
+import org.apache.flink.table.planner.calcite.FlinkTypeFactory
+import org.apache.flink.table.planner.plan.nodes.calcite.Sink
+import org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecSink
+import org.apache.flink.table.planner.plan.nodes.exec.{ExecEdge, ExecNode}
+import org.apache.flink.table.planner.plan.utils.ChangelogPlanUtils
+
+import org.apache.calcite.plan.{RelOptCluster, RelTraitSet}
+import org.apache.calcite.rel.RelNode
+
+import java.util
+
+/**
+ * Stream physical RelNode to to write data into an external sink defined by a
+ * [[DynamicTableSink]].
+ */
+class StreamPhysicalSink(
+ cluster: RelOptCluster,
+ traitSet: RelTraitSet,
+ inputRel: RelNode,
+ tableIdentifier: ObjectIdentifier,
+ catalogTable: CatalogTable,
+ tableSink: DynamicTableSink)
+ extends Sink(cluster, traitSet, inputRel, tableIdentifier, catalogTable, tableSink)
+ with StreamPhysicalRel {
+
+ override def requireWatermark: Boolean = false
+
+ override def copy(traitSet: RelTraitSet, inputs: util.List[RelNode]): RelNode = {
+ new StreamPhysicalSink(
+ cluster, traitSet, inputs.get(0), tableIdentifier, catalogTable, tableSink)
+ }
+
+ override def translateToExecNode(): ExecNode[_] = {
+ val inputChangelogMode = ChangelogPlanUtils.getChangelogMode(
+ getInput.asInstanceOf[StreamPhysicalRel]).get
+ new StreamExecSink(
+ tableIdentifier.toList,
+ catalogTable.getSchema,
+ tableSink,
+ inputChangelogMode,
+ ExecEdge.DEFAULT,
+ FlinkTypeFactory.toLogicalRowType(getRowType),
+ getRelDetailedDescription
+ )
+ }
+}
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/optimize/program/FlinkChangelogModeInferenceProgram.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/optimize/program/FlinkChangelogModeInferenceProgram.scala
index aa66f09..1112649 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/optimize/program/FlinkChangelogModeInferenceProgram.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/optimize/program/FlinkChangelogModeInferenceProgram.scala
@@ -117,7 +117,7 @@ class FlinkChangelogModeInferenceProgram extends FlinkOptimizeProgram[StreamOpti
rel: StreamPhysicalRel,
requiredTrait: ModifyKindSetTrait,
requester: String): StreamPhysicalRel = rel match {
- case sink: StreamExecSink =>
+ case sink: StreamPhysicalSink =>
val name = s"Table sink '${sink.tableIdentifier.asSummaryString()}'"
val queryModifyKindSet = deriveQueryDefaultChangelogMode(sink.getInput, name)
val sinkRequiredTrait = ModifyKindSetTrait.fromChangelogMode(
@@ -414,7 +414,7 @@ class FlinkChangelogModeInferenceProgram extends FlinkOptimizeProgram[StreamOpti
def visit(
rel: StreamPhysicalRel,
requiredTrait: UpdateKindTrait): Option[StreamPhysicalRel] = rel match {
- case sink: StreamExecSink =>
+ case sink: StreamPhysicalSink =>
val childModifyKindSet = getModifyKindSet(sink.getInput)
val onlyAfter = onlyAfterOrNone(childModifyKindSet)
val beforeAndAfter = beforeAfterOrNone(childModifyKindSet)
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/rules/FlinkStreamRuleSets.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/rules/FlinkStreamRuleSets.scala
index 229b802..fe7cb9b 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/rules/FlinkStreamRuleSets.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/rules/FlinkStreamRuleSets.scala
@@ -444,7 +444,7 @@ object FlinkStreamRuleSets {
StreamPhysicalCorrelateRule.INSTANCE,
StreamPhysicalPythonCorrelateRule.INSTANCE,
// sink
- StreamExecSinkRule.INSTANCE,
+ StreamPhysicalSinkRule.INSTANCE,
StreamExecLegacySinkRule.INSTANCE
)
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/stream/StreamExecSinkRule.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/stream/StreamPhysicalSinkRule.scala
similarity index 94%
rename from flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/stream/StreamExecSinkRule.scala
rename to flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/stream/StreamPhysicalSinkRule.scala
index a92265e..686377f 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/stream/StreamExecSinkRule.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/stream/StreamPhysicalSinkRule.scala
@@ -24,7 +24,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.FlinkLogicalSink
-import org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecSink
+import org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalSink
import org.apache.flink.table.types.logical.RowType
import org.apache.calcite.plan.RelOptRule
@@ -33,11 +33,11 @@ import org.apache.calcite.rel.convert.ConverterRule
import scala.collection.JavaConversions._
-class StreamExecSinkRule extends ConverterRule(
+class StreamPhysicalSinkRule extends ConverterRule(
classOf[FlinkLogicalSink],
FlinkConventions.LOGICAL,
FlinkConventions.STREAM_PHYSICAL,
- "StreamExecSinkRule") {
+ "StreamPhysicalSinkRule") {
def convert(rel: RelNode): RelNode = {
val sinkNode = rel.asInstanceOf[FlinkLogicalSink]
@@ -83,7 +83,7 @@ class StreamExecSinkRule extends ConverterRule(
val newInput = RelOptRule.convert(sinkNode.getInput, requiredTraitSet)
- new StreamExecSink(
+ new StreamPhysicalSink(
rel.getCluster,
newTrait,
newInput,
@@ -93,7 +93,7 @@ class StreamExecSinkRule extends ConverterRule(
}
}
-object StreamExecSinkRule {
- val INSTANCE = new StreamExecSinkRule
+object StreamPhysicalSinkRule {
+ val INSTANCE = new StreamPhysicalSinkRule
}
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/sinks/TableSinkUtils.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/sinks/TableSinkUtils.scala
index 5b08c98..ed020eb 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/sinks/TableSinkUtils.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/sinks/TableSinkUtils.scala
@@ -276,8 +276,8 @@ object TableSinkUtils {
/**
* Gets the NOT NULL physical field indices on the [[CatalogTable]].
*/
- def getNotNullFieldIndices(catalogTable: CatalogTable): Array[Int] = {
- val rowType = catalogTable.getSchema.toPhysicalRowDataType.getLogicalType.asInstanceOf[RowType]
+ def getNotNullFieldIndices(tableSchema: TableSchema): Array[Int] = {
+ val rowType = tableSchema.toPhysicalRowDataType.getLogicalType.asInstanceOf[RowType]
val fieldTypes = rowType.getFields.map(_.getType).toArray
fieldTypes.indices.filter { index =>
!fieldTypes(index).isNullable