Yaoxuan Wu created FLINK-41097:
----------------------------------
Summary: AVG returns NULL for a group with a NULL value once
adaptive local hash aggregation switches to bypass mode
Key: FLINK-41097
URL: https://issues.apache.org/jira/browse/FLINK-41097
Project: Flink
Issue Type: Bug
Components: Table SQL / Runtime
Affects Versions: 2.3.0
Reporter: Yaoxuan Wu
{code:python}
from pyflink.table import EnvironmentSettings, TableEnvironment, DataTypes as D
from pyflink.common import Configuration
c = Configuration()
c.set_string('parallelism.default', '1')
c.set_string('table.optimizer.agg-phase-strategy', 'TWO_PHASE')
c.set_string('table.exec.local-hash-agg.adaptive.sampling-threshold', '1')
# switch to bypass early
c.set_string('table.exec.local-hash-agg.adaptive.distinct-value-rate-threshold',
'0.0')
t =
TableEnvironment.create(EnvironmentSettings.new_instance().in_batch_mode().with_configuration(c).build())
t.create_temporary_view('t', t.from_elements([(1, 5), (1, None), (2, 7)],
D.ROW([D.FIELD('g', D.INT()), D.FIELD('x', D.INT())])))
t.execute_sql("SELECT g, AVG(x) FROM t GROUP BY g").print()
# expected: (1, 5), (2, 7)
# actual: (1, NULL), (2, 7)
{code}
Once the adaptive local hash aggregate (FLINK-30542) switches to bypass mode, a
row whose AVG argument is NULL makes the group's AVG NULL. The forced
thresholds above only make the switch happen early: with the default settings
(sampling 500000 rows, distinct-value rate 0.5) the same happens, e.g. 600001
distinct groups with the NULL row last. SUM, COUNT and MAX are correct, and so
is AVG with {{table.exec.local-hash-agg.adaptive.enabled=false}} or one-phase
aggregation.
Cause: after the switch,
{{ProjectionCodeGenerator.genAdaptiveLocalHashAggValueProjectionExpr}} turns
each input row into AVG's partial buffer as {{(sum = CAST(x AS sumType), count
= x IS NULL ? 0 : 1)}}, so a NULL x gives sum = NULL.
{{AvgAggFunction.mergeExpressions}} merges with {{sum = sum +
mergeOperand(sum)}} without a NULL check (AVG's sum buffer is otherwise never
NULL), so the global AVG's sum becomes NULL.
{{SumAggFunction.mergeExpressions}} checks for NULL, which is why SUM is
correct. Projecting a NULL argument to 0 for AVG, or a NULL-safe merge, would
fix it.
Reproduces on 2.0.1, 2.2.1, 2.3.0 and master.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
