github-actions[bot] commented on code in PR #40998:
URL: https://github.com/apache/beam/pull/40998#discussion_r4472367061
##########
sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/IcebergUtils.java:
##########
@@ -684,24 +691,62 @@ private static void addIcebergValue(
field.getType().getRowSchema(),
"Corrupted schema: Row type did not have associated nested
schema.");
if (icebergValue instanceof Record) {
- rowBuilder.addValue(icebergRecordToBeamRow(nestedSchema, (Record)
icebergValue));
+ values.add(icebergRecordToBeamRow(nestedSchema, (Record)
icebergValue));
} else if (icebergValue instanceof StructLike) {
- rowBuilder.addValue(structToRow(nestedSchema, (StructLike)
icebergValue));
+ values.add(structToRow(nestedSchema, (StructLike) icebergValue));
} else {
throw new UnsupportedOperationException(
"Unsupported row type: " + icebergValue.getClass());
}
break;
case LOGICAL_TYPE:
- rowBuilder.addValue(getLogicalTypeValue(icebergValue,
field.getType()));
+ values.add(getLogicalTypeValue(icebergValue, field.getType()));
break;
default:
throw new UnsupportedOperationException(
"Unsupported Beam type: " + field.getType().getTypeName());
}
}
- private static DateTime getBeamDateTimeValue(Object icebergValue) {
+ /**
+ * Copies a list or map, or one of its elements, the way {@link
Row.Builder#build()} would. A null
+ * fails where the Beam type is not nullable, and {@code binary}, which
Iceberg reads as a {@link
Review Comment:
**🔴 Critical**: Using `attachValues` here bypasses `RowUtils.verify`, which
causes three issues in `icebergRecordValueToBeamFieldValue`:
1. **`DATETIME` representation mismatch (`ClassCastException` during
encoding)**: For `TypeName.DATETIME`, `icebergRecordValueToBeamFieldValue`
returns an `org.joda.time.DateTime` (lines 798, 801, 804, 806). Previously,
`RowUtils.verifyDateTime` converted `ReadableDateTime` to
`org.joda.time.Instant`. With `attachValues`, the `DateTime` instance is stored
directly in the `Row`. When `RowCoder` encodes this row, `InstantCoder` is used
for `TypeName.DATETIME` and throws `ClassCastException: org.joda.time.DateTime
cannot be cast to org.joda.time.Instant`.
2. **`ITERABLE` fields (`ClassCastException` in
`icebergRecordValueToBeamFieldValue`)**: When the Beam schema has a
`TypeName.ITERABLE` field and the Iceberg record contains a non-`List`
`Iterable` (such as in
`IcebergUtilsTest.RecordToRowTests.testIterableOfRecords`, which passes
`Iterables.unmodifiableIterable(...)`), line 816 casts `(List) val`, throwing a
`ClassCastException` (previously `case ITERABLE:` was missing from the
`switch`, so it fell through to `default` and `RowUtils.verify` handled the
`Iterable`). Even if changed to `(Iterable<?>) val`, `Stream.collect(toList())`
produces an `ArrayList`, which changes the runtime type and breaks `Row.equals`
when compared against rows built with an `Iterable`.
3. **Missing nullability check**: In `icebergRecordValueToBeamFieldValue`,
`if (val == null) { return null; }` does not check `fieldType.getNullable()`.
Bypassing `RowUtils.verify` allows `null` values to be attached to non-nullable
schema fields without throwing an error.
To safely use `attachValues`, `icebergRecordValueToBeamFieldValue` should:
- Return `DateTime.toInstant()` (or `Instant.ofEpochMilli(...)`) for
`DATETIME` fields.
- Handle `Iterable<?>` (not just `List<?>`) in `case ARRAY: case ITERABLE:`.
- Validate that `val != null` when `!fieldType.getNullable()`.
--
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]
