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 85067f0d8063a15abeb9905d0caa29bb46663e8c Author: Sangkyoon Nam <[email protected]> AuthorDate: Fri Oct 30 15:38:47 2026 +0900 [FLINK-41228][python] Fix thread-mode conversion of arrays, lists and maps (#29630) --- .../pyflink/datastream/tests/test_data_stream.py | 27 ++++++ .../pyflink/fn_execution/embedded/converters.py | 8 +- .../flink/streaming/api/utils/PythonTypeUtils.java | 98 +++++++++++++++------- .../streaming/api/utils/PythonTypeUtilsTest.java | 59 +++++++++++++ 4 files changed, 159 insertions(+), 33 deletions(-) diff --git a/flink-python/pyflink/datastream/tests/test_data_stream.py b/flink-python/pyflink/datastream/tests/test_data_stream.py index 37d806a7aed..224e57110fe 100644 --- a/flink-python/pyflink/datastream/tests/test_data_stream.py +++ b/flink-python/pyflink/datastream/tests/test_data_stream.py @@ -1346,6 +1346,33 @@ class EmbeddedDataStreamStreamTests(DataStreamStreamingTests, PyFlinkStreamingTe expected = ['(%s,%s)' % (index, total) for index in range(4) for total in [1, 3]] self.assert_equals_sorted(expected, self.test_sink.get_results()) + def test_rows_in_containers(self): + self.addCleanup(self.env.set_parallelism, self.env.get_parallelism()) + self.env.set_parallelism(1) + row_type = Types.ROW([Types.INT()]) + output_type = Types.ROW([Types.OBJECT_ARRAY(row_type), + Types.LIST(row_type), + Types.MAP(Types.STRING(), row_type), + Types.OBJECT_ARRAY(Types.TUPLE([Types.INT()])), + Types.OBJECT_ARRAY(row_type)]) + + ds = self.env.from_collection([1], type_info=Types.INT()).map( + lambda _: Row([Row(1), Row(2)], [Row(3), Row(4)], {'a': Row(5), 'b': Row(6)}, + [(7,), (8,)], None), + output_type=output_type) + # The second consumer keeps the map above from being fused with the map below, so the + # rows are converted in both directions. + ds.add_sink(self.test_sink) + ds.map(lambda r: str(([v[0] for v in r[0]], [v[0] for v in r[1]], + sorted((k, v[0]) for k, v in r[2].items()), + [v[0] for v in r[3]], r[4])), + output_type=Types.STRING()) \ + .add_sink(self.test_sink) + self.env.execute('test_rows_in_containers') + expected = ['+I[[+I[1], +I[2]], [+I[3], +I[4]], {a=+I[5], b=+I[6]}, [(7), (8)], null]', + "([1, 2], [3, 4], [('a', 5), ('b', 6)], [7, 8], None)"] + self.assert_equals_sorted(expected, self.test_sink.get_results(False)) + def test_metrics(self): ds = self.env.from_collection( [('ab', 'a', decimal.Decimal(1)), diff --git a/flink-python/pyflink/fn_execution/embedded/converters.py b/flink-python/pyflink/fn_execution/embedded/converters.py index 4d8dab5915e..f3fced096f2 100644 --- a/flink-python/pyflink/fn_execution/embedded/converters.py +++ b/flink-python/pyflink/fn_execution/embedded/converters.py @@ -162,9 +162,15 @@ class ArrayDataConverter(ListDataConverter): super(ArrayDataConverter, self).__init__(field_converter) def to_internal(self, value) -> IN: + if value is None: + return None + return tuple(super(ArrayDataConverter, self).to_internal(value)) def to_external(self, value) -> OUT: + if value is None: + return None + return tuple(super(ArrayDataConverter, self).to_external(value)) @@ -234,7 +240,7 @@ def from_type_info_proto(type_info): elif type_name in (type_info_name.BASIC_ARRAY, type_info_name.OBJECT_ARRAY): return ArrayDataConverter(from_type_info_proto(type_info.collection_element_type)) - elif type_info == type_info_name.LIST: + elif type_name == type_info_name.LIST: return ListDataConverter(from_type_info_proto(type_info.collection_element_type)) elif type_name == type_info_name.MAP: return DictDataConverter(from_type_info_proto(type_info.map_type_info.key_type), diff --git a/flink-python/src/main/java/org/apache/flink/streaming/api/utils/PythonTypeUtils.java b/flink-python/src/main/java/org/apache/flink/streaming/api/utils/PythonTypeUtils.java index fcb02bcdebf..09a4f208b0f 100644 --- a/flink-python/src/main/java/org/apache/flink/streaming/api/utils/PythonTypeUtils.java +++ b/flink-python/src/main/java/org/apache/flink/streaming/api/utils/PythonTypeUtils.java @@ -988,16 +988,21 @@ public class PythonTypeUtils { private static final long serialVersionUID = 1L; private final DataConverter[] fieldDataConverters; + private final boolean reuse; private final Row reuseRow; private final Object[] reuseExternalRow; private final Object[] reuseExternalRowData; RowDataConverter(DataConverter[] dataConverters) { + this(dataConverters, true); + } + + RowDataConverter(DataConverter[] dataConverters, boolean reuse) { this.fieldDataConverters = dataConverters; + this.reuse = reuse; this.reuseRow = new Row(fieldDataConverters.length); this.reuseExternalRowData = new Object[fieldDataConverters.length]; this.reuseExternalRow = new Object[2]; - this.reuseExternalRow[1] = reuseExternalRowData; } @SuppressWarnings("unchecked") @@ -1007,13 +1012,14 @@ public class PythonTypeUtils { return null; } + Row row = reuse ? reuseRow : new Row(fieldDataConverters.length); RowKind rowKind = RowKind.fromByteValue(((Long) value[0]).byteValue()); - reuseRow.setKind(rowKind); + row.setKind(rowKind); Object[] fieldValues = (Object[]) value[1]; for (int i = 0; i < fieldValues.length; i++) { - reuseRow.setField(i, fieldDataConverters[i].toInternal(fieldValues[i])); + row.setField(i, fieldDataConverters[i].toInternal(fieldValues[i])); } - return reuseRow; + return row; } @SuppressWarnings("unchecked") @@ -1023,11 +1029,15 @@ public class PythonTypeUtils { return null; } - reuseExternalRow[0] = (long) value.getKind().toByteValue(); + Object[] externalRow = reuse ? reuseExternalRow : new Object[2]; + Object[] externalRowData = + reuse ? reuseExternalRowData : new Object[fieldDataConverters.length]; + externalRow[0] = (long) value.getKind().toByteValue(); + externalRow[1] = externalRowData; for (int i = 0; i < value.getArity(); i++) { - reuseExternalRowData[i] = fieldDataConverters[i].toExternal(value.getField(i)); + externalRowData[i] = fieldDataConverters[i].toExternal(value.getField(i)); } - return reuseExternalRow; + return externalRow; } @Override @@ -1041,7 +1051,8 @@ public class PythonTypeUtils { } final RowDataConverter other = (RowDataConverter) o; - if (other.fieldDataConverters.length != this.fieldDataConverters.length) { + if (other.reuse != this.reuse + || other.fieldDataConverters.length != this.fieldDataConverters.length) { return false; } @@ -1101,11 +1112,13 @@ public class PythonTypeUtils { public static final class TupleDataConverter extends DataConverter<Tuple, Object[]> { private final DataConverter[] fieldDataConverters; + private final boolean reuse; private final Tuple reuseTuple; private final Object[] reuseExternalTuple; - TupleDataConverter(DataConverter[] dataConverters) { + TupleDataConverter(DataConverter[] dataConverters, boolean reuse) { this.fieldDataConverters = dataConverters; + this.reuse = reuse; this.reuseTuple = Tuple.newInstance(dataConverters.length); this.reuseExternalTuple = new Object[dataConverters.length]; } @@ -1117,10 +1130,11 @@ public class PythonTypeUtils { return null; } + Tuple tuple = reuse ? reuseTuple : Tuple.newInstance(fieldDataConverters.length); for (int i = 0; i < value.length; i++) { - reuseTuple.setField(fieldDataConverters[i].toInternal(value[i]), i); + tuple.setField(fieldDataConverters[i].toInternal(value[i]), i); } - return reuseTuple; + return tuple; } @SuppressWarnings("unchecked") @@ -1130,10 +1144,12 @@ public class PythonTypeUtils { return null; } + Object[] externalTuple = + reuse ? reuseExternalTuple : new Object[fieldDataConverters.length]; for (int i = 0; i < value.getArity(); i++) { - reuseExternalTuple[i] = fieldDataConverters[i].toExternal(value.getField(i)); + externalTuple[i] = fieldDataConverters[i].toExternal(value.getField(i)); } - return reuseExternalTuple; + return externalTuple; } @Override @@ -1147,7 +1163,8 @@ public class PythonTypeUtils { } final TupleDataConverter other = (TupleDataConverter) o; - if (other.fieldDataConverters.length != this.fieldDataConverters.length) { + if (other.reuse != this.reuse + || other.fieldDataConverters.length != this.fieldDataConverters.length) { return false; } @@ -1358,9 +1375,19 @@ public class PythonTypeUtils { BasicTypeInfo.BYTE_TYPE_INFO.getTypeClass(), ByteDataConverter.INSTANCE); } - @SuppressWarnings("unchecked") public static <IN, OUT> DataConverter<IN, OUT> typeInfoDataConverter( TypeInformation<IN> typeInformation) { + return typeInfoDataConverter(typeInformation, true); + } + + /** + * Row and Tuple converters reuse one object per converter only when {@code reuse} is true. + * Converters below an array, a list or a map don't reuse, since the container keeps every + * converted element. + */ + @SuppressWarnings("unchecked") + private static <IN, OUT> DataConverter<IN, OUT> typeInfoDataConverter( + TypeInformation<IN> typeInformation, boolean reuse) { DataConverter<IN, OUT> dataConverter = typeInfoToDataConverterMap.get(typeInformation.getTypeClass()); @@ -1374,27 +1401,29 @@ public class PythonTypeUtils { if (typeInformation instanceof RowTypeInfo) { DataConverter[] fieldDataConverters = Arrays.stream(((RowTypeInfo) typeInformation).getFieldTypes()) - .map(TypeInfoToDataConverter::typeInfoDataConverter) + .map(fieldType -> typeInfoDataConverter(fieldType, reuse)) .toArray(DataConverter[]::new); - return (DataConverter<IN, OUT>) new RowDataConverter(fieldDataConverters); + return (DataConverter<IN, OUT>) + new RowDataConverter(fieldDataConverters, reuse); } if (typeInformation instanceof TupleTypeInfo) { - Object[] fieldDataConverters = + DataConverter[] fieldDataConverters = Arrays.stream(((TupleTypeInfo) typeInformation).getFieldTypes()) - .map(TypeInfoToDataConverter::typeInfoDataConverter) + .map(fieldType -> typeInfoDataConverter(fieldType, reuse)) .toArray(DataConverter[]::new); return (DataConverter<IN, OUT>) - new TupleDataConverter((DataConverter[]) fieldDataConverters); + new TupleDataConverter(fieldDataConverters, reuse); } if (typeInformation instanceof BasicArrayTypeInfo) { return (DataConverter<IN, OUT>) new ArrayDataConverter( ((BasicArrayTypeInfo) typeInformation).getComponentTypeClass(), - TypeInfoToDataConverter.typeInfoDataConverter( + typeInfoDataConverter( ((BasicArrayTypeInfo) typeInformation) - .getComponentInfo())); + .getComponentInfo(), + false)); } if (typeInformation instanceof ObjectArrayTypeInfo) { @@ -1403,25 +1432,29 @@ public class PythonTypeUtils { ((ObjectArrayTypeInfo) typeInformation) .getComponentInfo() .getTypeClass(), - TypeInfoToDataConverter.typeInfoDataConverter( + typeInfoDataConverter( ((ObjectArrayTypeInfo) typeInformation) - .getComponentInfo())); + .getComponentInfo(), + false)); } if (typeInformation instanceof ListTypeInfo) { return (DataConverter<IN, OUT>) new ListDataConverter( - TypeInfoToDataConverter.typeInfoDataConverter( - ((ListTypeInfo) typeInformation).getElementTypeInfo())); + typeInfoDataConverter( + ((ListTypeInfo) typeInformation).getElementTypeInfo(), + false)); } if (typeInformation instanceof MapTypeInfo) { return (DataConverter<IN, OUT>) new MapDataConverter( - TypeInfoToDataConverter.typeInfoDataConverter( - ((MapTypeInfo) typeInformation).getKeyTypeInfo()), - TypeInfoToDataConverter.typeInfoDataConverter( - ((MapTypeInfo) typeInformation).getValueTypeInfo())); + typeInfoDataConverter( + ((MapTypeInfo) typeInformation).getKeyTypeInfo(), + false), + typeInfoDataConverter( + ((MapTypeInfo) typeInformation).getValueTypeInfo(), + false)); } if (typeInformation instanceof InternalTypeInfo) { @@ -1433,9 +1466,10 @@ public class PythonTypeUtils { if (typeInformation instanceof ExternalTypeInfo) { return (DataConverter<IN, OUT>) - TypeInfoToDataConverter.typeInfoDataConverter( + typeInfoDataConverter( LegacyTypeInfoDataTypeConverter.toLegacyTypeInfo( - ((ExternalTypeInfo<?>) typeInformation).getDataType())); + ((ExternalTypeInfo<?>) typeInformation).getDataType()), + reuse); } } diff --git a/flink-python/src/test/java/org/apache/flink/streaming/api/utils/PythonTypeUtilsTest.java b/flink-python/src/test/java/org/apache/flink/streaming/api/utils/PythonTypeUtilsTest.java index a45e4b6e3bf..12b0fb7ab99 100644 --- a/flink-python/src/test/java/org/apache/flink/streaming/api/utils/PythonTypeUtilsTest.java +++ b/flink-python/src/test/java/org/apache/flink/streaming/api/utils/PythonTypeUtilsTest.java @@ -36,6 +36,7 @@ import org.apache.flink.api.common.typeutils.base.LongSerializer; import org.apache.flink.api.common.typeutils.base.ShortSerializer; import org.apache.flink.api.common.typeutils.base.array.BytePrimitiveArraySerializer; import org.apache.flink.api.common.typeutils.base.array.IntPrimitiveArraySerializer; +import org.apache.flink.api.java.tuple.Tuple1; import org.apache.flink.api.java.typeutils.TupleTypeInfo; import org.apache.flink.api.java.typeutils.runtime.RowSerializer; import org.apache.flink.api.java.typeutils.runtime.TupleSerializer; @@ -55,7 +56,9 @@ import org.apache.flink.types.Row; import org.junit.jupiter.api.Test; +import java.util.Arrays; import java.util.HashMap; +import java.util.List; import java.util.Map; import static org.assertj.core.api.Assertions.assertThat; @@ -292,4 +295,60 @@ class PythonTypeUtilsTest { assertThat(dataConverter).isEqualTo(expectedDataConverter); } + + @SuppressWarnings("unchecked") + @Test + void testDataConverterKeepsContainerElementsSeparate() { + Object[] external1 = new Object[] {0L, new Object[] {1L}}; + Object[] external2 = new Object[] {0L, new Object[] {2L}}; + Row row1 = Row.of(1); + Row row2 = Row.of(2); + + PythonTypeUtils.DataConverter arrayConverter = + PythonTypeUtils.TypeInfoToDataConverter.typeInfoDataConverter( + Types.OBJECT_ARRAY(Types.ROW(Types.INT))); + assertThat((Object[]) arrayConverter.toInternal(new Object[] {external1, external2})) + .containsExactly(row1, row2); + Object[] externalArray = (Object[]) arrayConverter.toExternal(new Row[] {row1, row2}); + assertThat((Object[]) ((Object[]) externalArray[0])[1]).containsExactly(1L); + assertThat((Object[]) ((Object[]) externalArray[1])[1]).containsExactly(2L); + + PythonTypeUtils.DataConverter nestedArrayConverter = + PythonTypeUtils.TypeInfoToDataConverter.typeInfoDataConverter( + Types.OBJECT_ARRAY(Types.ROW(Types.ROW(Types.INT)))); + assertThat( + (Object[]) + nestedArrayConverter.toInternal( + new Object[] { + new Object[] {0L, new Object[] {external1}}, + new Object[] {0L, new Object[] {external2}} + })) + .containsExactly(Row.of(row1), Row.of(row2)); + + PythonTypeUtils.DataConverter tupleArrayConverter = + PythonTypeUtils.TypeInfoToDataConverter.typeInfoDataConverter( + Types.OBJECT_ARRAY(Types.TUPLE(Types.INT))); + assertThat( + (Object[]) + tupleArrayConverter.toInternal( + new Object[] {new Object[] {1L}, new Object[] {2L}})) + .containsExactly(Tuple1.of(1), Tuple1.of(2)); + + PythonTypeUtils.DataConverter listConverter = + PythonTypeUtils.TypeInfoToDataConverter.typeInfoDataConverter( + Types.LIST(Types.ROW(Types.INT))); + assertThat((List<Object>) listConverter.toInternal(Arrays.asList(external1, external2))) + .containsExactly(row1, row2); + + PythonTypeUtils.DataConverter mapConverter = + PythonTypeUtils.TypeInfoToDataConverter.typeInfoDataConverter( + Types.MAP(Types.STRING, Types.ROW(Types.INT))); + Map<String, Object> externalMap = new HashMap<>(); + externalMap.put("a", external1); + externalMap.put("b", external2); + Map<String, Row> expectedMap = new HashMap<>(); + expectedMap.put("a", row1); + expectedMap.put("b", row2); + assertThat((Map<String, Row>) mapConverter.toInternal(externalMap)).isEqualTo(expectedMap); + } }
