You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ku...@apache.org on 2020/05/08 12:57:28 UTC
[flink] 01/05: [FLINK-17267] [table-planner] legacy stream planner
supports explain insert operation
This is an automated email from the ASF dual-hosted git repository.
kurt pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 59ca355f05b668f1cb41cae02ce182ccc7c4d707
Author: godfreyhe <go...@163.com>
AuthorDate: Fri Apr 24 15:22:09 2020 +0800
[FLINK-17267] [table-planner] legacy stream planner supports explain insert operation
---
.../table/calcite/RelTimeIndicatorConverter.scala | 30 +++
.../flink/table/plan/nodes/LogicalSink.scala | 55 +++++
.../org/apache/flink/table/plan/nodes/Sink.scala | 57 +++++
.../plan/nodes/datastream/DataStreamSink.scala | 204 ++++++++++++++++
.../plan/nodes/logical/FlinkLogicalSink.scala | 73 ++++++
.../flink/table/plan/rules/FlinkRuleSets.scala | 6 +-
.../plan/rules/datastream/DataStreamSinkRule.scala | 53 +++++
.../apache/flink/table/planner/StreamPlanner.scala | 263 ++++++---------------
.../flink/table/sinks/DataStreamTableSink.scala | 58 +++++
.../flink/table/api/TableEnvironmentITCase.scala | 4 +-
.../flink/table/api/stream/ExplainTest.scala | 75 ++++--
.../apache/flink/table/utils/TableTestBase.scala | 10 +
.../resources/testFromToDataStreamAndSqlUpdate.out | 23 +-
.../src/test/scala/resources/testInsert.out | 29 +++
.../test/scala/resources/testMultipleInserts.out | 55 +++++
.../resources/testSqlUpdateAndToDataStream.out | 20 +-
16 files changed, 785 insertions(+), 230 deletions(-)
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/calcite/RelTimeIndicatorConverter.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/calcite/RelTimeIndicatorConverter.scala
index 7f85245..82b3f2b 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/calcite/RelTimeIndicatorConverter.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/calcite/RelTimeIndicatorConverter.scala
@@ -32,6 +32,7 @@ import org.apache.flink.table.calcite.FlinkTypeFactory.{isRowtimeIndicatorType,
import org.apache.flink.table.catalog.BasicOperatorTable
import org.apache.flink.table.functions.sql.ProctimeSqlFunction
import org.apache.flink.table.plan.logical.rel._
+import org.apache.flink.table.plan.nodes.LogicalSink
import org.apache.flink.table.plan.schema.TimeIndicatorRelDataType
import java.util.{Collections => JCollections}
@@ -173,6 +174,30 @@ class RelTimeIndicatorConverter(rexBuilder: RexBuilder) extends RelShuttle {
case temporalTableJoin: LogicalTemporalTableJoin =>
visit(temporalTableJoin)
+ case sink: LogicalSink =>
+ var newInput = sink.getInput.accept(this)
+ var needsConversion = false
+
+ val projects = newInput.getRowType.getFieldList.map { field =>
+ if (isProctimeIndicatorType(field.getType)) {
+ needsConversion = true
+ rexBuilder.makeCall(ProctimeSqlFunction, new RexInputRef(field.getIndex, field.getType))
+ } else {
+ new RexInputRef(field.getIndex, field.getType)
+ }
+ }
+
+ // add final conversion if necessary
+ if (needsConversion) {
+ newInput = LogicalProject.create(newInput, projects, newInput.getRowType.getFieldNames)
+ }
+ new LogicalSink(
+ sink.getCluster,
+ sink.getTraitSet,
+ newInput,
+ sink.sink,
+ sink.sinkName)
+
case _ =>
throw new TableException(s"Unsupported logical operator: ${other.getClass.getSimpleName}")
}
@@ -392,6 +417,11 @@ object RelTimeIndicatorConverter {
val converter = new RelTimeIndicatorConverter(rexBuilder)
val convertedRoot = rootRel.accept(converter)
+ // the LogicalSink is converted in RelTimeIndicatorConverter before
+ if (rootRel.isInstanceOf[LogicalSink]) {
+ return convertedRoot
+ }
+
var needsConversion = false
// materialize remaining proctime indicators
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/nodes/LogicalSink.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/nodes/LogicalSink.scala
new file mode 100644
index 0000000..478bb85
--- /dev/null
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/nodes/LogicalSink.scala
@@ -0,0 +1,55 @@
+/*
+ * 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.plan.nodes
+
+import org.apache.flink.table.sinks.TableSink
+
+import org.apache.calcite.plan.{Convention, RelOptCluster, RelTraitSet}
+import org.apache.calcite.rel.RelNode
+
+import java.util
+
+import scala.collection.JavaConversions._
+
+/**
+ * Sub-class of [[Sink]] that is a relational expression
+ * which writes out data of input node into a [[TableSink]].
+ * This class corresponds to Calcite logical rel.
+ */
+final class LogicalSink(
+ cluster: RelOptCluster,
+ traitSet: RelTraitSet,
+ input: RelNode,
+ sink: TableSink[_],
+ sinkName: String)
+ extends Sink(cluster, traitSet, input, sink, sinkName) {
+
+ override def copy(traitSet: RelTraitSet, inputs: util.List[RelNode]): RelNode = {
+ new LogicalSink(cluster, traitSet, inputs.head, sink, sinkName)
+ }
+
+}
+
+object LogicalSink {
+
+ def create(input: RelNode, sink: TableSink[_], sinkName: String): LogicalSink = {
+ val traits = input.getCluster.traitSetOf(Convention.NONE)
+ new LogicalSink(input.getCluster, traits, input, sink, sinkName)
+ }
+}
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/nodes/Sink.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/nodes/Sink.scala
new file mode 100644
index 0000000..cd873d1
--- /dev/null
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/nodes/Sink.scala
@@ -0,0 +1,57 @@
+/*
+ * 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.plan.nodes
+
+import org.apache.flink.table.calcite.FlinkTypeFactory
+import org.apache.flink.table.sinks.TableSink
+
+import org.apache.calcite.plan.{RelOptCluster, RelTraitSet}
+import org.apache.calcite.rel.`type`.RelDataType
+import org.apache.calcite.rel.{RelNode, RelWriter, SingleRel}
+
+/**
+ * Relational expression that writes out data of input node into a [[TableSink]].
+ *
+ * @param cluster cluster that this relational expression belongs to
+ * @param traitSet the traits of this rel
+ * @param input input relational expression
+ * @param sink Table sink to write into
+ * @param sinkName Name of tableSink, which is not required property, that is, it could be null
+ */
+abstract class Sink(
+ cluster: RelOptCluster,
+ traitSet: RelTraitSet,
+ input: RelNode,
+ val sink: TableSink[_],
+ val sinkName: String)
+ extends SingleRel(cluster, traitSet, input) {
+
+ override def deriveRowType(): RelDataType = {
+ val typeFactory = getCluster.getTypeFactory.asInstanceOf[FlinkTypeFactory]
+ val tableSchema = sink.getTableSchema
+ typeFactory.buildLogicalRowType(tableSchema.getFieldNames, tableSchema.getFieldTypes)
+ }
+
+ override def explainTerms(pw: RelWriter): RelWriter = {
+ super.explainTerms(pw)
+ .itemIf("name", sinkName, sinkName != null)
+ .item("fields", sink.getTableSchema.getFieldNames.mkString(", "))
+ }
+
+}
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamSink.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamSink.scala
new file mode 100644
index 0000000..6f2df68
--- /dev/null
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamSink.scala
@@ -0,0 +1,204 @@
+/*
+ * 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.plan.nodes.datastream
+
+import org.apache.flink.api.common.typeinfo.TypeInformation
+import org.apache.flink.api.dag.Transformation
+import org.apache.flink.api.java.tuple.{Tuple2 => JTuple2}
+import org.apache.flink.streaming.api.datastream.{DataStream, DataStreamSink => StreamingDataStreamSink}
+import org.apache.flink.table.api.{TableException, TableSchema, ValidationException}
+import org.apache.flink.table.calcite.FlinkTypeFactory
+import org.apache.flink.table.plan.nodes.Sink
+import org.apache.flink.table.plan.util.UpdatingPlanChecker
+import org.apache.flink.table.planner.{DataStreamConversions, StreamPlanner}
+import org.apache.flink.table.runtime.types.CRow
+import org.apache.flink.table.sinks.{AppendStreamTableSink, DataStreamTableSink, RetractStreamTableSink, TableSink, UpsertStreamTableSink}
+import org.apache.flink.table.types.utils.TypeConversions
+
+import org.apache.calcite.plan._
+import org.apache.calcite.rel.RelNode
+
+import _root_.java.lang.{Boolean => JBool}
+
+import _root_.scala.collection.JavaConverters._
+
+
+/**
+ * Stream physical RelNode to to write data into an external sink defined by a [[TableSink]].
+ */
+class DataStreamSink(
+ cluster: RelOptCluster,
+ traitSet: RelTraitSet,
+ inputRel: RelNode,
+ sink: TableSink[_],
+ sinkName: String)
+ extends Sink(cluster, traitSet, inputRel, sink, sinkName)
+ with DataStreamRel {
+
+ override def copy(traitSet: RelTraitSet, inputs: java.util.List[RelNode]): RelNode = {
+ new DataStreamSink(cluster, traitSet, inputs.get(0), sink, sinkName)
+ }
+
+ override def translateToPlan(planner: StreamPlanner): DataStream[CRow] = {
+ val inputTransform = writeToSink(planner).asInstanceOf[Transformation[CRow]]
+ new DataStream(planner.getExecutionEnvironment, inputTransform)
+ }
+
+ private def writeToSink[T](planner: StreamPlanner): Transformation[_] = {
+ sink match {
+ case retractSink: RetractStreamTableSink[T] =>
+ val resultSink = writeToRetractSink(retractSink, planner)
+ resultSink.getTransformation
+
+ case upsertSink: UpsertStreamTableSink[T] =>
+ val resultSink = writeToUpsertSink(upsertSink, planner)
+ resultSink.getTransformation
+
+ case appendSink: AppendStreamTableSink[T] =>
+ val resultSink = writeToAppendSink(appendSink, planner)
+ resultSink.getTransformation
+
+ case dataStreamTableSink: DataStreamTableSink[_] =>
+ // if no change flags are requested, verify table is an insert-only (append-only) table.
+ if (!dataStreamTableSink.withChangeFlag && !UpdatingPlanChecker.isAppendOnly(getInput)) {
+ throw new ValidationException(
+ "Table is not an append-only table. " +
+ "Use the toRetractStream() in order to handle add and retract messages.")
+ }
+ val dataStream = translateInput(
+ planner,
+ getTableSchema,
+ dataStreamTableSink.getOutputType,
+ dataStreamTableSink.withChangeFlag)
+ dataStream.getTransformation
+
+ case _ =>
+ throw new ValidationException("Stream Tables can only be emitted by AppendStreamTableSink, "
+ + "RetractStreamTableSink, or UpsertStreamTableSink.")
+ }
+ }
+
+ private def writeToRetractSink[T](
+ sink: RetractStreamTableSink[T],
+ planner: StreamPlanner): StreamingDataStreamSink[_]= {
+ // retraction sink can always be used
+ val outputType = TypeConversions.fromDataTypeToLegacyInfo(sink.getConsumedDataType)
+ .asInstanceOf[TypeInformation[JTuple2[JBool, T]]]
+ // translate the Table into a DataStream and provide the type that the TableSink expects.
+ val result: DataStream[JTuple2[JBool, T]] =
+ translateToType(
+ planner,
+ withChangeFlag = true,
+ outputType)
+ // Give the DataStream to the TableSink to emit it.
+ sink.consumeDataStream(result)
+ }
+
+ private def writeToAppendSink[T](
+ sink: AppendStreamTableSink[T],
+ planner: StreamPlanner): StreamingDataStreamSink[_]= {
+ // verify table is an insert-only (append-only) table
+ if (!UpdatingPlanChecker.isAppendOnly(getInput)) {
+ throw new TableException(
+ "AppendStreamTableSink requires that Table has only insert changes.")
+ }
+ val outputType = TypeConversions.fromDataTypeToLegacyInfo(sink.getConsumedDataType)
+ .asInstanceOf[TypeInformation[T]]
+ val resultType = getTableSchema
+ // translate the Table into a DataStream and provide the type that the TableSink expects.
+ val result: DataStream[T] =
+ translateInput(
+ planner,
+ resultType,
+ outputType,
+ withChangeFlag = false)
+ // Give the DataStream to the TableSink to emit it.
+ sink.consumeDataStream(result)
+ }
+
+ private def writeToUpsertSink[T](
+ sink: UpsertStreamTableSink[T],
+ planner: StreamPlanner): StreamingDataStreamSink[_] = {
+ // optimize plan
+ // check for append only table
+ val isAppendOnlyTable = UpdatingPlanChecker.isAppendOnly(getInput)
+ sink.setIsAppendOnly(isAppendOnlyTable)
+ // extract unique key fields
+ val sinkFieldNames = sink.getTableSchema.getFieldNames
+ val tableKeys: Option[Array[String]] = UpdatingPlanChecker
+ .getUniqueKeyFields(getInput, sinkFieldNames)
+ // check that we have keys if the table has changes (is not append-only)
+ tableKeys match {
+ case Some(keys) => sink.setKeyFields(keys)
+ case None if isAppendOnlyTable => sink.setKeyFields(null)
+ case None if !isAppendOnlyTable => throw new TableException(
+ "UpsertStreamTableSink requires that Table has full primary keys if it is updated.")
+ }
+ val outputType = TypeConversions.fromDataTypeToLegacyInfo(sink.getConsumedDataType)
+ .asInstanceOf[TypeInformation[JTuple2[JBool, T]]]
+ val resultType = getTableSchema
+ // translate the Table into a DataStream and provide the type that the TableSink expects.
+ val result: DataStream[JTuple2[JBool, T]] =
+ translateInput(
+ planner,
+ resultType,
+ outputType,
+ withChangeFlag = true)
+ // Give the DataStream to the TableSink to emit it.
+ sink.consumeDataStream(result)
+ }
+
+ private def translateToType[A](
+ planner: StreamPlanner,
+ withChangeFlag: Boolean,
+ tpe: TypeInformation[A]): DataStream[A] = {
+ val rowType = getTableSchema
+
+ // if no change flags are requested, verify table is an insert-only (append-only) table.
+ if (!withChangeFlag && !UpdatingPlanChecker.isAppendOnly(input)) {
+ throw new ValidationException(
+ "Table is not an append-only table. " +
+ "Use the toRetractStream() in order to handle add and retract messages.")
+ }
+
+ // get CRow plan
+ translateInput(planner, rowType, tpe, withChangeFlag)
+ }
+
+ private def translateInput[A](
+ planner: StreamPlanner,
+ logicalSchema: TableSchema,
+ tpe: TypeInformation[A],
+ withChangeFlag: Boolean): DataStream[A] = {
+ val dataStream = getInput().asInstanceOf[DataStreamRel].translateToPlan(planner)
+ DataStreamConversions.convert(dataStream, logicalSchema, withChangeFlag, tpe, planner.getConfig)
+ }
+
+ /**
+ * Returns the record type of the optimized plan with field names of the logical plan.
+ */
+ private def getTableSchema: TableSchema = {
+ val fieldTypes = getInput.getRowType.getFieldList.asScala.map(_.getType)
+ .map(FlinkTypeFactory.toTypeInfo)
+ .map(TypeConversions.fromLegacyInfoToDataType)
+ .toArray
+ TableSchema.builder().fields(sink.getTableSchema.getFieldNames, fieldTypes).build()
+ }
+
+}
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/nodes/logical/FlinkLogicalSink.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/nodes/logical/FlinkLogicalSink.scala
new file mode 100644
index 0000000..2e671ac
--- /dev/null
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/nodes/logical/FlinkLogicalSink.scala
@@ -0,0 +1,73 @@
+/*
+ * 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.plan.nodes.logical
+
+import org.apache.flink.table.plan.nodes.{FlinkConventions, LogicalSink, Sink}
+import org.apache.flink.table.sinks.TableSink
+
+import org.apache.calcite.plan.{Convention, RelOptCluster, RelOptRule, RelTraitSet}
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.convert.ConverterRule
+
+import java.util
+
+import scala.collection.JavaConversions._
+
+/**
+ * Sub-class of [[Sink]] that is a relational expression
+ * which writes out data of input node into a [[TableSink]].
+ */
+class FlinkLogicalSink(
+ cluster: RelOptCluster,
+ traitSet: RelTraitSet,
+ input: RelNode,
+ sink: TableSink[_],
+ sinkName: String)
+ extends Sink(cluster, traitSet, input, sink, sinkName)
+ with FlinkLogicalRel {
+
+ override def copy(traitSet: RelTraitSet, inputs: util.List[RelNode]): RelNode = {
+ new FlinkLogicalSink(cluster, traitSet, inputs.head, sink, sinkName)
+ }
+
+}
+
+private class FlinkLogicalSinkConverter
+ extends ConverterRule(
+ classOf[LogicalSink],
+ Convention.NONE,
+ FlinkConventions.LOGICAL,
+ "FlinkLogicalSinkConverter") {
+
+ override def convert(rel: RelNode): RelNode = {
+ val sink = rel.asInstanceOf[LogicalSink]
+ val newInput = RelOptRule.convert(sink.getInput, FlinkConventions.LOGICAL)
+ FlinkLogicalSink.create(newInput, sink.sink, sink.sinkName)
+ }
+}
+
+object FlinkLogicalSink {
+ val CONVERTER: ConverterRule = new FlinkLogicalSinkConverter()
+
+ def create(input: RelNode, sink: TableSink[_], sinkName: String): FlinkLogicalSink = {
+ val cluster = input.getCluster
+ val traitSet = cluster.traitSetOf(FlinkConventions.LOGICAL).simplify()
+ new FlinkLogicalSink(cluster, traitSet, input, sink, sinkName)
+ }
+}
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/rules/FlinkRuleSets.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/rules/FlinkRuleSets.scala
index 964dd04..a872fb4 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/rules/FlinkRuleSets.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/rules/FlinkRuleSets.scala
@@ -146,7 +146,8 @@ object FlinkRuleSets {
FlinkLogicalTableFunctionScan.CONVERTER,
FlinkLogicalMatch.CONVERTER,
FlinkLogicalTableAggregate.CONVERTER,
- FlinkLogicalWindowTableAggregate.CONVERTER
+ FlinkLogicalWindowTableAggregate.CONVERTER,
+ FlinkLogicalSink.CONVERTER
)
/**
@@ -265,7 +266,8 @@ object FlinkRuleSets {
DataStreamTableAggregateRule.INSTANCE,
DataStreamGroupWindowTableAggregateRule.INSTANCE,
DataStreamPythonCalcRule.INSTANCE,
- DataStreamPythonCorrelateRule.INSTANCE
+ DataStreamPythonCorrelateRule.INSTANCE,
+ DataStreamSinkRule.INSTANCE
)
/**
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/rules/datastream/DataStreamSinkRule.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/rules/datastream/DataStreamSinkRule.scala
new file mode 100644
index 0000000..6d02308
--- /dev/null
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/rules/datastream/DataStreamSinkRule.scala
@@ -0,0 +1,53 @@
+/*
+ * 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.plan.rules.datastream
+
+import org.apache.flink.table.plan.nodes.FlinkConventions
+import org.apache.flink.table.plan.nodes.datastream.DataStreamSink
+import org.apache.flink.table.plan.nodes.logical.FlinkLogicalSink
+
+import org.apache.calcite.plan.{RelOptRule, RelTraitSet}
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.convert.ConverterRule
+
+class DataStreamSinkRule
+ extends ConverterRule(
+ classOf[FlinkLogicalSink],
+ FlinkConventions.LOGICAL,
+ FlinkConventions.DATASTREAM,
+ "DataStreamSinkRule") {
+
+ def convert(rel: RelNode): RelNode = {
+ val sink: FlinkLogicalSink = rel.asInstanceOf[FlinkLogicalSink]
+ val traitSet: RelTraitSet = rel.getTraitSet.replace(FlinkConventions.DATASTREAM)
+ val convInput: RelNode = RelOptRule.convert(sink.getInput(0), FlinkConventions.DATASTREAM)
+
+ new DataStreamSink(
+ rel.getCluster,
+ traitSet,
+ convInput,
+ sink.sink,
+ sink.sinkName
+ )
+ }
+}
+
+object DataStreamSinkRule {
+ val INSTANCE = new DataStreamSinkRule
+}
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/StreamPlanner.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/StreamPlanner.scala
index 659bcfe..f549168 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/StreamPlanner.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/StreamPlanner.scala
@@ -18,10 +18,8 @@
package org.apache.flink.table.planner
import org.apache.flink.annotation.VisibleForTesting
-import org.apache.flink.api.common.typeinfo.TypeInformation
import org.apache.flink.api.dag.Transformation
-import org.apache.flink.api.java.tuple.{Tuple2 => JTuple2}
-import org.apache.flink.streaming.api.datastream.{DataStream, DataStreamSink}
+import org.apache.flink.streaming.api.datastream.DataStream
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment
import org.apache.flink.table.api._
import org.apache.flink.table.calcite._
@@ -34,8 +32,8 @@ import org.apache.flink.table.factories.{TableFactoryUtil, TableSinkFactoryConte
import org.apache.flink.table.operations.OutputConversionModifyOperation.UpdateMode
import org.apache.flink.table.operations._
import org.apache.flink.table.plan.StreamOptimizer
+import org.apache.flink.table.plan.nodes.LogicalSink
import org.apache.flink.table.plan.nodes.datastream.DataStreamRel
-import org.apache.flink.table.plan.util.UpdatingPlanChecker
import org.apache.flink.table.runtime.types.CRow
import org.apache.flink.table.sinks._
import org.apache.flink.table.types.utils.TypeConversions
@@ -46,7 +44,6 @@ import org.apache.calcite.jdbc.CalciteSchemaBuilder.asRootSchema
import org.apache.calcite.plan.RelOptUtil
import org.apache.calcite.rel.RelNode
-import _root_.java.lang.{Boolean => JBool}
import _root_.java.util
import _root_.java.util.Objects
import _root_.java.util.function.{Supplier => JSupplier}
@@ -111,20 +108,55 @@ class StreamPlanner(
override def getParser: Parser = parser
- override def translate(tableOperations: util.List[ModifyOperation])
- : util.List[Transformation[_]] = {
- tableOperations.asScala.map(translate).filter(Objects.nonNull).asJava
+ override def translate(
+ tableOperations: util.List[ModifyOperation]): util.List[Transformation[_]] = {
+ val planner = createDummyPlanner()
+ tableOperations.asScala.map { operation =>
+ val (ast, updatesAsRetraction) = translateToRel(operation)
+ val optimizedPlan = optimizer.optimize(ast, updatesAsRetraction, getRelBuilder)
+ val dataStream = translateToCRow(planner, optimizedPlan)
+ dataStream.getTransformation.asInstanceOf[Transformation[_]]
+ }.filter(Objects.nonNull).asJava
}
override def explain(operations: util.List[Operation], extended: Boolean): String = {
- operations.asScala.map {
+ require(operations.asScala.nonEmpty, "operations should not be empty")
+ val astWithUpdatesAsRetractionTuples = operations.asScala.map {
case queryOperation: QueryOperation =>
- explain(queryOperation)
+ (getRelBuilder.tableOperation(queryOperation).build(), false)
case modifyOperation: ModifyOperation =>
- explain(modifyOperation.getChild)
- case operation =>
- throw new TableException(s"${operation.getClass.getCanonicalName} is not supported")
- }.mkString(s"${System.lineSeparator}${System.lineSeparator}")
+ translateToRel(modifyOperation)
+ case o => throw new TableException(s"Unsupported operation: ${o.getClass.getCanonicalName}")
+ }
+
+ val optimizedNodes = astWithUpdatesAsRetractionTuples.map {
+ case (ast, updatesAsRetraction) =>
+ optimizer.optimize(ast, updatesAsRetraction, getRelBuilder)
+ }
+
+ val planner = createDummyPlanner()
+ val dataStreams = optimizedNodes.map(p => translateToCRow(planner, p))
+
+ val astPlan = astWithUpdatesAsRetractionTuples.map {
+ p => RelOptUtil.toString(p._1)
+ }.mkString(System.lineSeparator)
+ val optimizedPlan = optimizedNodes.map(RelOptUtil.toString).mkString(System.lineSeparator)
+
+ val env = dataStreams.head.getExecutionEnvironment
+ val jsonSqlPlan = env.getExecutionPlan
+ val sqlPlan = PlanJsonParser.getSqlExecutionPlan(jsonSqlPlan, false)
+
+ s"== Abstract Syntax Tree ==" +
+ System.lineSeparator +
+ s"$astPlan" +
+ System.lineSeparator +
+ s"== Optimized Logical Plan ==" +
+ System.lineSeparator +
+ s"$optimizedPlan" +
+ System.lineSeparator +
+ s"== Physical Execution Plan ==" +
+ System.lineSeparator +
+ s"$sqlPlan"
}
override def getCompletionHints(
@@ -135,11 +167,10 @@ class StreamPlanner(
planner.getCompletionHints(statement, position)
}
- private def translate(tableOperation: ModifyOperation)
- : Transformation[_] = {
- tableOperation match {
- case s : UnregisteredSinkModifyOperation[_] =>
- writeToSink(s.getChild, s.getSink)
+ private def translateToRel(modifyOperation: ModifyOperation): (RelNode, Boolean) = {
+ modifyOperation match {
+ case s: UnregisteredSinkModifyOperation[_] =>
+ writeToSink(s.getChild, s.getSink, "UnregisteredSink")
case catalogSink: CatalogSinkModifyOperation =>
getTableSink(catalogSink.getTableIdentifier)
@@ -164,7 +195,10 @@ class StreamPlanner(
s"${classOf[OverwritableTableSink].getSimpleName} but actually got " +
sink.getClass.getName)
}
- writeToSink(catalogSink.getChild, sink)
+ writeToSink(
+ catalogSink.getChild,
+ sink,
+ catalogSink.getTableIdentifier.asSummaryString())
}) match {
case Some(t) => t
case None =>
@@ -178,41 +212,19 @@ class StreamPlanner(
case UpdateMode.UPSERT => (false, true)
}
- translateToType(
- tableOperation.getChild,
- isRetract,
- withChangeFlag,
- TypeConversions.fromDataTypeToLegacyInfo(outputConversion.getType)).getTransformation
+ val tableSink = new DataStreamTableSink(
+ outputConversion.getChild.getTableSchema,
+ TypeConversions.fromDataTypeToLegacyInfo(outputConversion.getType),
+ withChangeFlag)
+ val input = getRelBuilder.tableOperation(modifyOperation.getChild).build()
+ val sink = LogicalSink.create(input, tableSink, "DataStreamTableSink")
+ (sink, isRetract)
case _ =>
- throw new TableException(s"Unsupported ModifyOperation: $tableOperation")
+ throw new TableException(s"Unsupported ModifyOperation: $modifyOperation")
}
}
- private def explain(tableOperation: QueryOperation) = {
- val ast = getRelBuilder.tableOperation(tableOperation).build()
- val optimizedPlan = optimizer
- .optimize(ast, updatesAsRetraction = false, getRelBuilder)
- val dataStream = translateToCRow(optimizedPlan)
-
- val env = dataStream.getExecutionEnvironment
- val jsonSqlPlan = env.getExecutionPlan
-
- val sqlPlan = PlanJsonParser.getSqlExecutionPlan(jsonSqlPlan, false)
-
- s"== Abstract Syntax Tree ==" +
- System.lineSeparator +
- s"${RelOptUtil.toString(ast)}" +
- System.lineSeparator +
- s"== Optimized Logical Plan ==" +
- System.lineSeparator +
- s"${RelOptUtil.toString(optimizedPlan)}" +
- System.lineSeparator +
- s"== Physical Execution Plan ==" +
- System.lineSeparator +
- s"$sqlPlan"
- }
-
private def getFlinkPlanner: FlinkPlannerImpl = {
val currentCatalogName = catalogManager.getCurrentCatalog
val currentDatabase = catalogManager.getCurrentDatabase
@@ -232,9 +244,7 @@ class StreamPlanner(
private[flink] def getExecutionEnvironment: StreamExecutionEnvironment =
executor.asInstanceOf[StreamExecutor].getExecutionEnvironment
- private def translateToCRow(logicalPlan: RelNode): DataStream[CRow] = {
- val planner = createDummyPlanner()
-
+ private def translateToCRow(planner: StreamPlanner, logicalPlan: RelNode): DataStream[CRow] = {
logicalPlan match {
case node: DataStreamRel =>
getExecutionEnvironment.configure(
@@ -249,163 +259,38 @@ class StreamPlanner(
private def writeToSink[T](
tableOperation: QueryOperation,
- sink: TableSink[T])
- : Transformation[_] = {
+ sink: TableSink[T],
+ sinkName: String): (RelNode, Boolean) = {
- val resultSink = sink match {
+ val updatesAsRetraction = sink match {
case retractSink: RetractStreamTableSink[T] =>
retractSink match {
case _: PartitionableTableSink =>
throw new TableException("Partitionable sink in retract stream mode " +
"is not supported yet!")
- case _ =>
+ case _ => // do nothing
}
- writeToRetractSink(retractSink, tableOperation)
+ true
case upsertSink: UpsertStreamTableSink[T] =>
upsertSink match {
case _: PartitionableTableSink =>
throw new TableException("Partitionable sink in upsert stream mode " +
"is not supported yet!")
- case _ =>
+ case _ => // do nothing
}
- writeToUpsertSink(upsertSink, tableOperation)
+ false
- case appendSink: AppendStreamTableSink[T] =>
- writeToAppendSink(appendSink, tableOperation)
+ case _: AppendStreamTableSink[T] =>
+ false
case _ =>
throw new ValidationException("Stream Tables can only be emitted by AppendStreamTableSink, "
+ "RetractStreamTableSink, or UpsertStreamTableSink.")
}
- if (resultSink != null) {
- resultSink.getTransformation
- } else {
- null
- }
- }
-
- private def writeToRetractSink[T](
- sink: RetractStreamTableSink[T],
- tableOperation: QueryOperation)
- : DataStreamSink[_]= {
- // retraction sink can always be used
- val outputType = TypeConversions.fromDataTypeToLegacyInfo(sink.getConsumedDataType)
- .asInstanceOf[TypeInformation[JTuple2[JBool, T]]]
- // translate the Table into a DataStream and provide the type that the TableSink expects.
- val result: DataStream[JTuple2[JBool, T]] =
- translateToType(
- tableOperation,
- updatesAsRetraction = true,
- withChangeFlag = true,
- outputType)
- // Give the DataStream to the TableSink to emit it.
- sink.consumeDataStream(result)
- }
-
- private def writeToAppendSink[T](
- sink: AppendStreamTableSink[T],
- tableOperation: QueryOperation)
- : DataStreamSink[_]= {
- // optimize plan
- val relNode = getRelBuilder.tableOperation(tableOperation).build()
- val optimizedPlan = optimizer.optimize(relNode, updatesAsRetraction = false, getRelBuilder)
- // verify table is an insert-only (append-only) table
- if (!UpdatingPlanChecker.isAppendOnly(optimizedPlan)) {
- throw new TableException(
- "AppendStreamTableSink requires that Table has only insert changes.")
- }
- val outputType = TypeConversions.fromDataTypeToLegacyInfo(sink.getConsumedDataType)
- .asInstanceOf[TypeInformation[T]]
- val resultType = getTableSchema(tableOperation.getTableSchema.getFieldNames, optimizedPlan)
- // translate the Table into a DataStream and provide the type that the TableSink expects.
- val result: DataStream[T] =
- translateOptimized(
- optimizedPlan,
- resultType,
- outputType,
- withChangeFlag = false)
- // Give the DataStream to the TableSink to emit it.
- sink.consumeDataStream(result)
- }
-
- private def writeToUpsertSink[T](
- sink: UpsertStreamTableSink[T],
- tableOperation: QueryOperation)
- : DataStreamSink[_] = {
- // optimize plan
- val relNode = getRelBuilder.tableOperation(tableOperation).build()
- val optimizedPlan = optimizer.optimize(relNode, updatesAsRetraction = false, getRelBuilder)
- // check for append only table
- val isAppendOnlyTable = UpdatingPlanChecker.isAppendOnly(optimizedPlan)
- sink.setIsAppendOnly(isAppendOnlyTable)
- // extract unique key fields
- val sinkFieldNames = sink.getTableSchema.getFieldNames
- val tableKeys: Option[Array[String]] = UpdatingPlanChecker
- .getUniqueKeyFields(optimizedPlan, sinkFieldNames)
- // check that we have keys if the table has changes (is not append-only)
- tableKeys match {
- case Some(keys) => sink.setKeyFields(keys)
- case None if isAppendOnlyTable => sink.setKeyFields(null)
- case None if !isAppendOnlyTable => throw new TableException(
- "UpsertStreamTableSink requires that Table has full primary keys if it is updated.")
- }
- val outputType = TypeConversions.fromDataTypeToLegacyInfo(sink.getConsumedDataType)
- .asInstanceOf[TypeInformation[JTuple2[JBool, T]]]
- val resultType = getTableSchema(tableOperation.getTableSchema.getFieldNames, optimizedPlan)
- // translate the Table into a DataStream and provide the type that the TableSink expects.
- val result: DataStream[JTuple2[JBool, T]] =
- translateOptimized(
- optimizedPlan,
- resultType,
- outputType,
- withChangeFlag = true)
- // Give the DataStream to the TableSink to emit it.
- sink.consumeDataStream(result)
- }
-
- private def translateToType[A](
- table: QueryOperation,
- updatesAsRetraction: Boolean,
- withChangeFlag: Boolean,
- tpe: TypeInformation[A])
- : DataStream[A] = {
- val relNode = getRelBuilder.tableOperation(table).build()
- val dataStreamPlan = optimizer.optimize(relNode, updatesAsRetraction, getRelBuilder)
- val rowType = getTableSchema(table.getTableSchema.getFieldNames, dataStreamPlan)
-
- // if no change flags are requested, verify table is an insert-only (append-only) table.
- if (!withChangeFlag && !UpdatingPlanChecker.isAppendOnly(dataStreamPlan)) {
- throw new ValidationException(
- "Table is not an append-only table. " +
- "Use the toRetractStream() in order to handle add and retract messages.")
- }
-
- // get CRow plan
- translateOptimized(dataStreamPlan, rowType, tpe, withChangeFlag)
- }
-
- private def translateOptimized[A](
- optimizedPlan: RelNode,
- logicalSchema: TableSchema,
- tpe: TypeInformation[A],
- withChangeFlag: Boolean)
- : DataStream[A] = {
- val dataStream = translateToCRow(optimizedPlan)
- DataStreamConversions.convert(dataStream, logicalSchema, withChangeFlag, tpe, config)
- }
-
- /**
- * Returns the record type of the optimized plan with field names of the logical plan.
- */
- private def getTableSchema(originalNames: Array[String], optimizedPlan: RelNode): TableSchema = {
- val fieldTypes = optimizedPlan.getRowType.getFieldList.asScala.map(_.getType)
- .map(FlinkTypeFactory.toTypeInfo)
- .map(TypeConversions.fromLegacyInfoToDataType)
- .toArray
-
- TableSchema.builder().fields(originalNames, fieldTypes).build()
+ val input = getRelBuilder.tableOperation(tableOperation).build()
+ (LogicalSink.create(input, sink, sinkName), updatesAsRetraction)
}
private def getTableSink(objectIdentifier: ObjectIdentifier): Option[TableSink[_]] = {
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/sinks/DataStreamTableSink.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/sinks/DataStreamTableSink.scala
new file mode 100644
index 0000000..ba2b132
--- /dev/null
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/sinks/DataStreamTableSink.scala
@@ -0,0 +1,58 @@
+/*
+ * 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.sinks
+
+import org.apache.flink.annotation.Internal
+import org.apache.flink.api.common.typeinfo.TypeInformation
+import org.apache.flink.streaming.api.datastream.DataStream
+import org.apache.flink.table.api.{TableException, TableSchema}
+import org.apache.flink.table.operations.OutputConversionModifyOperation
+
+/**
+ * A special [[TableSink]] that represents information of [[OutputConversionModifyOperation]].
+ *
+ * @param outputType The [[TypeInformation]] that specifies the type of the [[DataStream]].
+ * @param withChangeFlag Set to true to emit records with change flags.
+ * @tparam T The type of the resulting [[DataStream]].
+ */
+@Internal
+class DataStreamTableSink[T](
+ tableSchema: TableSchema,
+ outputType: TypeInformation[T],
+ val withChangeFlag: Boolean)
+ extends TableSink[T] {
+
+ /**
+ * Return the type expected by this [[TableSink]].
+ *
+ * This type should depend on the types returned by [[getTableSchema]].
+ *
+ * @return The type expected by this [[TableSink]].
+ */
+ override def getOutputType: TypeInformation[T] = outputType
+
+ override def getTableSchema: TableSchema = tableSchema
+
+ override def configure(
+ fieldNames: Array[String],
+ fieldTypes: Array[TypeInformation[_]]): TableSink[T] = {
+ throw new TableException(s"configure is not supported.")
+ }
+
+}
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/TableEnvironmentITCase.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/TableEnvironmentITCase.scala
index 6a43957..c5e90bf 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/TableEnvironmentITCase.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/TableEnvironmentITCase.scala
@@ -221,9 +221,7 @@ class TableEnvironmentITCase(tableEnvName: String) {
}
val streamEnv = ScalaStreamExecutionEnvironment.getExecutionEnvironment
val streamTableEnv = ScalaStreamTableEnvironment.create(streamEnv, settings)
- val t = streamEnv.fromCollection(getPersonData)
- .toTable(streamTableEnv, 'first, 'id, 'score, 'last)
- streamTableEnv.registerTable("MyTable", t)
+ streamTableEnv.registerTableSource("MyTable", getPersonCsvTableSource)
val sink1Path = registerCsvTableSink(streamTableEnv, Array("first"), Array(STRING), "MySink1")
checkEmptyFile(sink1Path)
StreamITCase.clear
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/stream/ExplainTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/stream/ExplainTest.scala
index 62f6da3..14ba09b 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/stream/ExplainTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/stream/ExplainTest.scala
@@ -18,11 +18,14 @@
package org.apache.flink.table.api.stream
+import org.apache.flink.api.common.typeinfo.TypeInformation
import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
-import org.apache.flink.table.api.{EnvironmentSettings, Table}
import org.apache.flink.table.api.scala._
-import org.apache.flink.table.utils.TableTestUtil.streamTableNode
+import org.apache.flink.table.api.{EnvironmentSettings, Table, Types}
+import org.apache.flink.table.runtime.utils.CommonTestData
+import org.apache.flink.table.utils.MemoryTableSourceSinkUtil
+import org.apache.flink.table.utils.TableTestUtil.{readFromResource, replaceStageId, streamTableNode}
import org.apache.flink.test.util.AbstractTestBase
import org.junit.Assert.assertEquals
@@ -30,8 +33,6 @@ import org.junit._
class ExplainTest extends AbstractTestBase {
- private val testFilePath = this.getClass.getResource("/").getFile
-
@Test
def testFilter(): Unit = {
val env = StreamExecutionEnvironment.getExecutionEnvironment
@@ -41,10 +42,9 @@ class ExplainTest extends AbstractTestBase {
val scan = env.fromElements((1, "hello")).toTable(tEnv, 'a, 'b)
val table = scan.filter($"a" % 2 === 0)
- val result = replaceString(tEnv.explain(table))
+ val result = replaceStageId(tEnv.explain(table))
- val source = scala.io.Source.fromFile(testFilePath +
- "../../src/test/scala/resources/testFilterStream0.out").mkString
+ val source = readFromResource("testFilterStream0.out")
val expect = replaceString(source, scan)
assertEquals(expect, result)
}
@@ -59,34 +59,69 @@ class ExplainTest extends AbstractTestBase {
val table2 = env.fromElements((1, "hello")).toTable(tEnv, 'count, 'word)
val table = table1.unionAll(table2)
- val result = replaceString(tEnv.explain(table))
+ val result = replaceStageId(tEnv.explain(table))
- val source = scala.io.Source.fromFile(testFilePath +
- "../../src/test/scala/resources/testUnionStream0.out").mkString
+ val source = readFromResource("testUnionStream0.out")
val expect = replaceString(source, table1, table2)
assertEquals(expect, result)
}
+ @Test
+ def testInsert(): Unit = {
+ val env = StreamExecutionEnvironment.getExecutionEnvironment
+ val settings = EnvironmentSettings.newInstance().useOldPlanner().build()
+ val tEnv = StreamTableEnvironment.create(env, settings)
+
+ tEnv.registerTableSource("sourceTable", CommonTestData.getCsvTableSource)
+
+ val fieldNames = Array("d", "e")
+ val fieldTypes: Array[TypeInformation[_]] = Array(Types.STRING(), Types.INT())
+ val sink = new MemoryTableSourceSinkUtil.UnsafeMemoryAppendTableSink
+ tEnv.registerTableSink("targetTable", sink.configure(fieldNames, fieldTypes))
+
+ tEnv.sqlUpdate("INSERT INTO targetTable SELECT first, id FROM sourceTable")
+
+ val result = tEnv.explain(false)
+ val source = readFromResource("testInsert.out")
+ assertEquals(replaceStageId(source), replaceStageId(result))
+ }
+
+ @Test
+ def testMultipleInserts(): Unit = {
+ val env = StreamExecutionEnvironment.getExecutionEnvironment
+ val settings = EnvironmentSettings.newInstance().useOldPlanner().build()
+ val tEnv = StreamTableEnvironment.create(env, settings)
+
+ tEnv.registerTableSource("sourceTable", CommonTestData.getCsvTableSource)
+
+ val fieldNames = Array("d", "e")
+ val fieldTypes: Array[TypeInformation[_]] = Array(Types.STRING(), Types.INT())
+ val sink = new MemoryTableSourceSinkUtil.UnsafeMemoryAppendTableSink
+ tEnv.registerTableSink("targetTable1", sink.configure(fieldNames, fieldTypes))
+ tEnv.registerTableSink("targetTable2", sink.configure(fieldNames, fieldTypes))
+
+ tEnv.sqlUpdate("INSERT INTO targetTable1 SELECT first, id FROM sourceTable")
+ tEnv.sqlUpdate("INSERT INTO targetTable2 SELECT last, id FROM sourceTable")
+
+ val result = tEnv.explain(false)
+ val source = readFromResource("testMultipleInserts.out")
+ assertEquals(replaceStageId(source), replaceStageId(result))
+ }
+
def replaceString(s: String, t1: Table, t2: Table): String = {
- replaceSourceNode(replaceSourceNode(replaceString(s), t1, 0), t2, 1)
+ replaceSourceNode(replaceSourceNode(replaceStageId(s), t1, 0), t2, 1)
}
def replaceString(s: String, t: Table): String = {
- replaceSourceNode(replaceString(s), t, 0)
+ replaceSourceNode(replaceStageId(s), t, 0)
}
- private def replaceSourceNode(s: String, t: Table, idx: Int) = {
- replaceString(s)
+ private def replaceSourceNode(s: String, t: Table, idx: Int): String = {
+ replaceStageId(s)
.replace(
s"%logicalSourceNode$idx%", streamTableNode(t)
.replace("DataStreamScan", "FlinkLogicalDataStreamScan"))
.replace(s"%sourceNode$idx%", streamTableNode(t))
}
- def replaceString(s: String) = {
- /* Stage {id} is ignored, because id keeps incrementing in test class
- * while StreamExecutionEnvironment is up
- */
- s.replaceAll("\\r\\n", "\n").replaceAll("Stage \\d+", "")
- }
}
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/utils/TableTestBase.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/utils/TableTestBase.scala
index e2b916a..66b9c7f 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/utils/TableTestBase.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/utils/TableTestBase.scala
@@ -46,6 +46,7 @@ import org.junit.{ComparisonFailure, Rule}
import org.mockito.Mockito.{mock, when}
import _root_.scala.util.control.Breaks._
+import scala.io.Source
/**
* Test base for testing Table API / SQL plans.
@@ -214,6 +215,15 @@ object TableTestUtil {
s"DataStreamScan(id=[$id], fields=[${fieldNames.mkString(", ")}])"
}
+
+ def readFromResource(file: String): String = {
+ val source = s"${getClass.getResource("/").getFile}../../src/test/scala/resources/$file"
+ Source.fromFile(source).mkString
+ }
+
+ def replaceStageId(s: String): String = {
+ s.replaceAll("\\r\\n", "\n").replaceAll("Stage \\d+", "")
+ }
}
case class BatchTableTestUtil(
diff --git a/flink-table/flink-table-planner/src/test/scala/resources/testFromToDataStreamAndSqlUpdate.out b/flink-table/flink-table-planner/src/test/scala/resources/testFromToDataStreamAndSqlUpdate.out
index 9cc6251..76c0638 100644
--- a/flink-table/flink-table-planner/src/test/scala/resources/testFromToDataStreamAndSqlUpdate.out
+++ b/flink-table/flink-table-planner/src/test/scala/resources/testFromToDataStreamAndSqlUpdate.out
@@ -1,21 +1,22 @@
== Abstract Syntax Tree ==
-LogicalProject(first=[$0])
- FlinkLogicalDataStreamScan(fields=[first, id, score, last])
+LogicalSink(name=[default_catalog.default_database.MySink1], fields=[first])
+ LogicalProject(first=[$0])
+ LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
== Optimized Logical Plan ==
-DataStreamCalc(select=[first])
- DataStreamScan(fields=[first, id, score, last])
+DataStreamSink(name=[default_catalog.default_database.MySink1], fields=[first])
+ StreamTableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[first], source=[CsvTableSource(read fields: first)])
== Physical Execution Plan ==
: Data Source
content : collect elements with CollectionInputFormat
: Operator
- content : from: (first, id, score, last)
+ content : CsvTableSource(read fields: first)
ship_strategy : REBALANCE
: Operator
- content : where: (>(id, 0)), select: (last)
+ content : Map
ship_strategy : FORWARD
: Operator
@@ -23,14 +24,10 @@ DataStreamCalc(select=[first])
ship_strategy : FORWARD
: Operator
- content : from: (first, id, score, last)
+ content : Map
ship_strategy : REBALANCE
- : Operator
- content : select: (first)
+ : Data Sink
+ content : Sink: CsvTableSink(first)
ship_strategy : FORWARD
- : Data Sink
- content : Sink: Unnamed
- ship_strategy : FORWARD
-
diff --git a/flink-table/flink-table-planner/src/test/scala/resources/testInsert.out b/flink-table/flink-table-planner/src/test/scala/resources/testInsert.out
new file mode 100644
index 0000000..bddbee2
--- /dev/null
+++ b/flink-table/flink-table-planner/src/test/scala/resources/testInsert.out
@@ -0,0 +1,29 @@
+== Abstract Syntax Tree ==
+LogicalSink(name=[default_catalog.default_database.targetTable], fields=[d, e])
+ LogicalProject(first=[$0], id=[$1])
+ LogicalTableScan(table=[[default_catalog, default_database, sourceTable]])
+
+== Optimized Logical Plan ==
+DataStreamSink(name=[default_catalog.default_database.targetTable], fields=[d, e])
+ StreamTableSourceScan(table=[[default_catalog, default_database, sourceTable]], fields=[first, id], source=[CsvTableSource(read fields: first, id)])
+
+== Physical Execution Plan ==
+ : Data Source
+ content : collect elements with CollectionInputFormat
+
+ : Operator
+ content : CsvTableSource(read fields: first, id)
+ ship_strategy : REBALANCE
+
+ : Operator
+ content : Map
+ ship_strategy : FORWARD
+
+ : Operator
+ content : to: Row
+ ship_strategy : FORWARD
+
+ : Data Sink
+ content : Sink: UnsafeMemoryAppendTableSink(d, e)
+ ship_strategy : FORWARD
+
diff --git a/flink-table/flink-table-planner/src/test/scala/resources/testMultipleInserts.out b/flink-table/flink-table-planner/src/test/scala/resources/testMultipleInserts.out
new file mode 100644
index 0000000..0bb4d0f
--- /dev/null
+++ b/flink-table/flink-table-planner/src/test/scala/resources/testMultipleInserts.out
@@ -0,0 +1,55 @@
+== Abstract Syntax Tree ==
+LogicalSink(name=[default_catalog.default_database.targetTable1], fields=[d, e])
+ LogicalProject(first=[$0], id=[$1])
+ LogicalTableScan(table=[[default_catalog, default_database, sourceTable]])
+
+LogicalSink(name=[default_catalog.default_database.targetTable2], fields=[d, e])
+ LogicalProject(last=[$3], id=[$1])
+ LogicalTableScan(table=[[default_catalog, default_database, sourceTable]])
+
+== Optimized Logical Plan ==
+DataStreamSink(name=[default_catalog.default_database.targetTable1], fields=[d, e])
+ StreamTableSourceScan(table=[[default_catalog, default_database, sourceTable]], fields=[first, id], source=[CsvTableSource(read fields: first, id)])
+
+DataStreamSink(name=[default_catalog.default_database.targetTable2], fields=[d, e])
+ StreamTableSourceScan(table=[[default_catalog, default_database, sourceTable]], fields=[last, id], source=[CsvTableSource(read fields: last, id)])
+
+== Physical Execution Plan ==
+ : Data Source
+ content : collect elements with CollectionInputFormat
+
+ : Operator
+ content : CsvTableSource(read fields: first, id)
+ ship_strategy : REBALANCE
+
+ : Operator
+ content : Map
+ ship_strategy : FORWARD
+
+ : Operator
+ content : to: Row
+ ship_strategy : FORWARD
+
+ : Data Source
+ content : collect elements with CollectionInputFormat
+
+ : Operator
+ content : CsvTableSource(read fields: last, id)
+ ship_strategy : REBALANCE
+
+ : Operator
+ content : Map
+ ship_strategy : FORWARD
+
+ : Operator
+ content : to: Row
+ ship_strategy : FORWARD
+
+ : Data Sink
+ content : Sink: UnsafeMemoryAppendTableSink(d, e)
+ ship_strategy : FORWARD
+
+ : Data Sink
+ content : Sink: UnsafeMemoryAppendTableSink(d, e)
+ ship_strategy : FORWARD
+
diff --git a/flink-table/flink-table-planner/src/test/scala/resources/testSqlUpdateAndToDataStream.out b/flink-table/flink-table-planner/src/test/scala/resources/testSqlUpdateAndToDataStream.out
index 40fce87..76c0638 100644
--- a/flink-table/flink-table-planner/src/test/scala/resources/testSqlUpdateAndToDataStream.out
+++ b/flink-table/flink-table-planner/src/test/scala/resources/testSqlUpdateAndToDataStream.out
@@ -1,9 +1,11 @@
== Abstract Syntax Tree ==
-LogicalProject(first=[$0])
- LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
+LogicalSink(name=[default_catalog.default_database.MySink1], fields=[first])
+ LogicalProject(first=[$0])
+ LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
== Optimized Logical Plan ==
-StreamTableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[first], source=[CsvTableSource(read fields: first)])
+DataStreamSink(name=[default_catalog.default_database.MySink1], fields=[first])
+ StreamTableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[first], source=[CsvTableSource(read fields: first)])
== Physical Execution Plan ==
: Data Source
@@ -17,3 +19,15 @@ StreamTableSourceScan(table=[[default_catalog, default_database, MyTable]], fiel
content : Map
ship_strategy : FORWARD
+ : Operator
+ content : to: Row
+ ship_strategy : FORWARD
+
+ : Operator
+ content : Map
+ ship_strategy : REBALANCE
+
+ : Data Sink
+ content : Sink: CsvTableSink(first)
+ ship_strategy : FORWARD
+