From 3318891bc72d3a3e40c040503a19961ea1c4c819 Mon Sep 17 00:00:00 2001 From: rtudoran Date: Fri, 7 Apr 2017 15:44:57 +0200 Subject: [PATCH] Backbone implementation for supporting sort. Implementation includes util functions, sort aggregations (without retraction), rule and sort cases identification --- .../aggfunctions/SortAggFunction.scala | 166 ++++++++ .../nodes/datastream/DataStreamSort.scala | 167 ++++++++ .../table/plan/rules/FlinkRuleSets.scala | 4 +- .../rules/datastream/DataStreamSortRule.scala | 67 ++++ ...ProcTimeUnboundedSortProcessFunction.scala | 139 +++++++ .../table/runtime/aggregate/SortUtil.scala | 367 ++++++++++++++++++ .../api/scala/stream/sql/SqlITCase.scala | 100 +++++ 7 files changed, 1009 insertions(+), 1 deletion(-) create mode 100644 flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/aggfunctions/SortAggFunction.scala create mode 100644 flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamSort.scala create mode 100644 flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/rules/datastream/DataStreamSortRule.scala create mode 100644 flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/ProcTimeUnboundedSortProcessFunction.scala create mode 100644 flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/SortUtil.scala diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/aggfunctions/SortAggFunction.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/aggfunctions/SortAggFunction.scala new file mode 100644 index 00000000000000..0f863aed0fd39d --- /dev/null +++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/aggfunctions/SortAggFunction.scala @@ -0,0 +1,166 @@ +/* + * 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.functions.aggfunctions + +import java.math.BigDecimal +import java.util.{List => JList} + +import org.apache.flink.api.common.typeinfo.{BasicTypeInfo, TypeInformation} +import org.apache.flink.api.java.tuple.{Tuple2 => JTuple2} +import org.apache.flink.api.java.typeutils.TupleTypeInfo +import org.apache.flink.table.functions.{Accumulator, AggregateFunction} +import scala.collection.mutable.ArrayBuffer +import org.apache.flink.types.Row +import org.apache.flink.table.runtime.aggregate.UntypedOrdering +import org.apache.flink.table.runtime.aggregate.MultiOutputAggregateFunction +import java.util.{ List => JList,ArrayList } +import org.apache.flink.api.java.typeutils.RowTypeInfo +import org.apache.flink.api.java.typeutils.ListTypeInfo +import org.apache.calcite.rel.RelFieldCollation.Direction + +/** The initial accumulator for Sort aggregate function */ +class SortAccumulator extends ArrayBuffer[JTuple2[Row,Row]] with Accumulator with Serializable + +/** + * Base class for built-in Min aggregate function + * + * @tparam K the type for the key sort type + * @tparam T the type for the aggregation result + */ +abstract class SortAggFunction[K,T]( + val keyIndexes: Array[Int], + val keySortDirections: Array[Direction], + val orderings: Array[UntypedOrdering]) extends MultiOutputAggregateFunction[T] { + + override def createAccumulator(): Accumulator = { + val acc = new SortAccumulator + + acc + } + + override def accumulate(accumulator: Accumulator, value: Any): Unit = { + if (value != null) { + val v = value.asInstanceOf[Row] + val acc = accumulator.asInstanceOf[SortAccumulator] + + var i = 0 + //create the (compose) key of the new value + val keyV = new Row(keyIndexes.size) + while (i i += 1 //same key and need to sort on consequent keys + case g if g > 0 => { + acc.insert(j, new JTuple2(keyV,v)) //add new element in place + //end loop and ensure that the element is not added again + j = acc.size + 1 + i = keyIndexes.size + } + case _ => i = keyIndexes.size // continue with the next element + } + } + j += 1 + } + + //condition for the case when the element needs to be inserted at the end as greatest element + if (j==acc.size) { + acc.insert(j, new JTuple2(keyV,v)) + } + + } + } + + override def getValue(accumulator: Accumulator): T = { + null.asInstanceOf[T] + } + + override def getValues(accumulator: Accumulator): JList[T] = { + val acc = accumulator.asInstanceOf[SortAccumulator] + val res = new ArrayList[T](acc.size) + + var i=0 + while (i< acc.size) { + res.add(acc(i).f1.asInstanceOf[T]) + i += 1 + } + + res + } + + override def merge(accumulators: JList[Accumulator]): Accumulator = { + val ret = accumulators.get(0) + var i: Int = 1 + var j = 0 + while (i < accumulators.size()) { + val a = accumulators.get(i).asInstanceOf[SortAccumulator] + j = 0 + while (j < a.size) { + accumulate(ret.asInstanceOf[SortAccumulator], a(j).f1) + j += 1 + } + i += 1 + } + ret + } + + override def resetAccumulator(accumulator: Accumulator): Unit = { + accumulator.asInstanceOf[SortAccumulator].clear() + } + + override def getAccumulatorType(): TypeInformation[_] = { + + new ListTypeInfo( + new TupleTypeInfo(new JTuple2[Row,Row].getClass, + getValueTypeInfo, + getSortKeyTypeInfo) + ) + + } + + + def getValueTypeInfo: TypeInformation[_] + def getSortKeyTypeInfo: TypeInformation[_] +} + +/** + * Built-in Row sort aggregate function + */ +class RowSortAggFunction(keyIndexes: Array[Int], + keySortDirections: Array[Direction], + orderings: Array[UntypedOrdering], + rowType: RowTypeInfo, + sortKeyType:RowTypeInfo) + extends SortAggFunction[Row,Row](keyIndexes,keySortDirections,orderings) { + override def getValueTypeInfo = rowType + override def getSortKeyTypeInfo = sortKeyType +} + diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamSort.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamSort.scala new file mode 100644 index 00000000000000..98d9feeae2855d --- /dev/null +++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamSort.scala @@ -0,0 +1,167 @@ +/* + * 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.calcite.plan.{ RelOptCluster, RelTraitSet } +import org.apache.calcite.rel.`type`.RelDataType +import org.apache.calcite.rel.core.AggregateCall +import org.apache.calcite.rel.{ RelNode, RelWriter, SingleRel } +import org.apache.flink.api.java.tuple.Tuple +import org.apache.flink.streaming.api.datastream.{ AllWindowedStream, DataStream, KeyedStream, WindowedStream } +import org.apache.flink.streaming.api.windowing.assigners._ +import org.apache.flink.streaming.api.windowing.time.Time +import org.apache.flink.streaming.api.windowing.windows.{ Window => DataStreamWindow } +import org.apache.flink.table.api.StreamTableEnvironment +import org.apache.flink.table.calcite.FlinkRelBuilder.NamedWindowProperty +import org.apache.flink.table.calcite.FlinkTypeFactory +import org.apache.flink.table.expressions._ +import org.apache.flink.table.plan.logical._ +import org.apache.flink.table.plan.nodes.CommonAggregate +import org.apache.flink.table.plan.nodes.datastream.DataStreamAggregate._ +import org.apache.flink.table.runtime.aggregate.AggregateUtil._ +import org.apache.flink.table.runtime.aggregate._ +import org.apache.flink.table.typeutils.TypeCheckUtils.isTimeInterval +import org.apache.flink.table.typeutils.{ RowIntervalTypeInfo, TimeIntervalTypeInfo } +import org.apache.flink.types.Row +import org.apache.calcite.rel.logical.LogicalSort +import org.apache.calcite.sql.SqlAggFunction +import org.apache.flink.table.plan.nodes.datastream.DataStreamRel +import org.apache.flink.table.api.TableException +import org.apache.calcite.sql.fun.SqlSingleValueAggFunction +import org.apache.flink.api.common.functions.RichMapFunction +import org.apache.flink.api.common.functions.RichFlatMapFunction +import org.apache.flink.configuration.Configuration +import org.apache.flink.util.Collector +import org.apache.flink.api.common.state.ValueState +import org.apache.flink.api.common.state.ValueStateDescriptor +import org.apache.flink.api.java.typeutils.RowTypeInfo +import org.apache.flink.table.functions.ProcTimeType +import org.apache.flink.table.functions.RowTimeType +import org.apache.calcite.rel.core.Sort +import org.apache.flink.api.java.functions.NullByteKeySelector +import org.apache.calcite.rel.RelFieldCollation.Direction + + +/** + * Flink RelNode which matches along with Sort Rule. + * + */ +class DataStreamSort( + calc: LogicalSort, + cluster: RelOptCluster, + traitSet: RelTraitSet, + inputNode: RelNode, + rowRelDataType: RelDataType, + inputType: RelDataType, + description: String) + extends SingleRel(cluster, traitSet, inputNode) with DataStreamRel { + + override def deriveRowType(): RelDataType = rowRelDataType + + override def copy(traitSet: RelTraitSet, inputs: java.util.List[RelNode]): RelNode = { + new DataStreamSort( + calc, + cluster, + traitSet, + inputs.get(0), + rowRelDataType, + inputType, + description + calc.getId()) + } + + override def toString: String = { + s"Sort($calc)"+ + s"on fields: (${calc.collation.getFieldCollations})" + } + + override def explainTerms(pw: RelWriter): RelWriter = { + super.explainTerms(pw) + .item("aggregate", calc) + .item("sort fields",calc.collation.getFieldCollations) + .itemIf("offset", calc.offset, calc.offset!=null) + .itemIf("fetch", calc.fetch, calc.fetch!=null) + .item("input", inputNode) + } + + override def translateToPlan(tableEnv: StreamTableEnvironment): DataStream[Row] = { + + val inputDS = getInput.asInstanceOf[DataStreamRel].translateToPlan(tableEnv) + + //need to identify time between others order fields. Time needs to be first sort element + val timeType = SortUtil.getTimeType(calc,inputType) + + //time ordering needs to be ascending + if(SortUtil.getTimeDirection(calc)!=Direction.ASCENDING) { + throw new TableException("SQL/Table supports only ascending time ordering") + } + + + val (offset,fetch) = (calc.offset,calc.fetch) + + //enable to extend for other types of aggregates that will not be implemented in a window + timeType match { + case _: ProcTimeType => + (offset,fetch) match { + case (o:Any,f:Any) => null // offset and fetch needs retraction + case (_,f:Any) => null // offset needs retraction + case (o:Any,_) => null // fetch needs retraction + case _ => createSortProcTime(inputDS) //sort can be done with/without retraction + } + case _: RowTimeType => + throw new TableException("SQL/Table does not support sort on row time") + case _ => + throw new TableException("SQL/Table needs to have sort on time as first sort element") + } + + } + + /** + * Create Sort logic based on processing time + */ + def createSortProcTime( + inputDS: DataStream[Row]): DataStream[Row] = { + + /* + * [TODO] consider the partition case - In this implementation even if the stream is + * pre-partitioned all elements will be reshufled and sorted + */ + + + // get the output types + //Sort does not do project.= Hence it will output also the ordering proctime field + //[TODO]Do we need to drop the ordering fields? + val rowTypeInfo = FlinkTypeFactory.toInternalRowTypeInfo(getRowType).asInstanceOf[RowTypeInfo] + + val processFunction = SortUtil.createProcTimeSortFunction(calc,inputType) + + val result: DataStream[Row] = inputDS + .keyBy(new NullByteKeySelector[Row]) + .process(processFunction).setParallelism(1).setMaxParallelism(1) + .returns(rowTypeInfo) + .asInstanceOf[DataStream[Row]] + + result + + } + + ///need to check for offset/fetch + +} + + diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/rules/FlinkRuleSets.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/rules/FlinkRuleSets.scala index 1301c8dfe8cecc..ba31ed7da50c17 100644 --- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/rules/FlinkRuleSets.scala +++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/rules/FlinkRuleSets.scala @@ -182,7 +182,9 @@ object FlinkRuleSets { // scan optimization PushProjectIntoStreamTableSourceScanRule.INSTANCE, - PushFilterIntoStreamTableSourceScanRule.INSTANCE + PushFilterIntoStreamTableSourceScanRule.INSTANCE, + + DataStreamSortRule.INSTANCE ) /** diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/rules/datastream/DataStreamSortRule.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/rules/datastream/DataStreamSortRule.scala new file mode 100644 index 00000000000000..3952725632c1df --- /dev/null +++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/rules/datastream/DataStreamSortRule.scala @@ -0,0 +1,67 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.flink.table.plan.rules.datastream + +import org.apache.calcite.plan.volcano.RelSubset +import org.apache.calcite.plan.{ Convention, RelOptRule, RelOptRuleCall, RelTraitSet } +import org.apache.calcite.rel.RelNode +import org.apache.calcite.rel.convert.ConverterRule +import org.apache.calcite.rel.logical.{ LogicalFilter, LogicalCorrelate, LogicalTableFunctionScan } +import org.apache.calcite.rex.RexNode +import org.apache.flink.table.plan.nodes.datastream.DataStreamConvention +import org.apache.flink.table.plan.nodes.datastream.DataStreamCorrelate +import org.apache.calcite.rel.logical.LogicalSort +import org.apache.flink.table.plan.nodes.datastream.DataStreamSort + +/** + * Rule to convert a LogicalSort into a DataStreamSort. + */ +class DataStreamSortRule + extends ConverterRule( + classOf[LogicalSort], + Convention.NONE, + DataStreamConvention.INSTANCE, + "DataStreamSortRule") { + + override def matches(call: RelOptRuleCall): Boolean = { + super.matches(call) + } + + override def convert(rel: RelNode): RelNode = { + val calc: LogicalSort = rel.asInstanceOf[LogicalSort] + val traitSet: RelTraitSet = rel.getTraitSet.replace(DataStreamConvention.INSTANCE) + val convInput: RelNode = RelOptRule.convert(calc.getInput(0), DataStreamConvention.INSTANCE) + + val inputRowType = convInput.asInstanceOf[RelSubset].getOriginal.getRowType + + new DataStreamSort( + calc, + rel.getCluster, + traitSet, + convInput, + rel.getRowType, + inputRowType, + description + calc.getId()) + + } + +} + +object DataStreamSortRule { + val INSTANCE: RelOptRule = new DataStreamSortRule +} diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/ProcTimeUnboundedSortProcessFunction.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/ProcTimeUnboundedSortProcessFunction.scala new file mode 100644 index 00000000000000..72916ded0036d8 --- /dev/null +++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/ProcTimeUnboundedSortProcessFunction.scala @@ -0,0 +1,139 @@ +/* + * 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.runtime.aggregate + +import org.apache.flink.api.common.state.{ ListState, ListStateDescriptor } +import org.apache.flink.api.java.typeutils.RowTypeInfo +import org.apache.flink.configuration.Configuration +import org.apache.flink.runtime.state.{ FunctionInitializationContext, FunctionSnapshotContext } +import org.apache.flink.streaming.api.functions.ProcessFunction +import org.apache.flink.table.functions.{ Accumulator, AggregateFunction } +import org.apache.flink.types.Row +import org.apache.flink.util.{ Collector, Preconditions } +import org.apache.flink.api.common.state.ValueState +import org.apache.flink.api.common.state.ValueStateDescriptor +import scala.util.control.Breaks._ +import org.apache.flink.api.java.tuple.{ Tuple2 => JTuple2 } +import org.apache.flink.api.common.state.MapState +import org.apache.flink.api.common.state.MapStateDescriptor +import org.apache.flink.api.common.typeinfo.TypeInformation +import org.apache.flink.api.java.typeutils.ListTypeInfo +import java.util.{ ArrayList, LinkedList, List => JList } +import org.apache.flink.api.common.typeinfo.BasicTypeInfo +import org.apache.flink.table.runtime.aggregate.MultiOutputAggregateFunction + +/** + * Process Function used for the aggregate in bounded proc-time OVER window + * [[org.apache.flink.streaming.api.datastream.DataStream]] + * + * @param aggregates the [[org.apache.flink.table.functions.aggfunctions.SortAggFunction]] + * used for this sort aggregation + * @param fieldCount Is used to indicate fields in the current element to forward + * @param aggType It is used to mark the Aggregate type + */ +class ProcTimeUnboundedSortProcessFunction( + private val aggregates: MultiOutputAggregateFunction[_], + private val fieldCount: Int, + private val aggType: TypeInformation[Row]) + extends ProcessFunction[Row, Row] { + + Preconditions.checkNotNull(aggregates) + + private var output: Row = _ + private var accumulatorState: ValueState[Row] = _ + + override def open(config: Configuration) { + output = new Row(fieldCount) + + val stateDescriptor: ValueStateDescriptor[Row] = + new ValueStateDescriptor[Row]("sortState", aggType) + accumulatorState = getRuntimeContext.getState(stateDescriptor) + } + + override def processElement( + input: Row, + ctx: ProcessFunction[Row, Row]#Context, + out: Collector[Row]): Unit = { + + val currentTime = ctx.timerService.currentProcessingTime + //buffer the event incoming event + + //initialize the accumulators + var accumulators = accumulatorState.value() + if (null == accumulators) { + accumulators = new Row(1) + accumulators.setField(0, aggregates.createAccumulator) + } + + //we aggregate(sort) the events as they arrive. However, this works only + //if the onTimer is called before the processElement which should be the case + val accumulator = accumulators.getField(0).asInstanceOf[Accumulator] + aggregates.accumulate(accumulator, input) + accumulatorState.update(accumulators) + + //deduplication of multiple registered timers is done automatically + ctx.timerService.registerProcessingTimeTimer(currentTime + 1) + + } + + override def onTimer( + timestamp: Long, + ctx: ProcessFunction[Row, Row]#OnTimerContext, + out: Collector[Row]): Unit = { + + var i = 0 + + //initialize the accumulators + var accumulators = accumulatorState.value() + if (null == accumulators) { + accumulators = new Row(1) + accumulators.setField(0, aggregates.createAccumulator) + } + + val accumulator = accumulators.getField(0).asInstanceOf[Accumulator] + + //no retraction now + //... + + //get the list of elements of current proctime + var sortedValues = aggregates.getValues(accumulator) + + + //we need to build the output and emit for every event received at this proctime + var iElemenets = 0 + while (iElemenets < sortedValues.size) { + val input = sortedValues.get(iElemenets).asInstanceOf[Row] + + //set the fields of the last event to carry on with the aggregates + i = 0 + while (i < fieldCount) { + output.setField(i, input.getField(i)) + i += 1 + } + + out.collect(output) + iElemenets += 1 + } + + //update the value of accumulators for future incremental computation + aggregates.resetAccumulator(accumulator) + accumulatorState.update(accumulators) + + } + +} diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/SortUtil.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/SortUtil.scala new file mode 100644 index 00000000000000..ae4c1282e1783d --- /dev/null +++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/SortUtil.scala @@ -0,0 +1,367 @@ +/* + * 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.runtime.aggregate + +import org.apache.flink.table.calcite.FlinkTypeFactory +import org.apache.flink.api.java.typeutils.RowTypeInfo +import org.apache.flink.types.Row +import org.apache.calcite.rel.`type`._ +import org.apache.calcite.rel.logical.LogicalSort +import org.apache.flink.streaming.api.functions.ProcessFunction +import org.apache.flink.table.functions.AggregateFunction +import org.apache.calcite.sql.`type`.SqlTypeName +import org.apache.flink.table.api.TableException +import org.apache.calcite.sql.`type`.SqlTypeName +import org.apache.calcite.sql.`type`.SqlTypeName._ +import org.apache.flink.table.functions.Accumulator +import java.util.{ List => JList, ArrayList } +import org.apache.flink.api.common.typeinfo.{ SqlTimeTypeInfo, TypeInformation } +import org.apache.flink.api.java.typeutils.RowTypeInfo +import org.apache.flink.table.functions.aggfunctions.RowSortAggFunction +import java.sql.Timestamp +import org.apache.calcite.rel.RelFieldCollation +import org.apache.calcite.rel.RelFieldCollation.Direction + + +/** + * Class represents a collection of helper methods to build the sort logic. + * It encapsulates as well implementation for ordering and generic interfaces + */ + +object SortUtil { + + /** + * Function creates [org.apache.flink.streaming.api.functions.ProcessFunction] for sorting + * elements based on proctime and potentially other fields + * @param calcSort Sort logical object + * @param inputType input row type + * @return org.apache.flink.streaming.api.functions.ProcessFunction + */ + private[flink] def createProcTimeSortFunction( + calcSort: LogicalSort, + inputType: RelDataType): ProcessFunction[Row, Row] = { + + val keySortFields = getSortFieldIndexList(calcSort) + val keySortDirections = getSortFieldDirectionList(calcSort) + val sortAggregates = createSortAggregation(inputType, keySortFields,keySortDirections, false) + + val aggType = createSingleAccumulatorRowType(sortAggregates) + + new ProcTimeUnboundedSortProcessFunction( + sortAggregates, + inputType.getFieldCount, + aggType) + + } + + + /** + * Function creates a sorting aggregation object + * elements based on proctime and potentially other fields + * @param inputType input row type + * @param keyIndex the indexes of the fields on which the sorting is done. + * @param keySortDirections the directions of the sorts for each field. + * First is expected to be the time + * @return SortAggregationFunction + */ + private def createSortAggregation( + inputType: RelDataType, + keyIndex: Array[Int], + keySortDirections: Array[Direction], + retraction: Boolean): MultiOutputAggregateFunction[_] = { + + val orderings = createOrderingComparison(inputType, keyIndex) + + val sortKeyType = toKeySortInternalRowTypeInfo(inputType, keyIndex).asInstanceOf[RowTypeInfo] + + // get the output types + val rowTypeInfo = FlinkTypeFactory.toInternalRowTypeInfo(inputType).asInstanceOf[RowTypeInfo] + + val sortAggFunc = new RowSortAggFunction(keyIndex, + keySortDirections, orderings, rowTypeInfo, sortKeyType) + + sortAggFunc + + } + + /** + * Function creates a typed based comparison objects + * @param inputType input row type + * @param keyIndex the indexes of the fields on which the sorting is done. + * First is expected to be the time + * @return Array of ordering objects + */ + + def createOrderingComparison(inputType: RelDataType, + keyIndex: Array[Int]): Array[UntypedOrdering] = { + + var i = 0 + val orderings = new Array[UntypedOrdering](keyIndex.size) + + while (i < keyIndex.size) { + val sqlTypeName = inputType.getFieldList.get(keyIndex(i)).getType.getSqlTypeName + + orderings(i) = sqlTypeName match { + case TINYINT => + new ByteOrdering() + case SMALLINT => + new SmallOrdering() + case INTEGER => + new IntOrdering() + case BIGINT => + new LongOrdering() + case FLOAT => + new FloatOrdering() + case DOUBLE => + new DoubleOrdering() + case DECIMAL => + new DecimalOrdering() + case VARCHAR | CHAR => + new StringOrdering() + //should be updated when times are merged in master branch based on their types + case TIMESTAMP => + new TimestampOrdering() + case sqlType: SqlTypeName => + throw new TableException("Sort aggregate does no support type:" + sqlType) + } + i += 1 + } + orderings + } + + /** + * Function creates a type for sort aggregation + * @param sort input row type + * @param keyIndex the indexes of the fields on which the sorting is done. + * First is expected to be the time + * @return org.apache.flink.streaming.api.functions.ProcessFunction + */ + private def createSingleAccumulatorRowType( + sortAggregate: MultiOutputAggregateFunction[_]): RowTypeInfo = { + + val accType = sortAggregate.getAccumulatorType + new RowTypeInfo(accType) + } + + /** + * Extracts and converts a Calcite logical record into a Flink type information + * by selecting certain only a subset of the fields + */ + def toKeySortInternalRowTypeInfo(logicalRowType: RelDataType, + keyIndexes: Array[Int]): TypeInformation[Row] = { + var i = 0 + val fieldList = logicalRowType.getFieldList + val logicalFieldTypes = new Array[TypeInformation[_]](keyIndexes.size) + val logicalFieldNames = new Array[String](keyIndexes.size) + + while (i < keyIndexes.size) { + logicalFieldTypes(i) = (FlinkTypeFactory.toTypeInfo( + logicalRowType.getFieldList.get(i).getType)) + logicalFieldNames(i) = (logicalRowType.getFieldNames.get(i)) + i += 1 + } + + new RowTypeInfo(logicalFieldTypes.toArray, logicalFieldNames.toArray) + + } + + /** + * Function returns the array of indexes for the fields on which the sort is done + * @param calcSort The LogicalSort object + * @return [Array[Int]] + */ + def getSortFieldIndexList(calcSort: LogicalSort): Array[Int] = { + val keyFields = calcSort.collation.getFieldCollations + var i = 0 + val keySort = new Array[Int](keyFields.size()) + while (i < keyFields.size()) { + keySort(i) = keyFields.get(i).getFieldIndex + i += 1 + } + keySort + } + + /** + * Function returns the array of sort direction for the sort fields + * @param calcSort The LogicalSort object + * @return [Array[Int]] + */ + def getSortFieldDirectionList(calcSort: LogicalSort): Array[Direction] = { + val keyFields = calcSort.collation.getFieldCollations + var i = 0 + val keySortDirection = new Array[Direction](keyFields.size()) + while (i < keyFields.size()) { + keySortDirection(i) = getDirection(calcSort,i) + i += 1 + } + keySortDirection + } + + /** + * Function returns the time type in order clause. Expectation is that if exists if is the + * primary sort field + * @param calcSort The LogicalSort object + * @param rowType The data type of the input + * @return [Array[Int]] + */ + def getTimeType(calcSort: LogicalSort, rowType: RelDataType): RelDataType = { + + //need to identify time between others order fields + // + val ind = calcSort.getCollationList.get(0).getFieldCollations.get(0).getFieldIndex + rowType.getFieldList.get(ind).getValue + } + + /** + * Function returns the direction type of the time in order clause. + * @param calcSort The LogicalSort object + * @return [Array[Int]] + */ + def getTimeDirection(calcSort: LogicalSort):Direction = { + calcSort.getCollationList.get(0).getFieldCollations.get(0).direction + } + + /** + * Function returns the direction type of the field in order clause. + * @param calcSort The LogicalSort object + * @return [Array[Int]] + */ + def getDirection(calcSort: LogicalSort, sortField:Int):Direction = { + + calcSort.getCollationList.get(0).getFieldCollations.get(sortField).direction match { + case Direction.ASCENDING => Direction.ASCENDING + case Direction.DESCENDING => Direction.DESCENDING + case _ => throw new TableException("SQL/Table does not support such sorting") + } + + } + +} + + +/** + * Untyped interface for defining comparison method that can be override by typed implementations + * Each typed implementation will cast the generic type to the implicit ordering type used + */ + +trait UntypedOrdering extends Serializable{ + def compare(x: Any, y: Any): Int + +} + +class LongOrdering(implicit ord: Ordering[Long]) extends UntypedOrdering { + + override def compare(x: Any, y: Any): Int = { + val xL = x.asInstanceOf[Long] + val yL = y.asInstanceOf[Long] + ord.compare(xL, yL) + } +} + +class IntOrdering(implicit ord: Ordering[Int]) extends UntypedOrdering { + + override def compare(x: Any, y: Any): Int = { + val xI = x.asInstanceOf[Int] + val yI = y.asInstanceOf[Int] + ord.compare(xI, yI) + } +} + +class FloatOrdering(implicit ord: Ordering[Float]) extends UntypedOrdering { + + override def compare(x: Any, y: Any): Int = { + val xF = x.asInstanceOf[Float] + val yF = y.asInstanceOf[Float] + ord.compare(xF, yF) + } +} + +class DoubleOrdering(implicit ord: Ordering[Double]) extends UntypedOrdering { + + override def compare(x: Any, y: Any): Int = { + val xD = x.asInstanceOf[Double] + val yD = y.asInstanceOf[Double] + ord.compare(xD, yD) + } +} + +class DecimalOrdering(implicit ord: Ordering[BigDecimal]) extends UntypedOrdering { + + override def compare(x: Any, y: Any): Int = { + val xBD = x.asInstanceOf[BigDecimal] + val yBD = y.asInstanceOf[BigDecimal] + ord.compare(xBD, yBD) + } +} + +class ByteOrdering(implicit ord: Ordering[Byte]) extends UntypedOrdering { + + override def compare(x: Any, y: Any): Int = { + val xB = x.asInstanceOf[Byte] + val yB = y.asInstanceOf[Byte] + ord.compare(xB, yB) + } +} + +class SmallOrdering(implicit ord: Ordering[Short]) extends UntypedOrdering { + + override def compare(x: Any, y: Any): Int = { + val xS = x.asInstanceOf[Short] + val yS = y.asInstanceOf[Short] + ord.compare(xS, yS) + } +} + +class StringOrdering(implicit ord: Ordering[String]) extends UntypedOrdering { + + override def compare(x: Any, y: Any): Int = { + val xS = x.asInstanceOf[String] + val yS = y.asInstanceOf[String] + ord.compare(xS, yS) + } +} + +/** + * Ordering object for timestamps. As there is no implicit Ordering for [java.sql.Timestamp] + * we need to compare based on the Long value of the timestamp + */ +class TimestampOrdering(implicit ord: Ordering[Long]) extends UntypedOrdering { + + override def compare(x: Any, y: Any): Int = { + val xTs = x.asInstanceOf[Timestamp] + val yTs = y.asInstanceOf[Timestamp] + ord.compare(xTs.getTime, yTs.getTime) + } +} + + + +/** + * Called every time when a multi-output aggregation result should be materialized. + * The returned values could be either an early and incomplete result + * (periodically emitted as data arrive) or the final result of the + * aggregation. + * + * @param accumulator the accumulator which contains the current + * aggregated results + * @return the aggregation result + */ +abstract class MultiOutputAggregateFunction[T] extends AggregateFunction[T] { + + def getValues(accumulator: Accumulator): JList[T] +} diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/scala/stream/sql/SqlITCase.scala b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/scala/stream/sql/SqlITCase.scala index 80ff42ae599116..4f027def38f45b 100644 --- a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/scala/stream/sql/SqlITCase.scala +++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/scala/stream/sql/SqlITCase.scala @@ -696,6 +696,106 @@ class SqlITCase extends StreamingWithStateTestBase { "6,8,Hello world,51,9,5,9,1") assertEquals(expected.sorted, StreamITCase.testResults.sorted) } + + + /** + * Integration tests based on Proctime tests will be replaced with harness tests + */ + @Test + def testOrderBy(): Unit = { + val env = StreamExecutionEnvironment.getExecutionEnvironment + val tEnv = TableEnvironment.getTableEnvironment(env) + StreamITCase.testResults = mutable.MutableList() + + // for sum aggregation ensure that every time the order of each element is consistent + env.setParallelism(1) + + val t1 = env.fromCollection(data).toTable(tEnv).as('a, 'b, 'c) + + tEnv.registerTable("s1", t1) + + val t2 = StreamTestData.get5TupleDataStream(env) + .toTable(tEnv).as('a2, 'b2, 'c2, 'd2, 'e2) + + tEnv.registerTable("s2", t2) + + var sqlquery="SELECT a2 FROM s2 ORDER BY proctime() ASC" + + + val result = tEnv.sql(sqlquery).toDataStream[Row] + result.addSink(new StreamITCase.StringSink) + env.execute() + + //Sort does not do project. Hence it will output also the ordering proctime field + val expected = mutable.MutableList( + "1,1970-01-01 00:00:00.0", + "2,1970-01-01 00:00:00.0", + "2,1970-01-01 00:00:00.0", + "3,1970-01-01 00:00:00.0", + "3,1970-01-01 00:00:00.0", + "3,1970-01-01 00:00:00.0", + "4,1970-01-01 00:00:00.0", + "4,1970-01-01 00:00:00.0", + "4,1970-01-01 00:00:00.0", + "4,1970-01-01 00:00:00.0", + "5,1970-01-01 00:00:00.0", + "5,1970-01-01 00:00:00.0", + "5,1970-01-01 00:00:00.0", + "5,1970-01-01 00:00:00.0", + "5,1970-01-01 00:00:00.0") + + assertEquals(expected.take(5).sorted, StreamITCase.testResults.take(5).sorted) + } + + /** + * Integration tests based on Proctime tests will be replaced with harness tests + */ + @Test + def testOrderBy2(): Unit = { + val env = StreamExecutionEnvironment.getExecutionEnvironment + val tEnv = TableEnvironment.getTableEnvironment(env) + StreamITCase.testResults = mutable.MutableList() + + // for sum aggregation ensure that every time the order of each element is consistent + env.setParallelism(1) + + val t1 = env.fromCollection(data).toTable(tEnv).as('a, 'b, 'c) + + tEnv.registerTable("s1", t1) + + val t2 = StreamTestData.get5TupleDataStream(env) + .toTable(tEnv).as('a2, 'b2, 'c2, 'd2, 'e2) + + tEnv.registerTable("s2", t2) + + var sqlquery = "SELECT a2,CEIL(proctime() TO hour) FROM s2 " + + "ORDER BY CEIL(proctime() TO hour),c2 ASC "; + + val result = tEnv.sql(sqlquery).toDataStream[Row] + result.addSink(new StreamITCase.StringSink) + env.execute() + + //sorting fields proctime and c2 are kept as well + val expected = mutable.MutableList( + "1,1970-01-01 00:00:00.0,0", + "2,1970-01-01 00:00:00.0,1", + "2,1970-01-01 00:00:00.0,2", + "3,1970-01-01 00:00:00.0,3", + "3,1970-01-01 00:00:00.0,4", + "3,1970-01-01 00:00:00.0,5", + "4,1970-01-01 00:00:00.0,6", + "4,1970-01-01 00:00:00.0,7", + "4,1970-01-01 00:00:00.0,8", + "4,1970-01-01 00:00:00.0,9", + "5,1970-01-01 00:00:00.0,10", + "5,1970-01-01 00:00:00.0,11", + "5,1970-01-01 00:00:00.0,12", + "5,1970-01-01 00:00:00.0,13", + "5,1970-01-01 00:00:00.0,14") + assertEquals(expected.take(5).sorted, StreamITCase.testResults.take(5).sorted) + } + + } object SqlITCase {