This is an automated email from the ASF dual-hosted git repository.
dianfu pushed a commit to branch release-2.4
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-2.4 by this push:
new 7b23995d7d2 [FLINK-41199][python] Decode NULL values in nested
collected results (#29605)
7b23995d7d2 is described below
commit 7b23995d7d28f6bacd3ee3ae2b436bbee93bc258
Author: Luo Zhengjie <[email protected]>
AuthorDate: Tue Oct 27 11:05:10 2026 +0800
[FLINK-41199][python] Decode NULL values in nested collected results
(#29605)
---
flink-python/pyflink/table/table_result.py | 8 +--
.../table/tests/test_table_environment_api.py | 11 +++
flink-python/pyflink/table/tests/test_utils.py | 81 ++++++++++++++++++++++
flink-python/pyflink/table/utils.py | 3 +
4 files changed, 97 insertions(+), 6 deletions(-)
diff --git a/flink-python/pyflink/table/table_result.py
b/flink-python/pyflink/table/table_result.py
index 2073701417a..caeedf0e758 100644
--- a/flink-python/pyflink/table/table_result.py
+++ b/flink-python/pyflink/table/table_result.py
@@ -211,12 +211,8 @@ class CloseableIterator(object):
row_kind = RowKind(int.from_bytes(pickle_bytes[0], byteorder='big',
signed=False))
pickle_bytes = list(pickle_bytes[1:])
field_data = zip(pickle_bytes, self._data_types)
- fields = []
- for data, field_type in field_data:
- if len(data) == 0:
- fields.append(None)
- else:
- fields.append(pickled_bytes_to_python_converter(data,
field_type))
+ fields = [pickled_bytes_to_python_converter(data, field_type)
+ for data, field_type in field_data]
result_row = Row(*fields)
result_row.set_row_kind(row_kind)
return result_row
diff --git a/flink-python/pyflink/table/tests/test_table_environment_api.py
b/flink-python/pyflink/table/tests/test_table_environment_api.py
index 492c20bb8ff..765b9329e84 100644
--- a/flink-python/pyflink/table/tests/test_table_environment_api.py
+++ b/flink-python/pyflink/table/tests/test_table_environment_api.py
@@ -786,6 +786,17 @@ class DataStreamConversionTestCases(PyFlinkUTTestCase):
class StreamTableEnvironmentTests(PyFlinkStreamTableTestCase):
+ def test_collect_nullable_nested_values(self):
+ result = self.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:
+ self.assertEqual([Row(Row(None, ['row']), [None, 7], {'key':
None}, None)],
+ list(rows))
+
def test_collect_with_retract(self):
expected_row_kinds = [RowKind.INSERT, RowKind.UPDATE_BEFORE,
RowKind.UPDATE_AFTER,
RowKind.INSERT, RowKind.UPDATE_BEFORE,
RowKind.UPDATE_AFTER]
diff --git a/flink-python/pyflink/table/tests/test_utils.py
b/flink-python/pyflink/table/tests/test_utils.py
new file mode 100644
index 00000000000..14c2424e27a
--- /dev/null
+++ b/flink-python/pyflink/table/tests/test_utils.py
@@ -0,0 +1,81 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+
+import pickle
+import unittest
+
+from pyflink.common import Row
+from pyflink.table import DataTypes
+from pyflink.table.utils import pickled_bytes_to_python_converter
+
+
+class PickledBytesConverterTests(unittest.TestCase):
+
+ def test_null_marker(self):
+ # PythonBridgeUtils encodes NULL as byte[0], including nested fields.
+ for field_type in (DataTypes.STRING(), DataTypes.INT(),
+ DataTypes.ARRAY(DataTypes.STRING()),
+ DataTypes.MAP(DataTypes.STRING(), DataTypes.INT()),
+ DataTypes.ROW([DataTypes.FIELD('value',
DataTypes.STRING())])):
+ with self.subTest(field_type=field_type):
+ self.assertIsNone(pickled_bytes_to_python_converter(b'',
field_type))
+
+ def test_nullable_row_fields(self):
+ result = pickled_bytes_to_python_converter(
+ [b'\x00', b'', pickle.dumps([pickle.dumps('row')])],
+ DataTypes.ROW([DataTypes.FIELD('text', DataTypes.STRING()),
+ DataTypes.FIELD('tags',
DataTypes.ARRAY(DataTypes.STRING()))]))
+ self.assertEqual(Row(None, ['row']), result)
+
+ def test_array_of_nullable_rows(self):
+ result = pickled_bytes_to_python_converter(
+ pickle.dumps([b'', [b'\x00', b''], [b'\x00', pickle.dumps('ok')]]),
+ DataTypes.ARRAY(DataTypes.ROW([DataTypes.FIELD('value',
DataTypes.STRING())])))
+ self.assertEqual([None, Row(None), Row('ok')], result)
+
+ def test_nullable_map_values(self):
+ for field_type in (DataTypes.INT(),
+ DataTypes.ROW([DataTypes.FIELD('value',
DataTypes.STRING())])):
+ with self.subTest(field_type=field_type):
+ result = pickled_bytes_to_python_converter(
+ pickle.dumps([[pickle.dumps('key')], [b'']]),
+ DataTypes.MAP(DataTypes.STRING(), field_type))
+ self.assertEqual({'key': None}, result)
+
+ def test_empty_values_are_not_null(self):
+ for value, field_type, expected in (
+ ('', DataTypes.STRING(), ''), (b'', DataTypes.BYTES(), b''),
+ ([], DataTypes.ARRAY(DataTypes.STRING()), []),
+ ([[], []], DataTypes.MAP(DataTypes.STRING(), DataTypes.INT()),
{})):
+ with self.subTest(field_type=field_type):
+ self.assertEqual(expected, pickled_bytes_to_python_converter(
+ pickle.dumps(value), field_type))
+
+ def test_zero_and_false_are_not_null(self):
+ for value, field_type in ((0, DataTypes.INT()), (False,
DataTypes.BOOLEAN())):
+ with self.subTest(field_type=field_type):
+ self.assertEqual(value, pickled_bytes_to_python_converter(
+ pickle.dumps(value), field_type))
+
+ def test_invalid_nonempty_payload_is_rejected(self):
+ with self.assertRaises(pickle.UnpicklingError):
+ pickled_bytes_to_python_converter(b'not-a-pickle',
DataTypes.STRING())
+
+
+if __name__ == '__main__':
+ unittest.main()
diff --git a/flink-python/pyflink/table/utils.py
b/flink-python/pyflink/table/utils.py
index 68e23d310c4..0e644b2ae44 100644
--- a/flink-python/pyflink/table/utils.py
+++ b/flink-python/pyflink/table/utils.py
@@ -281,6 +281,9 @@ def to_expression_jarray(exprs):
@Internal()
def pickled_bytes_to_python_converter(data, field_type: DataType):
+ # PythonBridgeUtils encodes NULL as an empty byte array at every nesting
level.
+ if len(data) == 0:
+ return None
if isinstance(field_type, RowType):
row_kind = RowKind(int.from_bytes(data[0], byteorder='big',
signed=False))
data = zip(list(data[1:]), field_type.field_types())
