[FLINK-24704][table] Fix exception when the input record loses monotonicity on the sort key field of UpdatableTopNFunction #17605

lincoln-lil · 2021-10-29T12:09:07Z

What is the purpose of the change

Fix the exception when the input record loses monotonicity on the sort key field of UpdatableTopNFunction

Verifying this change

This change is already covered by RankHarnessTest

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, Kubernetes/Yarn, ZooKeeper: (no)
- The S3 file system connector: (no)

Documentation

Does this pull request introduce a new feature? (no)

…nicity on the sort key field of UpdatableTopNFunction

flinkbot · 2021-10-29T12:11:53Z

Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community
to review your pull request. We will use this comment to track the progress of the review.

Automated Checks

Last check on commit 10ca9cf (Fri Oct 29 12:11:53 UTC 2021)

Warnings:

No documentation files were touched! Remember to keep the Flink docs up to date!
This pull request references an unassigned Jira ticket. According to the code contribution guide, tickets need to be assigned before starting with the implementation work.

_{Mention the bot in a comment to re-run the automated checks.}

Review Progress

❓ 1. The [description] looks good.
❓ 2. There is [consensus] that the contribution should go into to Flink.
❓ 3. Needs [attention] from.
❓ 4. The change fits into the overall [architecture].
❓ 5. Overall code [quality] is good.

Please see the Pull Request Review Guide for a full explanation of the review process.

The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot approve description to approve one or more aspects (aspects: description, consensus, architecture and quality)
@flinkbot approve all to approve all aspects
@flinkbot approve-until architecture to approve everything until architecture
@flinkbot attention @username1 [@username2 ..] to require somebody's attention
@flinkbot disapprove architecture to remove an approval you gave earlier

flinkbot · 2021-10-29T12:12:41Z

CI report:

c16aa62 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

lincoln-lil · 2021-10-29T17:14:15Z

Some tests failed, I'll fix it.

lincoln-lil · 2021-10-31T12:35:46Z

@flinkbot run azure

JingsongLi

Thanks @lincoln-lil for your contribution, left some comments.

JingsongLi · 2021-11-03T01:35:22Z

...ntime/src/main/java/org/apache/flink/table/runtime/operators/rank/UpdatableTopNFunction.java

+    private final boolean lenient = true;
+
+    // data converter for logging only.
+    private final DataStructureConverter rowConverter;


This will break compatibility. We can just introduce two transient fields.

Good catch! transient is better.

JingsongLi · 2021-11-03T01:38:11Z

...-planner/src/test/scala/org/apache/flink/table/planner/runtime/harness/RankHarnessTest.scala

+    testHarness.processElement(binaryRecord(INSERT, "f", 1: JInt, 70: JInt))
+
+    testHarness.processElement(binaryRecord(UPDATE_AFTER, "b", 1: JInt, 10: JInt))
+


Just left one line is OK

tsreaper · 2021-11-03T03:22:07Z

...ntime/src/main/java/org/apache/flink/table/runtime/operators/rank/UpdatableTopNFunction.java

+                currentRow = rowKeyMap.get(rowKey).row;
+                if (oldRank <= currentRank) {
+                    if (currentRank == oldRank) {
+                        checkArgument(0 == sortKeyComparator.compare(curSortKey, oldSortKey));


Why is this true? It might be that the oldSortKey is unique and after the change it does not exist anymore.

It's a wrong check here, and it's unnecessary after think it over, so I'll remove it.

lincoln-lil · 2021-11-03T08:14:18Z

@flinkbot run azure

tsreaper · 2021-11-04T02:37:43Z

...ntime/src/main/java/org/apache/flink/table/runtime/operators/rank/UpdatableTopNFunction.java

+                if (oldRank <= currentRank) {
+                    if (currentRank == oldRank) {
+                        collectUpdateBefore(out, oldRow.row, oldRank);
+                    } else {
+                        collectUpdateBefore(out, prevRow, currentRank);
+                        collectUpdateAfter(out, prevRow, currentRank - 1);
+                        if (currentRank == newRank) {
+                            collectUpdateAfter(out, newRow, currentRank);
+                        }
+                    }
+                }


What if newSortKey > oldSortKey but oldRank == currentRank? The update after message will be lost. Please also add a test about this after fixing it.

It seems that a more proper solution to this is not to send any message if oldRank == currentRank.

Good catch! will add case to cover it.

tsreaper · 2021-11-04T02:46:08Z

...ntime/src/main/java/org/apache/flink/table/runtime/operators/rank/UpdatableTopNFunction.java

+        while (iterator.hasNext() && currentRank <= newRank) {
+            Map.Entry<RowData, Collection<RowData>> entry = iterator.next();
+            Collection<RowData> rowKeys = entry.getValue();
+            Iterator<RowData> rowKeyIter = rowKeys.iterator();
+            while (rowKeyIter.hasNext()) {
+                RowData rowKey = rowKeyIter.next();
+                currentRank += 1;
+                currentRow = rowKeyMap.get(rowKey).row;
+                if (oldRank <= currentRank) {
+                    if (currentRank == oldRank) {
+                        collectUpdateBefore(out, oldRow.row, oldRank);
+                    } else {
+                        collectUpdateBefore(out, prevRow, currentRank);
+                        collectUpdateAfter(out, prevRow, currentRank - 1);
+                        if (currentRank == newRank) {
+                            collectUpdateAfter(out, newRow, currentRank);
+                        }
+                    }
+                }
+                prevRow = currentRow;
+            }
+        }
+    }


This piece of algorithm seems awkward to me. Consider modifying it to:

while (iterator.hasNext() && currentRank < newRank) { // ... while (rowKeyIter.hasNext()) { // ... if (oldRank <= currentRank) { collectUpdateBefore(out, currentRow, currentRank + 1); collectUpdateAfter(out, currentRow, currentRank); } } } collectUpdateBefore(out, oldRow.row, oldRank); collectUpdateAfter(out, newRow, newRank);

so that there is no prevRow thingy. It is misleading to see a prevRow and a currentRank sending within the same message.

Cool! It's more simpler and readable. A small change is to emit the UB of old row before the following changes.

tsreaper · 2021-11-04T02:55:28Z

...-planner/src/test/scala/org/apache/flink/table/planner/runtime/harness/RankHarnessTest.scala

@@ -220,4 +221,74 @@ class RankHarnessTest(mode: StateBackendMode) extends HarnessTestBase(mode) {
    assertor.assertOutputEqualsSorted("result mismatch", expectedOutput, result)
    testHarness.close()
  }
+
+  @Test
+  def testUpdateRankWithRowNumber(): Unit = {


Add more test scenarios.

Calculate top 5 but there are more than 5 candidates.

Sort key drops but ranking does not change.

Sort key drops but does not drop to the last ranking.

Calculate top 5, 7 candidates, previous rank 3 drops to rank 6 (but it is still "rank 5").

Thanks for your review, I'll update it.

tsreaper

Looks good to me. Let's wait for the CI to pass.

JingsongLi

Looks good to me!

…s monotonicity on the sort key field of UpdatableTopNFunction This closes apache#17605

…s monotonicity on the sort key field of UpdatableTopNFunction This closes #17605

…s monotonicity on the sort key field of UpdatableTopNFunction This closes apache#17605

[FLINK-24704][table] Fix exception when the input record loses monoto…

10ca9cf

…nicity on the sort key field of UpdatableTopNFunction

rmetzger added the component=TableSQL/Runtime label Oct 29, 2021

simplify

c208976

fix tests

2e9cea2

JingsongLi requested changes Nov 3, 2021

View reviewed changes

tsreaper reviewed Nov 3, 2021

View reviewed changes

address review comments

0056d2c

tsreaper requested changes Nov 4, 2021

View reviewed changes

tsreaper reviewed Nov 4, 2021

View reviewed changes

address comments: add more tests

c16aa62

tsreaper approved these changes Nov 4, 2021

View reviewed changes

JingsongLi approved these changes Nov 5, 2021

View reviewed changes

JingsongLi merged commit 33aaabe into apache:master Nov 5, 2021

lincoln-lil added a commit to lincoln-lil/flink that referenced this pull request Nov 9, 2021

[FLINK-24704][table-runtime] Fix exception when the input record lose…

1d59aed

…s monotonicity on the sort key field of UpdatableTopNFunction This closes apache#17605

JingsongLi pushed a commit that referenced this pull request Nov 10, 2021

[FLINK-24704][table-runtime] Fix exception when the input record lose…

d31cfb9

…s monotonicity on the sort key field of UpdatableTopNFunction This closes #17605

niklassemmler pushed a commit to niklassemmler/flink that referenced this pull request Feb 3, 2022

[FLINK-24704][table-runtime] Fix exception when the input record lose…

d5ec905

…s monotonicity on the sort key field of UpdatableTopNFunction This closes apache#17605

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-24704][table] Fix exception when the input record loses monotonicity on the sort key field of UpdatableTopNFunction #17605

[FLINK-24704][table] Fix exception when the input record loses monotonicity on the sort key field of UpdatableTopNFunction #17605

lincoln-lil commented Oct 29, 2021

flinkbot commented Oct 29, 2021

flinkbot commented Oct 29, 2021 •

edited

lincoln-lil commented Oct 29, 2021 •

edited

lincoln-lil commented Oct 31, 2021

JingsongLi left a comment

JingsongLi Nov 3, 2021

lincoln-lil Nov 3, 2021

JingsongLi Nov 3, 2021

tsreaper Nov 3, 2021

lincoln-lil Nov 3, 2021

lincoln-lil commented Nov 3, 2021

tsreaper Nov 4, 2021

tsreaper Nov 4, 2021

lincoln-lil Nov 4, 2021

tsreaper Nov 4, 2021 •

edited

lincoln-lil Nov 4, 2021

tsreaper Nov 4, 2021

lincoln-lil Nov 4, 2021

tsreaper left a comment •

edited

JingsongLi left a comment

		testHarness.processElement(binaryRecord(INSERT, "f", 1: JInt, 70: JInt))

		testHarness.processElement(binaryRecord(UPDATE_AFTER, "b", 1: JInt, 10: JInt))

[FLINK-24704][table] Fix exception when the input record loses monotonicity on the sort key field of UpdatableTopNFunction #17605

[FLINK-24704][table] Fix exception when the input record loses monotonicity on the sort key field of UpdatableTopNFunction #17605

Conversation

lincoln-lil commented Oct 29, 2021

What is the purpose of the change

Verifying this change

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

Documentation

flinkbot commented Oct 29, 2021

Automated Checks

Review Progress

flinkbot commented Oct 29, 2021 • edited

CI report:

lincoln-lil commented Oct 29, 2021 • edited

lincoln-lil commented Oct 31, 2021

JingsongLi 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

lincoln-lil commented Nov 3, 2021

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tsreaper Nov 4, 2021 • 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

tsreaper left a comment • edited

Choose a reason for hiding this comment

JingsongLi left a comment

Choose a reason for hiding this comment

flinkbot commented Oct 29, 2021 •

edited

lincoln-lil commented Oct 29, 2021 •

edited

tsreaper Nov 4, 2021 •

edited

tsreaper left a comment •

edited