[
https://issues.apache.org/jira/browse/FLINK-41290?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18136972#comment-18136972
]
Sangkyoon Nam commented on FLINK-41290:
---------------------------------------
I reproduced this in both process and thread mode, and a two-line fix is ready
locally. There's no PR yet.
>From reading the code, the cause is in two places.
>{{PythonTableFunctionOperator}} skips the row kind on the LEFT padding branch.
>{{EmbeddedPythonTableFunctionOperator.processElement}} never sets it. The fix
>adds one {{reuseJoinedRow.setRowKind(...)}} line in each. The tests are in
>{{UserDefinedTableFunctionRowKindTests}} in {{test_udtf.py}}, and both the
>stream and thread test classes run them.
I ran this:
{code}
python -m pytest pyflink/table/tests/test_udtf.py -k 'RowKind or keeps_row_kind
or empty_result'
{code}
Before the fix (the two Java files reverted and rebuilt): 3 failed, on each of
3 runs. After the fix: 6 passed, on each of 3 runs. The full {{test_udtf.py}}
gives 16 passed.
Could a committer assign this ticket to me so I can open the PR?
> PyFlink streaming: Python UDTF in a lateral join loses the row kind of the
> input (NULL-padded LEFT JOIN rows; every row in thread mode)
> ---------------------------------------------------------------------------------------------------------------------------------------
>
> Key: FLINK-41290
> URL: https://issues.apache.org/jira/browse/FLINK-41290
> Project: Flink
> Issue Type: Bug
> Components: API / Python
> Affects Versions: 2.0.1, 2.3.0, 2.2.1, 2.4.0
> Reporter: Yaoxuan Wu
> Priority: Major
> Labels: correctness
>
> {code:python}
> from pyflink.table import EnvironmentSettings, TableEnvironment, DataTypes as
> D
> from pyflink.table.udf import udtf
> t_env = TableEnvironment.create(EnvironmentSettings.in_streaming_mode())
> t_env.get_config().set('parallelism.default', '1')
> t_env.create_temporary_view('t', t_env.from_elements([(1, 10), (1, 20), (2,
> 30)],
> D.ROW([D.FIELD('k', D.INT()), D.FIELD('v', D.INT())])))
> @udtf(result_types=[D.INT()])
> def explode(a):
> for x in a or []:
> yield x,
> t_env.create_temporary_function('explode', explode)
> agg = "(SELECT k, CASE WHEN COUNT(*) > 1 THEN ARRAY[k] END AS a, COUNT(*) AS
> c FROM t GROUP BY k)"
> t_env.execute_sql(f"SELECT k, c, x FROM {agg} LEFT JOIN LATERAL
> TABLE(explode(a)) AS u(x) ON TRUE").print()
> # expected (as with LEFT JOIN UNNEST(a)): +I[1, 1, NULL], -U[1, 1, NULL],
> +U[1, 2, 1], +I[2, 1, NULL]
> # actual: +I[1, 1, NULL], +I[1, 1, NULL], +U[1, 2, 1], +U[2, 1, NULL]
> {code}
> The NULL-padded rows of the LEFT lateral join get a stale row kind: the
> {{-U}} is emitted as {{+I}} (the final result keeps (1, 1, NULL) twice) and
> the {{+I}} of key 2 as {{+U}}. With {{python.execution-mode = thread}} the
> inner lateral join ({{..., LATERAL TABLE(explode(a))}} over {{SELECT k,
> ARRAY[k] AS a, COUNT(*) AS c FROM t GROUP BY k}}) is wrong too: every output
> row is {{+I}}, so the retracted (1, 1, 1) stays in the result.
> Cause: {{PythonTableFunctionOperator.emitResult}} calls
> {{reuseJoinedRow.setRowKind(input.getRowKind())}} only on the branch that
> emits a UDTF result; the LEFT-join padding branch ({{joinType ==
> FlinkJoinType.LEFT && !hasJoined}}) emits {{reuseJoinedRow.replace(input,
> <NULL row>)}} without it, and {{JoinedRowData.replace}} does not copy the row
> kind, so the padded row carries whatever kind the reused row had last.
> {{EmbeddedPythonTableFunctionOperator.processElement}} (thread mode) never
> sets the row kind on either branch. The native correlate copies the input row
> kind to every output row.
> Reproduces on 2.0.1, 2.2.1, 2.3.0 and master.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
