[FLINK-26819] AppendOnlyFirstNFunction produce wrong result when with offset and without rank number

• Type: Bug
• Status: Closed
• Priority: Major
• Resolution: Fixed
• Affects Version/s: 1.14.4, 1.15.0
• Fix Version/s: 1.16.0
• Component/s: Table SQL / Runtime
• Labels: pull-request-available
• Assignee: lincoln lee
• Reporter: lincoln lee
• Created: 23/Mar/22 08:05
• Updated: 23/Mar/22 14:21
• Resolved: 23/Mar/22 14:21

Description

FLINK-19896 introduced an optimized AppendOnlyFirstNFunction, but produces wrong result when a constant rank with offset and without rank number.

In test case `AppendOnlyFirstNFunctionTest.testConstantRankRangeWithOffset` produces 4 output while 2 output is expected because only want rank range be [2,2]

        AbstractTopNFunction func =
                createFunction(RankType.ROW_NUMBER, new ConstantRankRange(2, 2), true, false);
        OneInputStreamOperatorTestHarness<RowData, RowData> testHarness = createTestHarness(func);
        testHarness.open();
        testHarness.processElement(insertRecord("book", 2L, 12));
        testHarness.processElement(insertRecord("book", 2L, 19));
        testHarness.processElement(insertRecord("book", 2L, 11));
        testHarness.processElement(insertRecord("fruit", 1L, 33));
        testHarness.processElement(insertRecord("fruit", 1L, 44));
        testHarness.processElement(insertRecord("fruit", 1L, 22));
        testHarness.close();

        List<Object> expectedOutput = new ArrayList<>();
        expectedOutput.add(insertRecord("book", 2L, 12));
        expectedOutput.add(insertRecord("book", 2L, 19));
        expectedOutput.add(insertRecord("fruit", 1L, 33));
        expectedOutput.add(insertRecord("fruit", 1L, 44));
       ...

we should fix this.

Issue Links

• links to: GitHub Pull Request #19212