ahmedabu98 commented on code in PR #39344:
URL: https://github.com/apache/beam/pull/39344#discussion_r3589733368
##########
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:
Should be fine, it's a unit 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]
