mbutrovich commented on code in PR #18508:
URL: https://github.com/apache/iceberg/pull/18508#discussion_r4286866273
##########
flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/TestRowDataWrapper.java:
##########
@@ -91,4 +105,86 @@ protected void generateAndValidate(
assertThat(actual).isExhausted();
assertThat(expected).isExhausted();
}
+
+ @Test
+ void listElementsAreConverted() {
+ Schema schema =
+ new Schema(
+ Types.NestedField.optional(
+ 1, "tags", Types.ListType.ofOptional(2,
Types.StringType.get())));
+ RowData row =
+ GenericRowData.of(new GenericArrayData(new Object[]
{StringData.fromString("a"), null}));
+
+ assertThat(wrap(schema, row).get(0, List.class)).containsExactly("a",
null);
+ }
+
+ @Test
+ void listOfTimestampsIsConvertedToMicros() {
+ Schema schema =
+ new Schema(
+ Types.NestedField.optional(
+ 1, "times", Types.ListType.ofOptional(2,
Types.TimestampType.withoutZone())));
+ LocalDateTime time = LocalDateTime.of(2026, 10, 14, 12, 30, 45,
123_456_000);
+ RowData row =
+ GenericRowData.of(
+ new GenericArrayData(new Object[]
{TimestampData.fromLocalDateTime(time)}));
+
+ assertThat(wrap(schema, row).get(0, List.class))
+ .containsExactly(DateTimeUtil.microsFromTimestamp(time));
+ }
+
+ @Test
+ void listOfStructsHasIndependentElements() {
+ Schema schema =
+ new Schema(
+ Types.NestedField.optional(
+ 1,
+ "points",
+ Types.ListType.ofOptional(
+ 2,
+ Types.StructType.of(
+ Types.NestedField.required(3, "x",
Types.IntegerType.get())))));
+ RowData row =
+ GenericRowData.of(
+ new GenericArrayData(new Object[] {GenericRowData.of(1),
GenericRowData.of(2)}));
+
+ List<?> points = wrap(schema, row).get(0, List.class);
+ assertThat(points).hasSize(2);
+ assertThat(((StructLike) points.get(0)).get(0,
Integer.class)).isEqualTo(1);
+ assertThat(((StructLike) points.get(1)).get(0,
Integer.class)).isEqualTo(2);
+ }
+
+ @Test
+ void listOfUnknownElements() {
+ Schema schema =
+ new Schema(
+ Types.NestedField.optional(
+ 1, "unknowns", Types.ListType.ofOptional(2,
Types.UnknownType.get())));
+ RowData row = GenericRowData.of(new GenericArrayData(new Object[] {null}));
+
+ assertThat(wrap(schema, row).get(0, List.class)).containsExactly((Object)
null);
+ }
+
+ @Test
+ void mapKeysAndValuesAreConverted() {
+ Schema schema =
+ new Schema(
+ Types.NestedField.optional(
+ 1,
+ "prices",
+ Types.MapType.ofOptional(
+ 2, 3, Types.StringType.get(), Types.DecimalType.of(9,
2))));
+ BigDecimal price = new BigDecimal("1.50");
+ RowData row =
+ GenericRowData.of(
+ new GenericMapData(
+ ImmutableMap.of(
+ StringData.fromString("a"),
DecimalData.fromBigDecimal(price, 9, 2))));
+
+ assertThat(wrap(schema, row).get(0,
Map.class)).isEqualTo(ImmutableMap.of("a", price));
+ }
Review Comment:
Could you add cases for an empty list, a map with a `null` value, and a list
nested in a list? Those take paths the current tests don't reach: zero
iterations in `toList`, `convertOrNull` on a map value, and the recursive
`ARRAY` case in `buildConverter`.
##########
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:
This calls the public constructor for every struct element of every row. The
[constructor](https://github.com/apache/iceberg/blob/6f9679ecc0ab9812546b43aaa647d8856abfcb53/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/RowDataWrapper.java#L54-L71)
allocates the `types` and `getters` arrays and calls `buildGetter` for each
field, which also rebuilds any nested list or map converters. What do you think
about building the getters once and sharing them across the element wrappers?
For example, with a private constructor:
```java
private RowDataWrapper(LogicalType[] types, PositionalGetter<?>[] getters)
{
this.types = types;
this.getters = getters;
}
```
```suggestion
case ROW -> {
RowDataWrapper template = new RowDataWrapper((RowType) logicalType,
type.asStructType());
// Collection elements are read together, so each element needs its
own wrapper.
yield value -> new RowDataWrapper(template.types,
template.getters).wrap((RowData) value);
}
```
##########
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:
Should this move to `FlinkRowData` as `createElementGetter`, next to
[`createFieldGetter`](https://github.com/apache/iceberg/blob/9120cfa789d4ea29acd22c95973774f6a813a212/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/FlinkRowData.java#L29-L46)?
That class already holds the same `NullType` workaround for row fields, so
both Flink getter workarounds would live in one place for the backports.
##########
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) {
Review Comment:
Should `buildGetter` be built on top of `buildConverter`? As I read it, the
cases from `TINYINT` through `ROW` here repeat the conversions in
[`buildGetter`](https://github.com/apache/iceberg/blob/6f9679ecc0ab9812546b43aaa647d8856abfcb53/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/RowDataWrapper.java#L102-L180)
one for one. The only difference is that one reads from a `RowData` position
and the other takes a value. That leaves two switches to keep in sync here and
in the 1.20 and 2.2 backports.
One way is to read the raw value with `FlinkRowData.createFieldGetter` and
convert it, keeping `ROW` separate so a top-level struct still reuses one
wrapper:
```java
private static PositionalGetter<?> buildGetter(LogicalType logicalType,
Type type, int pos) {
if (logicalType instanceof RowType rowType) {
RowDataWrapper nestedWrapper = new RowDataWrapper(rowType,
type.asStructType());
return (row, p) -> nestedWrapper.wrap(row.getRow(p,
rowType.getFieldCount()));
}
RowData.FieldGetter fieldGetter =
FlinkRowData.createFieldGetter(logicalType, pos);
Function<Object, Object> converter = buildConverter(logicalType, type);
return (row, p) -> converter.apply(fieldGetter.getFieldOrNull(row));
}
```
The constructor would then call `buildGetter(types[i],
struct.fields().get(i).type(), i)` and drop the `null` fallback. The randomized
cases in `RecordWrapperTestBase` would then cover the same conversions that
list and map elements use. The cost is one extra box per `TIME` value, and per
`SMALLINT` value outside the `Short` cache, because the field getter boxes the
raw `int` or `short` before the conversion.
--
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]
