This is an automated email from the ASF dual-hosted git repository.
hxb pushed a commit to branch release-1.15
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.15 by this push:
new 507b93eef2a [FLINK-29477][python] Fix ClassCastException when collect
primitive array
507b93eef2a is described below
commit 507b93eef2af79ef2ad5752e1271e5c8915bb15f
Author: Juntao Hu <[email protected]>
AuthorDate: Fri Sep 30 13:01:30 2022 +0800
[FLINK-29477][python] Fix ClassCastException when collect primitive array
This closes #21016.
---
.../pyflink/datastream/tests/test_data_stream.py | 9 ++-
.../flink/api/common/python/PythonBridgeUtils.java | 80 ++++++++++++++++++++--
2 files changed, 83 insertions(+), 6 deletions(-)
diff --git a/flink-python/pyflink/datastream/tests/test_data_stream.py
b/flink-python/pyflink/datastream/tests/test_data_stream.py
index 3fd0a9fb372..e3f05936ed1 100644
--- a/flink-python/pyflink/datastream/tests/test_data_stream.py
+++ b/flink-python/pyflink/datastream/tests/test_data_stream.py
@@ -312,11 +312,18 @@ class DataStreamTests(object):
decimal.Decimal('2000000000000000000.061111111111111'
'11111111111111'))]
expected = test_data
- ds = self.env.from_collection(test_data).map(lambda a: a)
+ ds = self.env.from_collection(test_data)
with ds.execute_and_collect() as results:
actual = [result for result in results]
self.assert_equals_sorted(expected, actual)
+ test_data = [[1, 2, 3], [4, 5]]
+ expected = test_data
+ ds = self.env.from_collection(test_data,
type_info=Types.PRIMITIVE_ARRAY(Types.INT()))
+ with ds.execute_and_collect() as results:
+ actual = [r for r in results]
+ self.assert_equals_sorted(expected, actual)
+
def test_keyed_map(self):
from pyflink.util.java_utils import get_j_env_configuration
from pyflink.common import Configuration
diff --git
a/flink-python/src/main/java/org/apache/flink/api/common/python/PythonBridgeUtils.java
b/flink-python/src/main/java/org/apache/flink/api/common/python/PythonBridgeUtils.java
index 10d9cf30b7e..ebf4f346de3 100644
---
a/flink-python/src/main/java/org/apache/flink/api/common/python/PythonBridgeUtils.java
+++
b/flink-python/src/main/java/org/apache/flink/api/common/python/PythonBridgeUtils.java
@@ -44,6 +44,7 @@ import org.apache.flink.table.types.logical.RowType;
import org.apache.flink.table.types.logical.TimeType;
import org.apache.flink.table.types.logical.TimestampType;
import org.apache.flink.types.Row;
+import org.apache.flink.util.Preconditions;
import net.razorvine.pickle.Pickler;
import net.razorvine.pickle.Unpickler;
@@ -270,12 +271,16 @@ public final class PythonBridgeUtils {
return fieldBytes;
} else if (dataType instanceof BasicArrayTypeInfo
|| dataType instanceof PrimitiveArrayTypeInfo) {
- Object[] objects = (Object[]) obj;
+ Object[] objects;
+ TypeInformation<?> elementType;
+ if (dataType instanceof BasicArrayTypeInfo) {
+ objects = (Object[]) obj;
+ elementType = ((BasicArrayTypeInfo<?, ?>)
dataType).getComponentInfo();
+ } else {
+ objects = primitiveArrayConverter(obj, dataType);
+ elementType = ((PrimitiveArrayTypeInfo<?>)
dataType).getComponentType();
+ }
List<Object> serializedElements = new
ArrayList<>(objects.length);
- TypeInformation<?> elementType =
- dataType instanceof BasicArrayTypeInfo
- ? ((BasicArrayTypeInfo<?, ?>)
dataType).getComponentInfo()
- : ((PrimitiveArrayTypeInfo<?>)
dataType).getComponentType();
for (Object object : objects) {
serializedElements.add(getPickledBytesFromJavaObject(object, elementType));
}
@@ -323,6 +328,71 @@ public final class PythonBridgeUtils {
}
}
+ private static Object[] primitiveArrayConverter(
+ Object array, TypeInformation<?> arrayTypeInfo) {
+ Preconditions.checkArgument(arrayTypeInfo instanceof
PrimitiveArrayTypeInfo);
+ Preconditions.checkArgument(array.getClass().isArray());
+ Object[] objects;
+ if
(PrimitiveArrayTypeInfo.BOOLEAN_PRIMITIVE_ARRAY_TYPE_INFO.equals(arrayTypeInfo))
{
+ boolean[] booleans = (boolean[]) array;
+ objects = new Object[booleans.length];
+ for (int i = 0; i < booleans.length; i++) {
+ objects[i] = booleans[i];
+ }
+ } else if
(PrimitiveArrayTypeInfo.BYTE_PRIMITIVE_ARRAY_TYPE_INFO.equals(arrayTypeInfo)) {
+ byte[] bytes = (byte[]) array;
+ objects = new Object[bytes.length];
+ for (int i = 0; i < bytes.length; i++) {
+ objects[i] = bytes[i];
+ }
+ } else if
(PrimitiveArrayTypeInfo.SHORT_PRIMITIVE_ARRAY_TYPE_INFO.equals(arrayTypeInfo)) {
+ short[] shorts = (short[]) array;
+ objects = new Object[shorts.length];
+ for (int i = 0; i < shorts.length; i++) {
+ objects[i] = shorts[i];
+ }
+ } else if
(PrimitiveArrayTypeInfo.INT_PRIMITIVE_ARRAY_TYPE_INFO.equals(arrayTypeInfo)) {
+ int[] ints = (int[]) array;
+ objects = new Object[ints.length];
+ for (int i = 0; i < ints.length; i++) {
+ objects[i] = ints[i];
+ }
+ } else if
(PrimitiveArrayTypeInfo.LONG_PRIMITIVE_ARRAY_TYPE_INFO.equals(arrayTypeInfo)) {
+ long[] longs = (long[]) array;
+ objects = new Object[longs.length];
+ for (int i = 0; i < longs.length; i++) {
+ objects[i] = longs[i];
+ }
+ } else if
(PrimitiveArrayTypeInfo.FLOAT_PRIMITIVE_ARRAY_TYPE_INFO.equals(arrayTypeInfo)) {
+ float[] floats = (float[]) array;
+ objects = new Object[floats.length];
+ for (int i = 0; i < floats.length; i++) {
+ objects[i] = floats[i];
+ }
+ } else if
(PrimitiveArrayTypeInfo.DOUBLE_PRIMITIVE_ARRAY_TYPE_INFO.equals(arrayTypeInfo))
{
+ double[] doubles = (double[]) array;
+ objects = new Object[doubles.length];
+ for (int i = 0; i < doubles.length; i++) {
+ objects[i] = doubles[i];
+ }
+ } else if
(PrimitiveArrayTypeInfo.CHAR_PRIMITIVE_ARRAY_TYPE_INFO.equals(arrayTypeInfo)) {
+ char[] chars = (char[]) array;
+ objects = new Object[chars.length];
+ for (int i = 0; i < chars.length; i++) {
+ objects[i] = chars[i];
+ }
+ } else {
+ throw new UnsupportedOperationException(
+ String.format(
+ "Primitive array of %s is not supported in PyFlink
yet",
+ ((PrimitiveArrayTypeInfo<?>) arrayTypeInfo)
+ .getComponentType()
+ .getTypeClass()
+ .getSimpleName()));
+ }
+ return objects;
+ }
+
public static Object getPickledBytesFromRow(Row row, DataType[] dataTypes)
throws IOException {
LogicalType[] logicalTypes =
Arrays.stream(dataTypes).map(f ->
f.getLogicalType()).toArray(LogicalType[]::new);
