wuchong commented on a change in pull request #8203:
[Flink-12208][table-planner-blink] Support translation from StreamExecSort /
TemporalSort / SortLimit/ Limit to StreamTransformation.
URL: https://github.com/apache/flink/pull/8203#discussion_r277503077
##########
File path:
flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/nodes/physical/stream/StreamExecTemporalSort.scala
##########
@@ -63,4 +77,107 @@ class StreamExecTemporalSort(
.item("orderBy", RelExplainUtil.collationToString(sortCollation,
getRowType))
}
+ //~ ExecNode methods
-----------------------------------------------------------
+
+ /**
+ * Returns an array of this node's inputs. If there are no inputs,
+ * returns an empty list, not null.
+ *
+ * @return Array of this node's inputs
+ */
+ override def getInputNodes: util.List[ExecNode[StreamTableEnvironment, _]] =
{
+ List(getInput.asInstanceOf[ExecNode[StreamTableEnvironment, _]])
+ }
+
+ /**
+ * Internal method, translates this node into a Flink operator.
+ *
+ * @param tableEnv The [[StreamTableEnvironment]] of the translated Table.
+ */
+ override protected def translateToPlanInternal(
+ tableEnv: StreamTableEnvironment): StreamTransformation[BaseRow] = {
+ // time ordering needs to be ascending
+ if (SortUtil.getFirstSortDirection(sortCollation) != Direction.ASCENDING) {
+ throw new TableException(
+ "Sort: Primary sort order of a streaming table must be ascending on
time.\n" +
+ "please re-check sort statement according to the description above")
+ }
+
+ val input = getInputNodes.get(0).translateToPlan(tableEnv)
+ .asInstanceOf[StreamTransformation[BaseRow]]
+
+ val timeType = SortUtil.getFirstSortField(sortCollation,
getRowType).getType
+ timeType match {
+ case _ if FlinkTypeFactory.isProctimeIndicatorType(timeType) =>
+ createSortProcTime(input, tableEnv.getConfig)
+ case _ if FlinkTypeFactory.isRowtimeIndicatorType(timeType) =>
+ createSortRowTime(input, tableEnv.getConfig)
+ case _ =>
+ throw new TableException(
+ "Sort: Internal Error\n" +
+ "Normally, this happens unlikely. please contact customer support
for this"
+ )
+ }
+ }
+
+ /**
+ * Create Sort logic based on processing time
+ */
+ private def createSortProcTime(
+ input: StreamTransformation[BaseRow],
+ tableConfig: TableConfig): StreamTransformation[BaseRow] = {
+ val inputType = FlinkTypeFactory.toInternalRowType(getInput.getRowType)
+ val fieldCollations = sortCollation.getFieldCollations
+ // if the order has secondary sorting fields in addition to the proctime
+ if (fieldCollations.size() > 1) {
+ // strip off time collation
+ val (keys, orders, nullsIsLast) =
SortUtil.getKeysAndOrders(fieldCollations.tail)
+ // sort code gen
+ val keyTypes = keys.map(inputType.getTypeAt)
+ val rowComparator = ComparatorCodeGenerator.gen(tableConfig,
"ProcTimeSortComparator",
+ keys, keyTypes, orders, nullsIsLast)
+ val sortOperator = new ProcTimeSortOperator(inputType.toTypeInfo,
rowComparator)
+ val outputRowTypeInfo =
FlinkTypeFactory.toInternalRowType(getRowType).toTypeInfo
+ val ret = new OneInputTransformation(
+ input, "ProcTimeSortOperator", sortOperator, outputRowTypeInfo,
input.getParallelism)
+ val selector = NullBinaryRowKeySelector.INSTANCE
+ ret.setStateKeySelector(selector)
+ ret.setStateKeyType(selector.getProducedType)
+ ret
+ } else {
+ // if the order is done only on proctime we only need to forward the
elements
+ input
+ }
+ }
+
+ /**
+ * Create Sort logic based on row time
+ */
+ private def createSortRowTime(
+ input: StreamTransformation[BaseRow],
+ tableConfig: TableConfig): StreamTransformation[BaseRow] = {
+ val fieldCollations = sortCollation.getFieldCollations
+ val rowTimeIdx = fieldCollations.get(0).getFieldIndex
+ val inputType = FlinkTypeFactory.toInternalRowType(getInput.getRowType)
+ val rowComparator = if (fieldCollations.size() > 1) {
+ // strip off time collation
+ val (keys, orders, nullsIsLast) =
SortUtil.getKeysAndOrders(fieldCollations.tail)
+ // comparator code gen
+ val keyTypes = keys.map(inputType.getTypeAt)
+ ComparatorCodeGenerator.gen(tableConfig, "RowTimeSortComparator", keys,
keyTypes, orders,
+ nullsIsLast)
+ } else {
+ null
+ }
+ val sortOperator = new RowTimeSortOperator(inputType.toTypeInfo,
rowTimeIdx, rowComparator)
+ val outputRowTypeInfo =
FlinkTypeFactory.toInternalRowType(getRowType).toTypeInfo
+
+ // sets parallelism to 1 since StreamExecTemporalSort could only work in
global mode.
+ val ret = new OneInputTransformation(
+ input, "RowTimeSortOperator", sortOperator, outputRowTypeInfo, 1)
Review comment:
Separate into multi-lines?
Force maxParallelism to `1` too?
`ret.setMaxParallelism(1)`
----------------------------------------------------------------
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
