[
https://issues.apache.org/jira/browse/FLINK-41358?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18140509#comment-18140509
]
Yahya Kisana commented on FLINK-41358:
--------------------------------------
Hi, was able to confirm bug, can I take this?
> Streaming OVER window ordered by a non-time column (RANGE frame) drops all
> equal rows from the aggregate when one of them is retracted
> --------------------------------------------------------------------------------------------------------------------------------------
>
> Key: FLINK-41358
> URL: https://issues.apache.org/jira/browse/FLINK-41358
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Runtime
> Affects Versions: 2.3.0, 2.2.1
> Reporter: Yaoxuan Wu
> Priority: Major
> Labels: correctness
>
> {code:python}
> from pyflink.table import EnvironmentSettings, TableEnvironment
> t_env = TableEnvironment.create(EnvironmentSettings.in_streaming_mode())
> t_env.get_config().set("parallelism.default", "1")
> t_env.create_temporary_view("emp_versions", t_env.from_elements(
> [(1, 0, "eng", 100), (2, 0, "eng", 100), (2, 1, "ops", 100)],
> ["id", "ver", "dept", "salary"]))
> # latest version per employee: employee 2 moves from eng to ops
> t_env.execute_sql("""
> CREATE TEMPORARY VIEW emp AS SELECT id, dept, salary FROM (
> SELECT *, ROW_NUMBER() OVER (PARTITION BY id ORDER BY ver DESC) AS rn
> FROM emp_versions)
> WHERE rn = 1""")
> t_env.execute_sql("""
> SELECT UPPER(dept) AS dept, salary, COUNT(*) OVER w AS cnt, SUM(salary)
> OVER w AS total
> FROM emp
> WINDOW w AS (PARTITION BY dept ORDER BY salary RANGE BETWEEN UNBOUNDED
> PRECEDING AND CURRENT ROW)
> """).print()
> {code}
> Output (2.2.1, 2.3.0 and 2.4-SNAPSHOT):
> {noformat}
> | +I | ENG | 100 | 1 | 100 |
> | -U | ENG | 100 | 1 | 100 |
> | +U | ENG | 100 | 2 | 200 |
> | +I | ENG | 100 | 2 | 200 |
> | -U | ENG | 100 | 2 | 200 |
> | +U | ENG | 100 | 0 | NULL | <- employee 1 is still in eng: expected 1, 100
> | -D | ENG | 100 | 2 | 200 |
> | +U | OPS | 100 | 1 | 100 |
> {noformat}
> The final result is (ENG, 100, 0, NULL), (OPS, 100, 1, 100); it should be
> (ENG, 100, 1, 100), (OPS, 100, 1, 100). With {{ROWS BETWEEN UNBOUNDED
> PRECEDING AND CURRENT ROW}} the result is right.
> The OVER operator receives {{(dept, salary)}} only: {{UPPER(dept)}} is
> computed below the window and {{id}} is pruned. So its input is +I (eng,
> 100), +I (eng, 100), -U (eng, 100), two equal rows and the retraction of one
> of them. After the retraction, the remaining row is aggregated as if both
> rows were gone. The same happens whenever a retracted row equals another row
> in the same partition with the same ORDER BY value on all columns the
> operator stores; every aggregate (COUNT, SUM, AVG, MIN, MAX, ...) of that
> peer group and of all later sort keys is affected.
> Cause:
> {{NonTimeRangeUnboundedPrecedingFunction.reAccumulateIdsAndGetRemoveIndex}}
> (release-2.3.0 lines 385-395, unchanged on master) re-accumulates the peer
> group without the retracted row, but skips every row equal to it:
> {code:java}
> for (int j = 0; j < ids.size(); j++) {
> RowData curValue = valueMapState.get(ids.get(j));
> if (valueEqualiser.equals(curValue, delRow)) {
> removeIndex = j;
> } else {
> aggFuncs.accumulate(curValue);
> }
> }
> {code}
> {{removeFromSortedList}} then removes only one id
> ({{ids.remove(removeIndex)}}, line 361), stores this accumulator for the sort
> key (line 369) and carries it into the following sort keys
> ({{processRemainingElements}}, line 373). The ROWS variant,
> {{NonTimeRowsUnboundedPrecedingFunction.findIndexOfIdToBeRemoved}}, stops at
> the first equal row. Skipping only the first match, e.g. {{if (removeIndex ==
> -1 && valueEqualiser.equals(curValue, delRow))}}, fixes it.
> The operator was added in 2.1.0 (FLINK-19059). Related, different:
> FLINK-41099 (RANK ties in the same operator).
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
