[
https://issues.apache.org/jira/browse/FLINK-41199?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
ASF GitHub Bot updated FLINK-41199:
-----------------------------------
Labels: pull-request-available (was: )
> PyFlink result collection fails for NULL values inside nested rows and
> collections
> ----------------------------------------------------------------------------------
>
> Key: FLINK-41199
> URL: https://issues.apache.org/jira/browse/FLINK-41199
> Project: Flink
> Issue Type: Bug
> Components: API / Python
> Reporter: Luo Zhengjie
> Priority: Major
> Labels: pull-request-available
>
> h2. Problem
> PyFlink TableResult.collect() cannot decode the existing NULL marker when it
> occurs inside a ROW, ARRAY or MAP.
> PythonBridgeUtils.getPickledBytesFromJavaObject represents NULL as an empty
> byte array at every nesting level. CloseableIterator handles this only for
> top-level fields; pickled_bytes_to_python_converter attempts
> pickle.loads(b'') for nested scalars (EOFError), or accesses data[0] for a
> NULL ROW (IndexError).
> h2. Minimal decoder reproducer
> {code:python}
> import pickle
> from pyflink.table import DataTypes
> from pyflink.table.utils import pickled_bytes_to_python_converter
> pickled_bytes_to_python_converter(
> pickle.dumps([b'', pickle.dumps(7)]),
> DataTypes.ARRAY(DataTypes.INT()))
> # Expected: [None, 7]; actual: EOFError
> pickled_bytes_to_python_converter(
> pickle.dumps([b'']),
> DataTypes.ARRAY(DataTypes.ROW([DataTypes.FIELD('value',
> DataTypes.STRING())])))
> # Expected: [None]; actual: IndexError
> {code}
> h2. SQL regression
> {code:python}
> from pyflink.table import EnvironmentSettings, TableEnvironment
> t_env = TableEnvironment.create(EnvironmentSettings.in_batch_mode())
> result = t_env.execute_sql(
> "SELECT CAST(ROW(CAST(NULL AS STRING), ARRAY['row']) "
> "AS ROW<text STRING,tags ARRAY<STRING>>) AS nested, "
> "ARRAY[CAST(NULL AS INT), 7] AS numbers, "
> "MAP['key', CAST(NULL AS INT)] AS attributes, "
> "CAST(NULL AS STRING) AS missing")
> with result.collect() as rows:
> print(list(rows))
> {code}
> Expected: one row containing Row(None, ['row']), [None, 7], {'key': None},
> and None.
> h2. Proposed fix and verification
> Recognize the existing NULL marker at the recursive converter entry and
> delegate top-level fields to it. Properly pickled empty strings, bytes and
> collections remain non-NULL. Keep the existing Row(*fields) implementation:
> this is not a repeat of FLINK-37616. No Java bridge, wire-format, public API,
> dependency or state-serializer changes.
> Seven focused converter tests reproduced nine NULL-related error subcases
> without the fix and pass with it; the same seven tests and Flake8 pass on
> community master 0addadb5caeead0aedaa780443cce89ae6ff0487. An SQL collect
> regression is included in the proposed patch, but a complete community
> Maven/integration run has not yet passed locally. These are not claims of
> full integration verification.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
