claudevdm commented on code in PR #39344:
URL: https://github.com/apache/beam/pull/39344#discussion_r3596179503
##########
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:
I see some other call sites that do not use the update compat override
https://github.com/apache/beam/blob/2cd12e5968c496df7ec31bba05bfe3ad1bc7297a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/IcebergCatalogConfig.java#L205
https://github.com/apache/beam/blob/2cd12e5968c496df7ec31bba05bfe3ad1bc7297a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/CdcOutputUtils.java#L139
##########
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:
What if it is a Timestamp.NANOS? Should we throw if the logical type is not
explicitly MICROS?
##########
sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/catalog/BigQueryMetastoreCatalogIT.java:
##########
@@ -115,6 +116,8 @@ public void testWriteToPartitionedAndValidateWithBQQuery()
.map(tr -> BigQueryUtils.toBeamRow(BEAM_SCHEMA, tr))
.collect(Collectors.toList());
+ System.out.println("returned: " + beamRows);
Review Comment:
Remove?
##########
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:
Will using timestamp logical type break xlang python reads?
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryUtils.java:
##########
@@ -932,7 +932,14 @@ public static Row toBeamRow(Schema rowSchema, TableSchema
bqSchema, TableRow jso
return java.time.Instant.parse(jsonBQString);
}
} else if (fieldType.isLogicalType(Timestamp.IDENTIFIER)) {
- return VAR_PRECISION_FORMATTER.parse(jsonBQString,
java.time.Instant::from);
+ try {
Review Comment:
This is also used by load jobs and direct reads and both of them use the UTC
string version right?
I am a bit concerned about cost of try to parse -> throw exception
(expensive) -> try again for every single row in those cases.
Can we either do a conditional (check if ends with UTC or if contains :) to
infer which approach we should take?
Maybe a future improvement we can make is the caller can pass the API they
used to read so we know what type to expect.
##########
sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/catalog/BigQueryMetastoreCatalogIT.java:
##########
@@ -58,6 +58,7 @@ public Catalog createCatalog() {
.put("gcp_project", OPTIONS.getProject())
.put("gcp_location", "us-central1")
.put("warehouse", warehouse)
+ .put("io-impl", "org.apache.iceberg.gcp.gcs.GCSFileIO")
Review Comment:
What is this for?
##########
sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryUtilsTest.java:
##########
Review Comment:
Update test name since it doesnt just test UTC anymore?
##########
sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryUtilsTest.java:
##########
@@ -1449,17 +1450,27 @@ public void testToBeamRow_timestampMicros_utcSuffix() {
// BigQuery format with " UTC" suffix
String timestamp = "2024-08-10 16:52:07.123456 UTC";
-
- Row beamRow = BigQueryUtils.toBeamRow(schema, new TableRow().set("ts",
timestamp));
-
- java.time.Instant actual = (java.time.Instant) beamRow.getValue("ts");
- assertEquals(2024, actual.atZone(java.time.ZoneOffset.UTC).getYear());
- assertEquals(8, actual.atZone(java.time.ZoneOffset.UTC).getMonthValue());
- assertEquals(10, actual.atZone(java.time.ZoneOffset.UTC).getDayOfMonth());
- assertEquals(16, actual.atZone(java.time.ZoneOffset.UTC).getHour());
- assertEquals(52, actual.atZone(java.time.ZoneOffset.UTC).getMinute());
- assertEquals(7, actual.atZone(java.time.ZoneOffset.UTC).getSecond());
- assertEquals(123456000, actual.getNano());
+ String parsableTimestamp = "2024-08-10T16:52:07.123456Z";
Review Comment:
Can we add a negative epoch and micros with leading zeros (e.g. .012345)
test?
--
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]
