[
https://issues.apache.org/jira/browse/FLINK-41292?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18141140#comment-18141140
]
Sangkyoon Nam commented on FLINK-41292:
---------------------------------------
I reproduced this, and I'd like to fix it.
On a Flink 2.5-SNAPSHOT build with Python 3.11 in batch mode, {{SELECT
pandas_count(v) FROM t WHERE v > 5}} returns {{[]}} in 3 of 3 runs of a
standalone script. The built-in {{COUNT(v)}} on the same query returns
{{[Row(0)]}} in 3 of 3 runs. With {{GROUP BY}}, an empty input gives {{[]}} as
expected. Without {{GROUP BY}}, a non-empty input returns {{[Row(2)]}}.
The fix is in {{BatchArrowPythonGroupAggregateFunctionOperator}} and
{{BatchExecPythonGroupAggregate}}. The planner tells the operator when the
grouping is empty. If no row arrived, the operator calls Python once with an
empty batch in {{endInput()}}.
I ran {{python -m pytest}} on
{{flink-python/pyflink/table/tests/test_pandas_udaf.py}} from the
{{flink-python}} directory with {{-k test_group_aggregate_without_keys}}. The
new test {{test_group_aggregate_without_keys_on_empty_input}} fails 3 of 3
before the fix (expected {{[Row(0)]}}, got {{[]}}) and passes 3 of 3 after.
With the fix, the whole file passes (18 tests, one run). I haven't run the Java
unit tests.
Could a committer assign this ticket to me?
> 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.0.1, 2.3.0
> Reporter: Yaoxuan Wu
> Priority: Major
> Labels: correctness
>
> {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)
