Yaoxuan Wu created FLINK-41290:
----------------------------------
Summary: 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.2.1, 2.3.0, 2.0.1, 2.4.0
Reporter: Yaoxuan Wu
{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)
