Luo Zhengjie created FLINK-41199:
------------------------------------
Summary: 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
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)
