Sangkyoon Nam created FLINK-41496:
-------------------------------------
Summary: Python Table API thread mode overwrites ROW values inside
ARRAY and MAP
Key: FLINK-41496
URL: https://issues.apache.org/jira/browse/FLINK-41496
Project: Flink
Issue Type: Bug
Components: API / Python
Reporter: Sangkyoon Nam
In Python thread mode, the Table API converters can overwrite ROW values inside
an ARRAY or MAP with a later row's values, both for UDF arguments and for UDF
results. No exception is raised, and process mode returns the expected values.
I reproduced it on master at {{cac69274d4e}} and haven't tested released
versions.
Reproducer:
{code:python}
import sys
from pyflink.common import Row
from pyflink.table import DataTypes, EnvironmentSettings, TableEnvironment
from pyflink.table.udf import udf
table_env = TableEnvironment.create(EnvironmentSettings.in_batch_mode())
config = table_env.get_config()
config.set("python.execution-mode", "thread")
config.set("python.executable", sys.executable)
config.set("python.client.executable", sys.executable)
child = DataTypes.ROW([DataTypes.FIELD("x", DataTypes.BIGINT())])
table = table_env.sql_query(
"SELECT ARRAY[CAST(ROW(1) AS ROW<x BIGINT>), CAST(ROW(2) AS ROW<x BIGINT>)]
AS a")
identity = udf(lambda v: v, result_type=DataTypes.ARRAY(child))
constant = udf(lambda v: [Row(10), Row(20)], result_type=DataTypes.ARRAY(child))
with table.select(identity(table.a), constant(table.a)).execute().collect() as
rows:
print([tuple(row) for row in rows])
{code}
Expected output (also what process mode prints):
{noformat}
[([<Row(1)>, <Row(2)>], [<Row(10)>, <Row(20)>])]
{noformat}
Thread mode:
{noformat}
[([<Row(2)>, <Row(2)>], [<Row(20)>, <Row(20)>])]
{noformat}
{{MAP<STRING, ROW<...>>}} values are affected too, and an
{{ARRAY<ARRAY<ROW<...>>>}} of {{[[Row(1), Row(2)], [Row(3)]]}} comes back as
{{[[Row(3), Row(3)], [Row(3)]]}}.
Cause: {{RowDataConverter}} in
{{org.apache.flink.table.runtime.typeutils.PythonTypeUtils}} returns the same
{{Row}} / {{Object[]}} on every call (reuse added in FLINK-28549, 1.16.0). That
is safe for a top-level row, but {{ArrayDataConverter}} and
{{MapDataConverter}} store each converted element, so later elements overwrite
earlier ones.
FLINK-41228 fixed the same issue in the DataStream converters by disabling
reuse below arrays, lists and maps. I have a patch applying the same approach
to the Table API converters, with tests in {{PythonTypeUtilsTest}} and
{{test_udf.py}}. It's based on my FLINK-41426 null-handling fix. Could you
assign this to me?
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
