yusufyildiz34 commented on code in PR #18508:
URL: https://github.com/apache/iceberg/pull/18508#discussion_r4287865860
##########
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);
+ }
Review Comment:
Done. The getters are now built once in a template and shared through a
private constructor, via `copyFor(RowData)`, the same pattern as
`InternalRecordWrapper.copyFor`. Sharing them is also why ROW no longer reuses
one nested wrapper; details in the buildGetter thread.
--
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]
