This is an automated email from the ASF dual-hosted git repository.
dianfu pushed a commit to branch release-1.20
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.20 by this push:
new 770a086545b [FLINK-41228][python] Fix thread-mode conversion of
arrays, lists and maps (#29630)
770a086545b is described below
commit 770a086545bd8218b2d4c7ffb91e19740a224655
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 4b41ee6323b..17d0a0650a1 100644
--- a/flink-python/pyflink/datastream/tests/test_data_stream.py
+++ b/flink-python/pyflink/datastream/tests/test_data_stream.py
@@ -1347,6 +1347,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 8829d48074e..b2b2f8d0c82 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 7ee63382c1c..9b0414ccf12 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);
+ }
}
