ahmedabu98 commented on code in PR #39344:
URL: https://github.com/apache/beam/pull/39344#discussion_r3605814214
##########
sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/IcebergUtils.java:
##########
@@ -81,9 +83,11 @@ private IcebergUtils() {}
.put(SqlTypes.DATETIME.getIdentifier(),
Types.TimestampType.withoutZone())
.put(SqlTypes.UUID.getIdentifier(), Types.UUIDType.get())
.put(MicrosInstant.IDENTIFIER, Types.TimestampType.withZone())
+ .put(Timestamp.IDENTIFIER, Types.TimestampType.withZone())
Review Comment:
Hmmm yes it probably will... looks like we need to create the Python
representation of Timestamp logical type first
##########
sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/IcebergUtils.java:
##########
@@ -114,36 +125,51 @@ private static Schema.FieldType
icebergTypeToBeamFieldType(final Type type) {
case DECIMAL:
return Schema.FieldType.DECIMAL;
case STRUCT:
- return
Schema.FieldType.row(icebergStructTypeToBeamSchema(type.asStructType()));
+ return Schema.FieldType.row(
+ icebergStructTypeToBeamSchema(type.asStructType(),
updateCompatibilityVersion));
case LIST:
- return
Schema.FieldType.array(icebergTypeToBeamFieldType(type.asListType().elementType()));
+ return Schema.FieldType.array(
+ icebergTypeToBeamFieldType(
+ type.asListType().elementType(), updateCompatibilityVersion));
case MAP:
return Schema.FieldType.map(
- icebergTypeToBeamFieldType(type.asMapType().keyType()),
- icebergTypeToBeamFieldType(type.asMapType().valueType()));
+ icebergTypeToBeamFieldType(type.asMapType().keyType(),
updateCompatibilityVersion),
+ icebergTypeToBeamFieldType(type.asMapType().valueType(),
updateCompatibilityVersion));
default:
throw new RuntimeException("Unrecognized Iceberg Type: " +
type.typeId());
}
}
- private static Schema.Field icebergFieldToBeamField(final Types.NestedField
field) {
- return Schema.Field.of(field.name(),
icebergTypeToBeamFieldType(field.type()))
+ private static Schema.Field icebergFieldToBeamField(
+ final Types.NestedField field, @Nullable String
updateCompatibilityVersion) {
+ return Schema.Field.of(
+ field.name(), icebergTypeToBeamFieldType(field.type(),
updateCompatibilityVersion))
.withNullable(field.isOptional());
}
/** Converts an Iceberg {@link org.apache.iceberg.Schema} to a Beam {@link
Schema}. */
public static Schema icebergSchemaToBeamSchema(final
org.apache.iceberg.Schema schema) {
+ return icebergSchemaToBeamSchema(schema, null);
+ }
+
+ /**
+ * Converts an Iceberg {@link org.apache.iceberg.Schema} to a Beam {@link
Schema}, accounting for
+ * update compatibility.
+ */
+ public static Schema icebergSchemaToBeamSchema(
Review Comment:
First one is only called by SQL, should be fine.
Second one is still not called anywhere (ongoing CDC read work: #38831).
It'll come later so no need to wire update compat there
--
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]
