gemini-code-assist[bot] commented on code in PR #39344:
URL: https://github.com/apache/beam/pull/39344#discussion_r3589688889
##########
sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/IcebergUtils.java:
##########
@@ -613,21 +639,28 @@ private static Object getLogicalTypeValue(Object
icebergValue, Schema.FieldType
return LocalTime.parse(strValue);
} else if (type.isLogicalType(SqlTypes.DATETIME.getIdentifier())) {
return LocalDateTime.parse(strValue);
+ } else if (type.isLogicalType(Timestamp.IDENTIFIER)) {
+ return java.time.Instant.parse(strValue);
}
Review Comment:
Using `java.time.Instant.parse(strValue)` will throw a
`DateTimeParseException` if the string contains a timezone offset other than
`Z` (for example, `-08:00` or `+03:27`), because `Instant.parse` strictly
requires a UTC instant representation ending in `Z`.
To robustly parse any ISO-8601 string with a timezone offset, use
`OffsetDateTime.parse(strValue).toInstant()` instead.
```suggestion
} else if (type.isLogicalType(Timestamp.IDENTIFIER)) {
return OffsetDateTime.parse(strValue).toInstant();
}
```
##########
sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/IcebergIOReadTest.java:
##########
@@ -731,6 +737,48 @@ public void testBatchReadBetweenTimestamps() throws
IOException {
runReadWithBoundary(false, false);
}
+ @Test
+ public void testTimestampUpdateCompat() throws IOException {
+ String val = "2026-07-15T13:18:20.053123+03:27";
+ OffsetDateTime ts = OffsetDateTime.parse(val);
+
+ TableIdentifier tableId =
+ TableIdentifier.of("default", "table" +
Long.toString(UUID.randomUUID().hashCode(), 16));
+ org.apache.iceberg.Schema schema =
+ new org.apache.iceberg.Schema(
+ Collections.singletonList(required(1, "ts",
Types.TimestampType.withZone())),
+ ImmutableSet.of(1));
+ Table table = warehouse.createTable(tableId, schema);
+ DataFile file =
+ warehouse.writeData(
+ "date.parquet", schema,
Collections.singletonList(ImmutableMap.of("ts", ts)));
+ table.newFastAppend().appendFile(file).commit();
+
+ IcebergIO.ReadRows read =
IcebergIO.readRows(catalogConfig()).from(tableId);
+ if (useIncrementalScan) {
+ read = read.withCdc().toSnapshot(table.currentSnapshot().snapshotId());
+ }
+
+ Schema expectedBeamSchema =
+ Schema.builder().addLogicalTypeField("ts", Timestamp.MICROS).build();
+ Row expectedRow =
Row.withSchema(expectedBeamSchema).addValue(ts.toInstant()).build();
+
+ PCollection<Row> output = testPipeline.apply(read).apply(new PrintRow());
+ PAssert.that(output).containsInAnyOrder(expectedRow);
+ testPipeline.run().waitUntilFinish();
+
+ // test again but with older versions that require primitive DATETIME type
+ Schema expectedLegacyBeamSchema =
Schema.builder().addDateTimeField("ts").build();
+ Row expectedLegacyRow =
+
Row.withSchema(expectedLegacyBeamSchema).addValue(DateTime.parse(val)).build();
+
+ Pipeline testPipeline2 = Pipeline.create();
+
testPipeline2.getOptions().as(StreamingOptions.class).setUpdateCompatibilityVersion("2.75.0");
+ PCollection<Row> outputLegacy = testPipeline2.apply(read).apply(new
PrintRow());
+ PAssert.that(outputLegacy).containsInAnyOrder(expectedLegacyRow);
+ testPipeline2.run().waitUntilFinish();
+ }
Review Comment:
Using `Pipeline.create()` with no arguments in tests is an anti-pattern
because it does not inherit the pipeline options (such as the runner or
classpath configurations) set up for the test suite.
Additionally, running two pipelines sequentially within a single `@Test`
method prevents the use of the `@Rule` `TestPipeline` for both runs.
It is highly recommended to split this test into two separate test methods
(e.g., `testTimestampMicros` and `testTimestampLegacyCompat`). This allows both
tests to use the `TestPipeline` rule properly and ensures they run with the
correct test runner and options.
```java
private Table createTableWithTimestamp(TableIdentifier tableId,
OffsetDateTime ts) throws IOException {
org.apache.iceberg.Schema schema =
new org.apache.iceberg.Schema(
Collections.singletonList(required(1, "ts",
Types.TimestampType.withZone())),
ImmutableSet.of(1));
Table table = warehouse.createTable(tableId, schema);
DataFile file =
warehouse.writeData(
"date.parquet", schema,
Collections.singletonList(ImmutableMap.of("ts", ts)));
table.newFastAppend().appendFile(file).commit();
return table;
}
@Test
public void testTimestampMicros() throws IOException {
String val = "2026-07-15T13:18:20.053123+03:27";
OffsetDateTime ts = OffsetDateTime.parse(val);
TableIdentifier tableId =
TableIdentifier.of("default", "table" +
Long.toString(UUID.randomUUID().hashCode(), 16));
Table table = createTableWithTimestamp(tableId, ts);
IcebergIO.ReadRows read =
IcebergIO.readRows(catalogConfig()).from(tableId);
if (useIncrementalScan) {
read = read.withCdc().toSnapshot(table.currentSnapshot().snapshotId());
}
Schema expectedBeamSchema =
Schema.builder().addLogicalTypeField("ts", Timestamp.MICROS).build();
Row expectedRow =
Row.withSchema(expectedBeamSchema).addValue(ts.toInstant()).build();
PCollection<Row> output = testPipeline.apply(read).apply(new PrintRow());
PAssert.that(output).containsInAnyOrder(expectedRow);
testPipeline.run().waitUntilFinish();
}
@Test
public void testTimestampLegacyCompat() throws IOException {
String val = "2026-07-15T13:18:20.053123+03:27";
OffsetDateTime ts = OffsetDateTime.parse(val);
TableIdentifier tableId =
TableIdentifier.of("default", "table" +
Long.toString(UUID.randomUUID().hashCode(), 16));
Table table = createTableWithTimestamp(tableId, ts);
IcebergIO.ReadRows read =
IcebergIO.readRows(catalogConfig()).from(tableId);
if (useIncrementalScan) {
read = read.withCdc().toSnapshot(table.currentSnapshot().snapshotId());
}
Schema expectedLegacyBeamSchema =
Schema.builder().addDateTimeField("ts").build();
Row expectedLegacyRow =
Row.withSchema(expectedLegacyBeamSchema).addValue(DateTime.parse(val)).build();
testPipeline.getOptions().as(StreamingOptions.class).setUpdateCompatibilityVersion("2.75.0");
PCollection<Row> outputLegacy = testPipeline.apply(read).apply(new
PrintRow());
PAssert.that(outputLegacy).containsInAnyOrder(expectedLegacyRow);
testPipeline.run().waitUntilFinish();
}
```
--
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]
