yusufyildiz34 commented on code in PR #18508:
URL: https://github.com/apache/iceberg/pull/18508#discussion_r4287880385
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/RowDataWrapper.java:
##########
@@ -155,8 +166,114 @@ private static PositionalGetter<?>
buildGetter(LogicalType logicalType, Type typ
RowDataWrapper nestedWrapper = new RowDataWrapper(rowType, structType);
return (row, pos) -> nestedWrapper.wrap(row.getRow(pos,
rowType.getFieldCount()));
+ case ARRAY:
+ Function<Object, Object> listConverter = buildConverter(logicalType,
type);
+ return (row, pos) -> listConverter.apply(row.getArray(pos));
+
+ case MAP:
+ Function<Object, Object> mapConverter = buildConverter(logicalType,
type);
+ return (row, pos) -> mapConverter.apply(row.getMap(pos));
+
default:
return null;
}
}
+
+ private static Function<Object, Object> buildConverter(LogicalType
logicalType, Type type) {
+ return switch (logicalType.getTypeRoot()) {
+ case TINYINT -> value -> ((Byte) value).intValue();
+ case SMALLINT -> value -> ((Short) value).intValue();
+ case CHAR, VARCHAR -> Object::toString;
+ case BINARY, VARBINARY ->
+ Type.TypeID.UUID == type.typeId()
+ ? value -> UUIDUtil.convert((byte[]) value)
+ : value -> ByteBuffer.wrap((byte[]) value);
+ case DECIMAL -> value -> ((DecimalData) value).toBigDecimal();
+ case TIME_WITHOUT_TIME_ZONE -> value -> ((Integer) value).longValue() *
1_000;
+ case TIMESTAMP_WITHOUT_TIME_ZONE ->
+ type.typeId() == Type.TypeID.TIMESTAMP_NANO
+ ? value -> DateTimeUtil.nanosFromTimestamp(((TimestampData)
value).toLocalDateTime())
+ : value ->
+ DateTimeUtil.microsFromTimestamp(((TimestampData)
value).toLocalDateTime());
+ case TIMESTAMP_WITH_LOCAL_TIME_ZONE ->
+ type.typeId() == Type.TypeID.TIMESTAMP_NANO
+ ? value -> {
+ TimestampData timestampData = (TimestampData) value;
+ return timestampData.getMillisecond() * 1_000_000L
+ + timestampData.getNanoOfMillisecond();
+ }
+ : value -> {
+ TimestampData timestampData = (TimestampData) value;
+ return timestampData.getMillisecond() * 1000L
+ + timestampData.getNanoOfMillisecond() / 1000;
+ };
+ case ROW -> {
+ RowType rowType = (RowType) logicalType;
+ Types.StructType structType = type.asStructType();
+ // Collection elements are read together, so each element needs its
own wrapper.
+ yield value -> new RowDataWrapper(rowType, structType).wrap((RowData)
value);
+ }
+ case ARRAY -> {
+ LogicalType elementType = ((ArrayType) logicalType).getElementType();
+ ArrayData.ElementGetter elementGetter = elementGetter(elementType);
+ Function<Object, Object> elementConverter =
+ buildConverter(elementType, type.asListType().elementType());
+ yield value -> toList((ArrayData) value, elementGetter,
elementConverter);
+ }
+ case MAP -> {
+ MapType mapType = (MapType) logicalType;
+ Types.MapType icebergMapType = type.asMapType();
+ ArrayData.ElementGetter keyGetter =
elementGetter(mapType.getKeyType());
+ ArrayData.ElementGetter valueGetter =
elementGetter(mapType.getValueType());
+ Function<Object, Object> keyConverter =
+ buildConverter(mapType.getKeyType(), icebergMapType.keyType());
+ Function<Object, Object> valueConverter =
+ buildConverter(mapType.getValueType(), icebergMapType.valueType());
+ yield value -> toMap((MapData) value, keyGetter, keyConverter,
valueGetter, valueConverter);
+ }
+ default -> Function.identity();
+ };
+ }
+
+ private static ArrayData.ElementGetter elementGetter(LogicalType
elementType) {
+ // Iceberg's unknown type is Flink's NULL type, which createElementGetter
rejects.
+ if (elementType instanceof NullType) {
+ return (array, pos) -> null;
+ }
+
+ return ArrayData.createElementGetter(elementType);
+ }
Review Comment:
Done, moved it to `FlinkRowData.createElementGetter`, next to
`createFieldGetter`. I made it package-private since it's only used in this
package. For the 1.20 backport, it will also need the same null check as
`createFieldGetter`, because Flink 1.20's element getter skips it for NOT NULL
types.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
