[FLINK-11074] [table][tests] Enable harness tests with RocksdbStateBackend and add harness tests for CollectAggFunction #7253

dianfu · 2018-12-06T12:38:25Z

What is the purpose of the change

This pull request enables the harness tests to test with RocksdbStateBackend and adds harness test for CollectAggFunction

Brief change log

Adds AggFunctionHarnessTest
Fix an issue in CollectAggFunction

Verifying this change

Added tests in AggFunctionHarnessTest

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): (no)
The public API, i.e., is any changed class annotated with @Public(Evolving): (no)
The serializers: (no)
The runtime per-record code paths (performance sensitive): (no)
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Yarn/Mesos, ZooKeeper: (no)
The S3 file system connector: (no)

Documentation

Does this pull request introduce a new feature? (no)
If yes, how is the feature documented? (not applicable)

walterddr

Thanks for putting the PR together so quickly @dianfu . I haven't got a close look and will checkout and play around over the weekend. I just left a few suggestions based on a quick look. Please kindly take a look and see if they make sense :-)

walterddr · 2018-12-07T01:05:16Z

...ink-table/src/test/scala/org/apache/flink/table/runtime/harness/AggFunctionHarnessTest.scala

+
+  @Test
+  def testCollectAggregate(): Unit = {
+    val processFunction = new LegacyKeyedProcessOperator[String, CRow, CRow](


this is deprecated right?

Correct. That's because GroupAggProcessFunction extends ProcessFunction, not KeyedProcessFunction and so we have to use LegacyKeyedProcessOperator at present. We can definitely improve GroupAggProcessFunction to extend KeyedProcessFunction, what about improving it in another ticket?

walterddr · 2018-12-07T01:06:20Z

...ink-table/src/test/scala/org/apache/flink/table/runtime/harness/AggFunctionHarnessTest.scala

+  def testCollectAggregate(): Unit = {
+    val processFunction = new LegacyKeyedProcessOperator[String, CRow, CRow](
+      new GroupAggProcessFunction(
+        genCollectAggFunction,


since this is the only place related to the actual agg function. we can probably make this more generic (see how window operator test harness was implemented)

I have reimplemented this PR. Please help to take a look if this issue still exists.

walterddr · 2018-12-07T01:08:09Z

...ink-table/src/test/scala/org/apache/flink/table/runtime/harness/AggFunctionHarnessTest.scala

+    expectedOutput.add(new StreamRecord(CRow("aaa", Map(1 -> 1).asJava), 1))
+    testHarness.processElement(new StreamRecord(CRow(2L: JLong, 1: JInt, "bbb"), 1))
+    expectedOutput.add(new StreamRecord(CRow("bbb", Map(1 -> 1).asJava), 1))
+


can we add something like:

// do a snapshot & close State snapshot = testHarness.snapshot(0L, 0L); testHarness.close(); // reopen and restore testHarness = createTestHarness(operator); testHarness.setup(); testHarness.initializeState(snapshot); testHarness.open();

this will catch some of the weird serialization/deserialization problem as well.

Good idea. What about adding this kind of tests for the operator tests, such as GroupAggregateHarnessTest, JoinHarnessTest, etc as I think it's more useful for operator test.

dianfu · 2018-12-07T11:37:38Z

@twalthr @sunjincheng121 @walterddr I have added an AggFunctionHarnessTest in another way from the current harness test by constructing the operator to test by compiling a sql query which can generate the operator, not by wrapping a hand written GeneratedAggregate directly. I think it has the following benefits:

The harness test can also cover AggregateCodeGenerator as the operator is now code generated. For example, we are able to test if the DataView is actually replaced with StateDataView right now.
Writing a new harness test becomes easy. We just need to construct a sql query which can generate the operator we want to test

Currently I have not changed other harness tests. We can improve other harness test in this way in the following PRs once we are sure this direction is what we want.

Could you help to take a look at this PR? Thanks a lot.

walterddr

Thanks for the quick update @dianfu. I myself like the way how the operator code is now constructed through SQL instead of hard-coded codegen. The only concern I have is that this might lead to more of an "ITCase" instead of unit-test as it adds possibility that test exception can now come from compilation. Maybe @twalthr can provide more insight on this.

Overall the change looks very good to me. I only added minor comments. If we all agree with how the harness test codegen can be replaced by actual compilation, we can definitely work together and change others as well. 👍

walterddr · 2018-12-10T18:09:30Z

...ries/flink-table/src/test/scala/org/apache/flink/table/runtime/harness/HarnessTestBase.scala

@@ -87,7 +85,7 @@ class HarnessTestBase {
    new RowTypeInfo(distinctCountAggregates.map(getAccumulatorTypeOfAggregateFunction(_)): _*)

  protected val distinctCountDescriptor: String = EncodingUtils.encodeObjectToString(
-    new MapStateDescriptor("distinctAgg0", distinctCountAggregationStateType, Types.LONG))


This change doesn't seem necessary

walterddr · 2018-12-10T18:12:02Z

...ries/flink-table/src/test/scala/org/apache/flink/table/runtime/harness/HarnessTestBase.scala

+        } else {
+          extractExpectedTransformation(one.getInput, prefixOperatorName)
+        }
+      case two: TwoInputTransformation[_, _, _] =>


Let's throw unsupported operation for now, since there's no code path that executes two input transform yet. we can always add this logic later when necessary.

walterddr · 2018-12-10T18:14:38Z

...ink-table/src/test/scala/org/apache/flink/table/runtime/harness/AggFunctionHarnessTest.scala

+    testHarness.close()
+  }
+
+  private def getState(


This can probably be put into HarnessTestBase as well. As of now I can only image the Operator to be accessible only when we need to manipulate the internal State to mock statebackend operation on top.

Make sense. Done.

dianfu

@walterddr Thanks a lot for your review and comments. Have updated the PR accordingly.

dianfu · 2018-12-11T01:31:28Z

...ink-table/src/test/scala/org/apache/flink/table/runtime/harness/AggFunctionHarnessTest.scala

+    testHarness.close()
+  }
+
+  private def getState(


Make sense. Done.

dianfu · 2018-12-11T01:31:39Z

...ries/flink-table/src/test/scala/org/apache/flink/table/runtime/harness/HarnessTestBase.scala

@@ -87,7 +85,7 @@ class HarnessTestBase {
    new RowTypeInfo(distinctCountAggregates.map(getAccumulatorTypeOfAggregateFunction(_)): _*)

  protected val distinctCountDescriptor: String = EncodingUtils.encodeObjectToString(
-    new MapStateDescriptor("distinctAgg0", distinctCountAggregationStateType, Types.LONG))


dianfu · 2018-12-11T01:33:28Z

...ries/flink-table/src/test/scala/org/apache/flink/table/runtime/harness/HarnessTestBase.scala

+        } else {
+          extractExpectedTransformation(one.getInput, prefixOperatorName)
+        }
+      case two: TwoInputTransformation[_, _, _] =>


dianfu · 2018-12-11T08:58:43Z

@twalthr Sorry for interrupt. It would be great if you can share some thoughts on the changes of this PR. Thanks in advance.

dianfu · 2018-12-13T05:47:51Z

@twalthr @sunjincheng121 Could you help to take a look at this PR? You can see from the harness test of #7286 that this change makes it very easy to write a new harness test.

sunjincheng121

Hi @dianfu Thanks for your great job, I like your way of solving HarnessTest, which overrides the codegen logic and covers the test of the use of dataView by acc. Very good！
I only left two minor suggestions.
Thanks,
Jincheng

sunjincheng121 · 2018-12-13T23:26:47Z

...ink-table/src/test/scala/org/apache/flink/table/runtime/harness/AggFunctionHarnessTest.scala

+    expectedOutput.add(new StreamRecord(CRow(false, "aaa", Map(1 -> 2).asJava), 1))
+    expectedOutput.add(new StreamRecord(CRow("aaa", Map(1 -> 2, 2 -> 1).asJava), 1))
+
+    // remove some state: state may be cleaned up by the state backend if not accessed more than ttl


more than ttl -> beyond ttl time?

sunjincheng121 · 2018-12-14T00:41:48Z

...ries/flink-table/src/test/scala/org/apache/flink/table/runtime/harness/HarnessTestBase.scala

@@ -491,13 +491,83 @@ class HarnessTestBase {
    distinctCountFuncName,
    distinctCountAggCode)

+  def createHarnessTester[KEY, IN, OUT](
+      dataStream: DataStream[_],
+      prefixOperatorName: String)


How a bout add aggFieldAlias: String = "", for resolve scenarios where GroupBy is included in multiple UNION clauses. e.g:

(SELECT b, max(a) as maxA FROM T GROUP BY b) UNION ( SELECT b, min(a) as minA FROM ( SELECT a, b FROM T GROUP BY a, b ) GROUP BY b )

And we using this method as follows: createHarnessTester(xx, "groupBy", "minA")。
I didn't find a case where I had to test it in this way, it was just an enhanced suggestion.
What do you think?

createHarnessTester will be used not only in agg related tests but also in other harness tests, such as stream join tests, temporal join tests, sort tests, etc. So field aggFieldAlias seems a little wired from my point of view. What about adding it when we really need it? At that time we may have a better idea on how such a field will look like. Thoughts?

Hmm, If this tester will using in join test, we should add TwoInputTransformation check in the extractExpectedTransformation logic, then we also need add a xxName(may be not named aggFieldAlias), e.g.: In multiple join scenes？

Yes, you're right. Will do that when updating the join related harness tests.

…ckend and add harness tests for CollectAggFunction

dianfu · 2018-12-14T02:18:30Z

@sunjincheng121 Thanks a lot for your review, appreciate! Updated the PR accordingly.

sunjincheng121 · 2018-12-14T03:12:49Z

Thanks for the update and feedback @dianfu
This PR is a base code of the harness test, and I am fine that improve the tester in your following JIRAs.
From the points of my view the PR looks good.

Bests,
Jincheng

sunjincheng121 · 2018-12-17T01:28:21Z

Merging...

dianfu · 2018-12-17T06:20:26Z

@sunjincheng121 @walterddr Thanks a lot for the review and merge. Very appreciated!

…ckend and add harness tests for CollectAggFunction This closes apache#7253

dianfu mentioned this pull request Dec 6, 2018

[FLINK-7208] [table] Optimize Min/MaxWithRetractAggFunction with DataView #7201

Closed

dianfu force-pushed the optimize_harness_test branch from 7c7eb80 to 60e5d6f Compare December 6, 2018 12:46

walterddr reviewed Dec 7, 2018

View reviewed changes

dianfu force-pushed the optimize_harness_test branch from cecc443 to 0dfe579 Compare December 7, 2018 10:23

walterddr reviewed Dec 10, 2018

View reviewed changes

dianfu commented Dec 11, 2018

View reviewed changes

dianfu mentioned this pull request Dec 12, 2018

[FLINK-8739] [table] Optimize DISTINCT aggregates to use the same distinct accumulator if possible #7286

Closed

sunjincheng121 self-assigned this Dec 13, 2018

sunjincheng121 reviewed Dec 14, 2018

View reviewed changes

dianfu added 4 commits December 14, 2018 10:16

[FLINK-11074] [table][tests] Enable harness tests with RocksdbStateBa…

10f8c7e

…ckend and add harness tests for CollectAggFunction

update

20c33a2

address comments

1a35f90

address comments

56aa9e6

dianfu force-pushed the optimize_harness_test branch from 712c88a to 56aa9e6 Compare December 14, 2018 02:16

asfgit closed this in 9a45fca Dec 17, 2018

dianfu mentioned this pull request Dec 18, 2018

[FLINK-11176] [table][tests] Improve the harness tests to use the code-generated operator #7322

Closed

tisonkun pushed a commit to tisonkun/flink that referenced this pull request Jan 17, 2019

[FLINK-11074] [table][tests] Enable harness tests with RocksdbStateBa…

76ae59a

…ckend and add harness tests for CollectAggFunction This closes apache#7253

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

dianfu deleted the optimize_harness_test branch June 10, 2020 02:49

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

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-11074] [table][tests] Enable harness tests with RocksdbStateBackend and add harness tests for CollectAggFunction #7253

[FLINK-11074] [table][tests] Enable harness tests with RocksdbStateBackend and add harness tests for CollectAggFunction #7253

dianfu commented Dec 6, 2018

walterddr left a comment

walterddr Dec 7, 2018

dianfu Dec 7, 2018

walterddr Dec 7, 2018 •

edited

dianfu Dec 7, 2018

walterddr Dec 7, 2018 •

edited

dianfu Dec 7, 2018

dianfu commented Dec 7, 2018 •

edited

walterddr left a comment

walterddr Dec 10, 2018

dianfu Dec 11, 2018

walterddr Dec 10, 2018

dianfu Dec 11, 2018

walterddr Dec 10, 2018

dianfu Dec 11, 2018

dianfu left a comment

dianfu Dec 11, 2018

dianfu Dec 11, 2018

dianfu Dec 11, 2018

dianfu commented Dec 11, 2018

dianfu commented Dec 13, 2018

sunjincheng121 left a comment

sunjincheng121 Dec 13, 2018

sunjincheng121 Dec 14, 2018 •

edited

dianfu Dec 14, 2018

sunjincheng121 Dec 14, 2018

dianfu Dec 14, 2018

dianfu commented Dec 14, 2018

sunjincheng121 commented Dec 14, 2018

sunjincheng121 commented Dec 17, 2018

dianfu commented Dec 17, 2018

[FLINK-11074] [table][tests] Enable harness tests with RocksdbStateBackend and add harness tests for CollectAggFunction #7253

[FLINK-11074] [table][tests] Enable harness tests with RocksdbStateBackend and add harness tests for CollectAggFunction #7253

Conversation

dianfu commented Dec 6, 2018

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

walterddr left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

walterddr Dec 7, 2018 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

walterddr Dec 7, 2018 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

dianfu commented Dec 7, 2018 • edited

walterddr 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

dianfu 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

dianfu commented Dec 11, 2018

dianfu commented Dec 13, 2018

sunjincheng121 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

sunjincheng121 Dec 14, 2018 • edited

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

dianfu commented Dec 14, 2018

sunjincheng121 commented Dec 14, 2018

sunjincheng121 commented Dec 17, 2018

dianfu commented Dec 17, 2018

walterddr Dec 7, 2018 •

edited

walterddr Dec 7, 2018 •

edited

dianfu commented Dec 7, 2018 •

edited

sunjincheng121 Dec 14, 2018 •

edited