yusufyildiz34 commented on code in PR #18508:
URL: https://github.com/apache/iceberg/pull/18508#discussion_r4327153187
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/RowDataWrapper.java:
##########
@@ -84,79 +92,119 @@ public <T> void set(int pos, T value) {
"Could not set a field in the RowDataWrapper because rowData is
read-only");
}
- private interface PositionalGetter<T> {
- T get(RowData data, int pos);
+ private RowDataWrapper copyFor(RowData data) {
+ return new RowDataWrapper(types, getters).wrap(data);
+ }
+
+ private static RowData.FieldGetter buildGetter(
+ LogicalType logicalType, Type type, int pos, boolean
reuseStructWrappers) {
+ RowData.FieldGetter fieldGetter =
FlinkRowData.createFieldGetter(logicalType, pos);
+ if (reuseStructWrappers && logicalType instanceof RowType rowType) {
+ // A struct field is read one row at a time, so its wrapper can be
reused.
+ RowDataWrapper structWrapper = new RowDataWrapper(rowType,
type.asStructType());
+ return row -> structWrapper.wrap((RowData)
fieldGetter.getFieldOrNull(row));
+ }
+
+ // get() returns null fields before calling a getter, so values are never
null here.
+ Function<Object, Object> converter = buildConverter(logicalType, type);
+ return row -> converter.apply(fieldGetter.getFieldOrNull(row));
+ }
+
+ 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 ->
+ // Flink's time is milliseconds (int), Iceberg's time is
microseconds (long)
+ 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 -> {
+ // Values read together, like list elements, need their own wrappers.
The template
+ // doesn't reuse struct wrappers, so its getters can be shared by the
copies.
+ RowDataWrapper template =
+ new RowDataWrapper(
+ (RowType) logicalType, type.asStructType(), false /* new
wrapper per value */);
+ yield value -> template.copyFor((RowData) value);
+ }
+ case ARRAY -> {
+ LogicalType elementType = ((ArrayType) logicalType).getElementType();
+ ArrayData.ElementGetter elementGetter =
FlinkRowData.createElementGetter(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 =
FlinkRowData.createElementGetter(mapType.getKeyType());
+ ArrayData.ElementGetter valueGetter =
+ FlinkRowData.createElementGetter(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();
Review Comment:
Done in fe531933f, thanks!
--
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]
