This is an automated email from the ASF dual-hosted git repository. dianfu pushed a commit to branch release-2.2 in repository https://gitbox.apache.org/repos/asf/flink.git
commit 4dfff5e41001afe5565df320c5cbe21ba22ed163 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 505d1dbd26f..66161e37564 100644 --- a/flink-python/pyflink/table/tests/test_table_environment_api.py +++ b/flink-python/pyflink/table/tests/test_table_environment_api.py @@ -669,6 +669,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 ed92ad230af..96005bc8141 100644 --- a/flink-python/pyflink/table/utils.py +++ b/flink-python/pyflink/table/utils.py @@ -112,6 +112,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())
