Yaoxuan Wu created FLINK-41507:
----------------------------------
Summary: Streaming MAX / MIN above a LEFT JOIN with append-only
inputs are planned without retraction and keep retracted values
Key: FLINK-41507
URL: https://issues.apache.org/jira/browse/FLINK-41507
Project: Flink
Issue Type: Bug
Components: Table SQL / Planner
Reporter: Yaoxuan Wu
{code:python}
from collections import Counter
from pyflink.common import RowKind
from pyflink.table import TableEnvironment, EnvironmentSettings, DataTypes
INT = lambda n: DataTypes.ROW([DataTypes.FIELD(n, DataTypes.INT())])
def final(tbl):
c = Counter()
for r in tbl.execute().collect():
c[tuple(r)] += 1 if r.get_row_kind() in (RowKind.INSERT,
RowKind.UPDATE_AFTER) else -1
return sorted(k for k, v in c.items() for _ in range(v))
for mode in ('batch', 'streaming'):
s = EnvironmentSettings.in_batch_mode() if mode == 'batch' else
EnvironmentSettings.in_streaming_mode()
t_env = TableEnvironment.create(s)
t_env.get_config().set('parallelism.default', '1')
t_env.create_temporary_view('a', t_env.from_elements([(3,), (-4,), (7,)],
INT('x')))
t_env.create_temporary_view('b', t_env.from_elements([(7,), (3,)],
INT('y')))
# rows of a without a match in b: only -4
q = "SELECT MAX(x), COUNT(*), SUM(x) FROM (SELECT x FROM a LEFT JOIN b ON x
= y WHERE y IS NULL)"
print(mode, final(t_env.sql_query(q)))
# batch [(-4, 1, -4)]
# streaming [(3, 1, -4)]
{code}
Same on 2.0.1, 2.2.1, 2.3.0 and a 2.4-SNAPSHOT build. The left join first emits
the null-padded rows {{(3, NULL)}} and {{(7, NULL)}}, then retracts them when
the matching {{y}} arrives. {{WHERE y IS NULL}} passes both the inserts and the
retractions. COUNT and SUM in the same {{GroupAggregate}} apply the
retractions, but MAX keeps a retracted value. MIN has the same problem: with
{{c = (-9), (-4)}} and {{d = (-9)}}, {{SELECT MIN(x) ... FROM c LEFT JOIN d ...
WHERE y IS NULL}} returned -9 on 2.3.0. That run depends on arrival order: on
2.0 and the 2.4 snapshot it returned -4 in the same run. {{LEFT JOIN ... WHERE
y IS NULL}} is the usual way to write an anti join. The Table API
{{x.in_(sub).is_not_true}} is planned as the same left join, which is where we
first saw this.
Cause (planner, master):
- {{AggregateUtil.deriveAggCallNeedRetractions}} plans {{MAX}} without
retraction ({{MaxWithRetract}} -> {{Max}}) when {{getRelModifiedMonotonicity}}
reports its output field as INCREASING, and {{MIN}} when it reports DECREASING.
- {{FlinkRelMdModifiedMonotonicity.getRelModifiedMonotonicity(Join)}} returns
{{null}} ("contains delete") only for ANTI joins and for inputs that contain
deletes. For other join types with append-only inputs and constant join keys,
it returns the inputs' field monotonicities (all CONSTANT). An outer join emits
deletes even when both inputs are insert-only: it retracts null-padded rows. So
LEFT / RIGHT / FULL joins should report {{null}} here.
- The Calc ({{IS NULL(y)}}) keeps CONSTANT, and on a CONSTANT input the
aggregate reports MAX as INCREASING and MIN as DECREASING, so both lose their
retraction. The changelog modes in the plan are right ({{Join ...
changelogMode=[I,UB,UA,D]}}, {{GroupAggregate ... changelogMode=[I,UA,D]}});
only the per-call retraction flags are wrong.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
