Yaoxuan Wu created FLINK-41292:
----------------------------------
Summary: PyFlink batch: pandas UDAF in an aggregation without
GROUP BY returns no row for an empty input
Key: FLINK-41292
URL: https://issues.apache.org/jira/browse/FLINK-41292
Project: Flink
Issue Type: Bug
Components: API / Python
Affects Versions: 2.3.0, 2.0.1
Reporter: Yaoxuan Wu
{code:python}
from pyflink.table import EnvironmentSettings, TableEnvironment, DataTypes as D
from pyflink.table.udf import udaf
t_env = TableEnvironment.create(EnvironmentSettings.in_batch_mode())
t_env.create_temporary_view('t', t_env.from_elements([(1,), (2,)],
D.ROW([D.FIELD('v', D.INT())])))
t_env.create_temporary_function('pandas_count',
udaf(lambda s: len(s), result_type=D.BIGINT(), func_type='pandas'))
t_env.execute_sql("SELECT COUNT(v) FROM t WHERE v > 5").print() # one
row: 0
t_env.execute_sql("SELECT pandas_count(v) FROM t WHERE v > 5").print() #
expected one row: 0; actual: Empty set
{code}
An aggregation without GROUP BY returns exactly one row, also for an empty
input; the native aggregate does, the pandas UDAF returns none.
Cause: the batch pandas aggregate is executed by
{{BatchArrowPythonGroupAggregateFunctionOperator}} (planned by
{{BatchExecPythonGroupAggregate}}), also when the grouping key is empty. It
only produces output for groups it has seen: {{bufferInput}} adds a group to
{{forwardedInputQueue}} when a row arrives, {{invokeCurrentBatch}} calls Python
only {{if (currentBatchCount > 0)}}, and {{emitResult}} emits one row per
result. With no input row there is no group and no output. The native keyless
aggregate instead emits the initial accumulator values at end of input when it
saw no row ({{AggWithoutKeysCodeGenerator}}, final phase, {{!hasInput}} branch
of the end-input code).
Reproduces on 2.0.1, 2.2.1, 2.3.0 and master.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
