KurtYoung commented on a change in pull request #8035:
[FLINK-11975][table-planner-blink] Support to run a sql query : 'SELECT * FROM
(VALUES (1, 2, 3)) T(a, b, c)'
URL: https://github.com/apache/flink/pull/8035#discussion_r268977642
##########
File path:
flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/nodes/physical/stream/StreamExecSink.scala
##########
@@ -53,5 +67,136 @@ class StreamExecSink[T](
new StreamExecSink(cluster, traitSet, inputs.get(0), sink, sinkName)
}
+ override def getInputNodes: util.List[ExecNode[StreamTableEnvironment, _]] =
{
+ List(getInput.asInstanceOf[ExecNode[StreamTableEnvironment, _]])
+ }
+
+ /**
+ * Internal method, translates this node into a Flink operator.
+ *
+ * @param tableEnv The [[StreamTransformation]] of the translated Table.
+ */
+ override protected def translateToPlanInternal(
+ tableEnv: StreamTableEnvironment): StreamTransformation[Any] = {
+ val convertTransformation = sink match {
+
+ case _: RetractStreamTableSink[T] =>
+ // translate the Table into a DataStream and provide the type that the
TableSink expects.
+ translateToStreamTransformation(withChangeFlag = true, tableEnv)
+
+ case upsertSink: UpsertStreamTableSink[T] =>
+ // check for append only table
+ val isAppendOnlyTable = UpdatingPlanChecker.isAppendOnly(this)
+ upsertSink.setIsAppendOnly(isAppendOnlyTable)
+ // translate the Table into a DataStream and provide the type that the
TableSink expects.
+ translateToStreamTransformation(withChangeFlag = true, tableEnv)
+
+ case _: AppendStreamTableSink[T] =>
+ // verify table is an insert-only (append-only) table
+ if (!UpdatingPlanChecker.isAppendOnly(this)) {
+ throw new TableException(
+ "AppendStreamTableSink requires that Table has only insert
changes.")
+ }
+
+ val accMode =
this.getTraitSet.getTrait(AccModeTraitDef.INSTANCE).getAccMode
+ if (accMode == AccMode.AccRetract) {
+ throw new TableException(
+ "AppendStreamTableSink can not be used to output retraction
messages.")
+ }
+
+ // translate the Table into a DataStream and provide the type that the
TableSink expects.
+ translateToStreamTransformation(withChangeFlag = false, tableEnv)
+
+ case s: DataStreamTableSink[_] =>
+ // In case of table to stream through
BatchTableEnvironment#translateToDataStream,
+ // 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 StreamTransformation is ok.
+ return translateToStreamTransformation(s.withChangeFlag, tableEnv)
+ .asInstanceOf[StreamTransformation[Any]]
+ case _ =>
+ throw new TableException("Stream Tables can only be emitted by
AppendStreamTableSink, " +
+ "RetractStreamTableSink, or
UpsertStreamTableSink.")
+ }
+ val stream = new DataStream(tableEnv.execEnv, convertTransformation)
+ val resultTransformation = emitDataStream(tableEnv.getConfig.getConf,
stream).getTransformation
+ resultTransformation.asInstanceOf[StreamTransformation[Any]]
+ }
+
+ /**
+ * Translates a logical [[RelNode]] into a [[StreamTransformation]].
+ *
+ * @param withChangeFlag Set to true to emit records with change flags.
+ * @return The [[StreamTransformation]] that corresponds to the translated
[[Table]].
+ */
+ private def translateToStreamTransformation(
+ withChangeFlag: Boolean,
+ tableEnv: StreamTableEnvironment): StreamTransformation[T] = {
+ val inputNode = getInput
+ // if no change flags are requested, verify table is an insert-only
(append-only) table.
+ if (!withChangeFlag && !UpdatingPlanChecker.isAppendOnly(inputNode)) {
+ throw new TableException(
+ "Table is not an append-only table. " +
+ "Use the toRetractStream() in order to handle add and retract
messages.")
+ }
+
+ // get BaseRow plan
+ val parTransformation = inputNode match {
+ // Sink's input must be StreamExecNode[BaseRow] now.
+ case node: StreamExecNode[BaseRow] =>
+ node.translateToPlan(tableEnv)
+ 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 logicalType = inputNode.getRowType
+ val rowtimeFields = logicalType.getFieldList
+ .filter(f =>
FlinkTypeFactory.isRowtimeIndicatorType(f.getType))
+
+ if (rowtimeFields.size > 1) {
+ throw new TableException(
+ s"Found more than one rowtime field:
[${rowtimeFields.map(_.getName).mkString(", ")}] in " +
+ s"the table that should be converted to a DataStream.\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 resultType = sink.getOutputType
+ // TODO support SinkConversion after FLINK-11974 is done
+ val typeClass = extractTableSinkTypeClass(sink)
+ if (CodeGenUtils.isInternalClass(typeClass, resultType)) {
+ parTransformation.asInstanceOf[StreamTransformation[T]]
+ } else {
+ throw new TableException(s"Not support SinkConvention now.")
+ }
+
+ }
+
+ /**
+ * emit [[DataStream]].
+ *
+ * @param dataStream The [[DataStream]] to emit.
+ */
+ private def emitDataStream(
+ tableConf: Configuration,
+ dataStream: DataStream[T]) : DataStreamSink[_] = {
+ sink match {
+
+ case retractSink: RetractStreamTableSink[T] =>
Review comment:
these 3 cases can be combined into `StreamTableSink`
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
