Yaoxuan Wu created FLINK-41358:
----------------------------------
Summary: 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.2.1, 2.3.0
Reporter: Yaoxuan Wu
{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)
