[FLINK-6075] - Support Limit/Top(Sort) for Stream SQL #3889

rtudoran · 2017-05-12T18:45:33Z

Order by time and optionally other fields is supported

Thanks for contributing to Apache Flink. Before you open your pull request, please take the following check list into consideration.
If your changes take all of the items into account, feel free to open your pull request. For more information and/or questions please refer to the How To Contribute guide.
In addition to going through the list, please provide a meaningful description of your changes.

[x ] General
- The pull request references the related JIRA issue ("[FLINK-XXX] Jira title text")
- The pull request addresses only one issue
- Each commit in the PR has a meaningful commit message (including the JIRA id)
[x ] Documentation
- Documentation has been added for new functionality
- Old documentation affected by the pull request has been updated
- JavaDoc for public methods has been added
[x ] Tests & Build
- Functionality added by the pull request is covered by tests
- mvn clean verify has been executed successfully locally or a Travis build has passed

Order by time and optionally other fields is supported

rtudoran · 2017-05-12T18:47:21Z

@fhueske I updated the code to work with 1.3 (CRow, FlinkLogicalSort....) and all the implied changes.
We can close #3714 after we merge this i guess
..that one is the same logical code only on the old interface

rtudoran · 2017-05-15T01:37:41Z

@hongyuhong @stefanobortoli @shijinkui i forgot to add you to the PR

fhueske

Hi @rtudoran,

thanks for the PR. The overall approach looks good. I make a couple of comments about code style and refactorings but nothing major.

Thanks, Fabian

fhueske · 2017-05-15T10:15:18Z

...flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamSort.scala

+  sortCollation: RelCollation,
+  sortOffset: RexNode,
+  sortFetch: RexNode,
+  cluster: RelOptCluster,


Please keep the order of parameters consistent to other classes (for instance DataStreamCalc), i.e., cluster, traitset, input, inputschema, schema, sortCollation, offset, fetch, description.

fhueske · 2017-05-15T10:19:05Z

...flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamSort.scala

+  }
+
+  override def toString: String = {
+    s"Sort(by: ($SortUtil.getSortFieldToString(sortCollation, rowRelDataType))," +


Move the methods which are common among DataSetSort and DataStreamSort (e.g. methods to create strings) into a new class CommonSort, similar to CommonCalc or CommonCorrelate.

fhueske · 2017-05-15T11:38:44Z

...libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/SortUtil.scala

+
+    //need to identify time between others ordering fields
+    val ind = collationSort.getFieldCollations.get(0).getFieldIndex
+    rowType.getFieldList.get(ind).getValue


use getType instead of getValue

fhueske · 2017-05-15T11:48:35Z

...flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamSort.scala

+    //enable to extend for other types of aggregates that will not be implemented in a window
+    timeType match {
+        case _ if FlinkTypeFactory.isProctimeIndicatorType(timeType)  =>
+            (sortOffset,sortFetch) match {


fhueske · 2017-05-15T11:49:09Z

...flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamSort.scala

+    timeType match {
+        case _ if FlinkTypeFactory.isProctimeIndicatorType(timeType)  =>
+            (sortOffset,sortFetch) match {
+              case (o: Any, f: Any)  => // offset and fetch needs retraction


Why not use the more concrete type RexNode instead of Any?
if o and f are not used, you can write case (_: RexNode, _: RexNode) =>

fhueske · 2017-05-15T16:30:25Z

...ries/flink-table/src/test/scala/org/apache/flink/table/api/scala/stream/sql/SortITCase.scala

+
+object SortITCase {
+
+  class EventTimeSourceFunction[T](


Make the EventTimeSourceFunction in OverWindowITCase a util class and reuse it here instead of duplicating the code.

fhueske · 2017-05-15T16:37:03Z

...raries/flink-table/src/test/scala/org/apache/flink/table/api/scala/stream/sql/SortTest.scala

+        "DataStreamCalc",
+        unaryNode("DataStreamSort",
+          streamTableNode(0),
+          term("orderBy", "proctime ASC, c ASC], offset=[null], fetch=[unlimited")),


The string seems to be broken. First there is a closing ] and later an opening [

yeah - i was trying to make it match with respect to the number of parenthesis with the incoming value. The problem was that i was messing up with the string and no matter how many i pair i was adding it was replacing and adding some additional ones at the begin and end. The actual problem was that order by / offset and fetch needed to be standalone terms

fhueske · 2017-05-15T16:40:25Z

...raries/flink-table/src/test/scala/org/apache/flink/table/api/scala/stream/sql/SortTest.scala

+    streamUtil.verifySql(sqlQuery, expected)
+  }
+
+   @Test


-1 space. Check expected exceptions with @Test(expected = classOf[YourExpectedException])

fhueske · 2017-05-15T16:40:44Z

...raries/flink-table/src/test/scala/org/apache/flink/table/api/scala/stream/sql/SortTest.scala

+    }
+  }
+
+    @Test


fhueske · 2017-05-15T16:41:04Z

...raries/flink-table/src/test/scala/org/apache/flink/table/api/scala/stream/sql/SortTest.scala

+    @Test
+   def testSortProcessingTimeSecondaryField() = {
+
+    val sqlQuery = "SELECT a FROM MyTable ORDER BY c, proctime"


Please add a comment why the tests are expected to fail

refactor the tests, correct for styles errors

rtudoran · 2017-05-25T12:41:18Z

@fhueske @stefanobortoli @shijinkui @hongyuhong
I have updated the PR. In principle all remarks were addressed. This should be ready to merge (i hope)

fhueske

Hi @rtudoran, the PR looks quite good. I made quite a few comments for minor changes but did not find a major issue.

Let me know what you think,
Fabian

fhueske · 2017-06-23T09:36:17Z

...ries/flink-table/src/test/scala/org/apache/flink/table/api/scala/stream/sql/SortITCase.scala

+    tEnv.registerTable("T1", t1)
+
+    val  sqlQuery = "SELECT b FROM T1 " +
+      "ORDER BY rowtime, b ASC ";


fhueske · 2017-06-23T09:36:53Z

...ries/flink-table/src/test/scala/org/apache/flink/table/api/scala/stream/sql/SortITCase.scala

+      "ORDER BY rowtime, b ASC ";
+
+
+    val result = tEnv.sql(sqlQuery).toDataStream[Row]


toDataStream is deprecated. Please use toAppendStream instead.

In the flink version from where i started it it is not available. I think if i rebase now i will have to create a new PR... I would prefer to keep this and do an update after the merge.

OK, will do that before merging

fhueske · 2017-06-23T09:39:02Z

...es/flink-table/src/main/scala/org/apache/flink/table/calcite/RelTimeIndicatorConverter.scala

+      rexBuilder,
+      input.getRowType.getFieldList.map(_.getType))
+
+    val offset = if(sort.offset != null) sort.offset.accept(materializer) else null


offset and fetch should not contain time attributes. If they do, we cannot execute the query.

fhueske · 2017-06-23T09:40:47Z

flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/CommonSort.scala

+    val offsetToString = s"$offset"
+    offsetToString
+  }
+


rm new line

fhueske · 2017-06-23T09:43:44Z

flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/CommonSort.scala

+
+trait CommonSort {
+
+  private[flink] def offsetToString(offset: RexNode): String = {


IMO, this method can be removed. It simply calls offset.toString()

It is more for consistency - to have a method for "ToString" for each feature of the sort (direction, fetch, fields...). As there is not cost for it I suggest to keep it

fhueske · 2017-06-23T14:22:21Z

...le/src/test/scala/org/apache/flink/table/runtime/aggregate/TimeSortProcessFunctionTest.scala

+
+    val inputCRowType = CRowTypeInfo(rT)
+
+    val processFunction = new KeyedProcessOperator[Integer,CRow,CRow](


+space after comma

fhueske · 2017-06-23T14:22:32Z

...le/src/test/scala/org/apache/flink/table/runtime/aggregate/TimeSortProcessFunctionTest.scala

+      INT_TYPE_INFO,
+      STRING_TYPE_INFO,
+      LONG_TYPE_INFO),
+      Array("a","b","c","d","e"))


space after comma

fhueske · 2017-06-23T14:22:38Z

...le/src/test/scala/org/apache/flink/table/runtime/aggregate/TimeSortProcessFunctionTest.scala

+
+    val rTA =  new RowTypeInfo(Array[TypeInformation[_]](
+     LONG_TYPE_INFO), Array("count"))
+    val indexes = Array(1,2)


fhueske · 2017-06-23T14:24:14Z

...le/src/test/scala/org/apache/flink/table/runtime/aggregate/TimeSortProcessFunctionTest.scala

+        Row.of(1: JInt, 10L: JLong, 0: JInt, "aaa", 11L: JLong), true), 2006))
+
+    //move the timestamp to ensure the execution
+    testHarness.setProcessingTime(1005)


Please run at least two sorts to ensure that all intermediate state is correctly reset.
Also verify the state size to ensure that the operator is not leaking.

I am not sure I understand - you mean within the same test or in different tests?

I wanted to trigger a sort a second time by adding more data and setting the processing time another time. Basically, just what you did. Same would have been good for the rowtime test.

fhueske · 2017-06-23T14:26:31Z

...le/src/test/scala/org/apache/flink/table/runtime/aggregate/TimeSortProcessFunctionTest.scala

+import org.apache.flink.table.runtime.types.{CRow, CRowTypeInfo}
+import org.apache.flink.streaming.api.TimeCharacteristic
+
+class TimeSortProcessFunctionTest{


Please add a similar test for the event-time sorter which sorts for multiple timestamps and validates that all rows preserve their timestamp.

I do not understand - this test was added (see below testSortRowTimeHarnessPartitioned) ...perhaps this is an old comment ...

- multiple indentations, spaces and style things - further refactor methods towards CommonSort - extend and update the tests

rtudoran · 2017-06-28T15:58:52Z

@fhueske I have addressed the remarks you have done (with 1-2 minor exceptions for which you can see my comments). Please have a look. I think it could be ready to be merged and than I can start the preparation for the offset and fetch

fhueske

Thanks for the update @rtudoran.
I will make some minor fixes and refactorings and merge this PR.

Thanks, Fabian

fhueske · 2017-06-30T10:05:15Z

...es/flink-table/src/main/scala/org/apache/flink/table/calcite/RelTimeIndicatorConverter.scala

+
+    val input = sort.getInput.accept(this)
+
+    val materializer = new RexTimeIndicatorMaterializer(


unused. Should be removed

fhueske · 2017-06-30T10:05:22Z

...es/flink-table/src/main/scala/org/apache/flink/table/calcite/RelTimeIndicatorConverter.scala

+      rexBuilder,
+      input.getRowType.getFieldList.map(_.getType))
+
+    //val offset = if(sort.offset != null) sort.offset.accept(materializer) else null


Should be removed

fhueske · 2017-06-30T13:36:19Z

...flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamSort.scala

+
+package org.apache.flink.table.plan.nodes.datastream
+
+import org.apache.calcite.plan.{ RelOptCluster, RelTraitSet }


About half of the imports are unused. Other classes have unused imports as well.

fhueske · 2017-06-30T14:53:37Z

...le/src/test/scala/org/apache/flink/table/runtime/aggregate/TimeSortProcessFunctionTest.scala

+    expectedOutput.add(new StreamRecord(new CRow(
+      Row.of(1: JInt, 1L: JLong, 0: JInt, "aaa", 11L: JLong),true), 1006))
+    expectedOutput.add(new StreamRecord(new CRow(
+      Row.of(1: JInt, 2L: JLong, 0: JInt, "aaa", 11L: JLong),true), 1006))

    TestHarnessUtil.assertOutputEqualsSorted("Output was not correct.",


The test should check that the ProcessFunction emit the rows in the correct order. assertOutputEqualsSorted sorts the result and expected data before comparing them. We have to use assertOutputEquals instead.

fhueske · 2017-06-30T14:55:06Z

...le/src/test/scala/org/apache/flink/table/runtime/aggregate/TimeSortProcessFunctionTest.scala

+        Row.of(1: JInt, 10L: JLong, 0: JInt, "aaa", 11L: JLong), true), 2006))
+
+    //move the timestamp to ensure the execution
+    testHarness.setProcessingTime(1005)


I wanted to trigger a sort a second time by adding more data and setting the processing time another time. Basically, just what you did. Same would have been good for the rowtime test.

fhueske · 2017-06-30T15:13:03Z

...ries/flink-table/src/test/scala/org/apache/flink/table/api/scala/stream/sql/SortITCase.scala

+      "ORDER BY rowtime, b ASC ";
+
+
+    val result = tEnv.sql(sqlQuery).toDataStream[Row]


OK, will do that before merging

This closes apache#3889.

rtudoran · 2017-07-03T07:38:35Z

@fhueske Thanks for the review and for handling the the fixes (sorry for still kipping some :( ). I will create this week the PR for the rest of the functions.

Sort support moved to 1.3 interface

233577a

Order by time and optionally other fields is supported

rtudoran mentioned this pull request May 12, 2017

[FLINK-6075] - Support Limit/Top(Sort) for Stream SQL #3714

Closed

fhueske reviewed May 15, 2017

View reviewed changes

Integrated the review remarks: refactor sort utils in common class,

90218cb

refactor the tests, correct for styles errors

rtudoran mentioned this pull request May 28, 2017

[FLINK-6075] - Support Limit/Top(Sort) for Stream SQL #4003

Closed

fhueske reviewed Jun 23, 2017

View reviewed changes

Addressed the remarks on PR:

04512e8

- multiple indentations, spaces and style things - further refactor methods towards CommonSort - extend and update the tests

fhueske reviewed Jun 30, 2017

View reviewed changes

fhueske pushed a commit to fhueske/flink that referenced this pull request Jun 30, 2017

[FLINK-6075] [table] Add ORDER BY support for streaming table.

7b0fb13

This closes apache#3889.

asfgit closed this in b8c8f20 Jun 30, 2017

zhenzhongxu pushed a commit to zhenzhongxu/flink that referenced this pull request Jun 30, 2017

[FLINK-6075] [table] Add ORDER BY support for streaming table.

48875e9

This closes apache#3889.

rmetzger added the component=API/TableSQL label Mar 14, 2019

flinkbot added component=TableSQL/API and removed component=API/TableSQL labels Mar 17, 2022

+                  }
+                }
+                  @Test

		"ORDER BY rowtime, b ASC ";


		val result = tEnv.sql(sqlQuery).toDataStream[Row]


		trait CommonSort {

		private[flink] def offsetToString(offset: RexNode): String = {


		val inputCRowType = CRowTypeInfo(rT)

		val processFunction = new KeyedProcessOperator[Integer,CRow,CRow](


		val input = sort.getInput.accept(this)

		val materializer = new RexTimeIndicatorMaterializer(


		package org.apache.flink.table.plan.nodes.datastream

		import org.apache.calcite.plan.{ RelOptCluster, RelTraitSet }

[FLINK-6075] - Support Limit/Top(Sort) for Stream SQL #3889

[FLINK-6075] - Support Limit/Top(Sort) for Stream SQL #3889

Conversation

rtudoran commented May 12, 2017

rtudoran commented May 12, 2017

rtudoran commented May 15, 2017

fhueske left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

rtudoran commented May 25, 2017

fhueske left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

rtudoran commented Jun 28, 2017

fhueske left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

rtudoran commented Jul 3, 2017