From 2fb2103d4f65129dac4f51ee335f93ef0ade635b Mon Sep 17 00:00:00 2001 From: Ahmed Abualsaud Date: Wed, 15 Jul 2026 13:57:59 -0400 Subject: [PATCH 1/7] switch to new timestamp logical type --- .../apache/beam/sdk/io/iceberg/IcebergIO.java | 14 +++- .../sdk/io/iceberg/IcebergScanConfig.java | 9 ++- .../beam/sdk/io/iceberg/IcebergUtils.java | 65 ++++++++++++++----- .../sdk/io/iceberg/IncrementalScanSource.java | 4 +- .../beam/sdk/io/iceberg/ReadFromTasks.java | 4 +- .../beam/sdk/io/iceberg/ScanSource.java | 4 +- .../beam/sdk/io/iceberg/ScanTaskReader.java | 5 +- .../sdk/io/iceberg/IcebergIOReadTest.java | 48 ++++++++++++++ .../beam/sdk/io/iceberg/IcebergUtilsTest.java | 43 +++++++++++- ...ebergWriteSchemaTransformProviderTest.java | 13 +++- .../iceberg/catalog/IcebergCatalogBaseIT.java | 7 +- 11 files changed, 187 insertions(+), 29 deletions(-) diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/IcebergIO.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/IcebergIO.java index 04feea5037d1..7e930dbf406a 100644 --- a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/IcebergIO.java +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/IcebergIO.java @@ -25,6 +25,7 @@ import java.util.Map; import org.apache.beam.sdk.annotations.Internal; import org.apache.beam.sdk.io.Read; +import org.apache.beam.sdk.options.StreamingOptions; import org.apache.beam.sdk.schemas.Schema; import org.apache.beam.sdk.transforms.PTransform; import org.apache.beam.sdk.values.PBegin; @@ -668,12 +669,23 @@ public PCollection expand(PBegin input) { Table table = TableCache.get(getCatalogConfig(), tableId); + @Nullable + String updateCompatibilityVersion = + input + .getPipeline() + .getOptions() + .as(StreamingOptions.class) + .getUpdateCompatibilityVersion(); + IcebergScanConfig scanConfig = IcebergScanConfig.builder() .setCatalogConfig(getCatalogConfig()) .setScanType(IcebergScanConfig.ScanType.TABLE) .setTableIdentifier(tableId) - .setSchema(IcebergUtils.icebergSchemaToBeamSchema(table.schema())) + .setSchema( + IcebergUtils.icebergSchemaToBeamSchema( + table.schema(), updateCompatibilityVersion)) + .setUpdateCompatibilityVersion(updateCompatibilityVersion) .setFromSnapshotInclusive(getFromSnapshot()) .setToSnapshot(getToSnapshot()) .setFromTimestamp(getFromTimestamp()) diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/IcebergScanConfig.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/IcebergScanConfig.java index 95ea6cf1bd4b..d184a84edf96 100644 --- a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/IcebergScanConfig.java +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/IcebergScanConfig.java @@ -164,7 +164,8 @@ public org.apache.iceberg.Schema recordIdSchema() { public Schema rowIdBeamSchema() { if (cachedRowIdBeamSchema == null) { - cachedRowIdBeamSchema = icebergSchemaToBeamSchema(recordIdSchema()); + cachedRowIdBeamSchema = + icebergSchemaToBeamSchema(recordIdSchema(), getUpdateCompatibilityVersion()); } return cachedRowIdBeamSchema; } @@ -237,6 +238,9 @@ public Expression getFilter() { @Pure public abstract boolean getUseCdc(); + @Pure + public abstract @Nullable String getUpdateCompatibilityVersion(); + @Pure public abstract @Nullable Boolean getStreaming(); @@ -335,6 +339,9 @@ public Builder setTableIdentifier(String... names) { public abstract Builder setUseCdc(boolean useCdc); + public abstract Builder setUpdateCompatibilityVersion( + @Nullable String updateCompatibilityVersion); + public abstract Builder setStreaming(@Nullable Boolean streaming); public abstract Builder setPollInterval(@Nullable Duration pollInterval); diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/IcebergUtils.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/IcebergUtils.java index 309205707a95..0e7210de8d6c 100644 --- a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/IcebergUtils.java +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/IcebergUtils.java @@ -40,10 +40,13 @@ import org.apache.beam.sdk.schemas.logicaltypes.MicrosInstant; import org.apache.beam.sdk.schemas.logicaltypes.PassThroughLogicalType; import org.apache.beam.sdk.schemas.logicaltypes.SqlTypes; +import org.apache.beam.sdk.schemas.logicaltypes.Timestamp; import org.apache.beam.sdk.util.Preconditions; +import org.apache.beam.sdk.util.construction.TransformUpgrader; import org.apache.beam.sdk.values.PCollection; import org.apache.beam.sdk.values.Row; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting; +import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Strings; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableList; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap; import org.apache.iceberg.catalog.TableIdentifier; @@ -81,9 +84,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()) .build(); - private static Schema.FieldType icebergTypeToBeamFieldType(final Type type) { + private static Schema.FieldType icebergTypeToBeamFieldType( + final Type type, @Nullable String updateCompatibilityVersion) { switch (type.typeId()) { case BOOLEAN: return Schema.FieldType.BOOLEAN; @@ -102,7 +107,13 @@ private static Schema.FieldType icebergTypeToBeamFieldType(final Type type) { case TIMESTAMP: Types.TimestampType ts = (Types.TimestampType) type.asPrimitiveType(); if (ts.shouldAdjustToUTC()) { - return Schema.FieldType.DATETIME; + // timestamptz. The micros-precision Timestamp logical type preserves microseconds, while + // the legacy DATETIME (joda) mapping truncates to millis. Gated for update compatibility. + if (!Strings.isNullOrEmpty(updateCompatibilityVersion) + && TransformUpgrader.compareVersions(updateCompatibilityVersion, "2.76.0") < 0) { + return Schema.FieldType.DATETIME; + } + return Schema.FieldType.logicalType(Timestamp.MICROS); } return Schema.FieldType.logicalType(SqlTypes.DATETIME); case STRING: @@ -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( + final org.apache.iceberg.Schema schema, @Nullable String updateCompatibilityVersion) { Schema.Builder builder = Schema.builder(); for (Types.NestedField f : schema.columns()) { - builder.addField(icebergFieldToBeamField(f)); + builder.addField(icebergFieldToBeamField(f, updateCompatibilityVersion)); } return builder.build(); } - private static Schema icebergStructTypeToBeamSchema(final Types.StructType struct) { + private static Schema icebergStructTypeToBeamSchema( + final Types.StructType struct, @Nullable String updateCompatibilityVersion) { Schema.Builder builder = Schema.builder(); for (Types.NestedField f : struct.fields()) { - builder.addField(icebergFieldToBeamField(f)); + builder.addField(icebergFieldToBeamField(f, updateCompatibilityVersion)); } return builder.build(); } @@ -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); } } else if (icebergValue instanceof Long) { if (type.isLogicalType(SqlTypes.TIME.getIdentifier())) { return DateTimeUtil.timeFromMicros((Long) icebergValue); } else if (type.isLogicalType(SqlTypes.DATETIME.getIdentifier())) { return DateTimeUtil.timestampFromMicros((Long) icebergValue); + } else if (type.isLogicalType(Timestamp.IDENTIFIER)) { + // timestamptz stored as micros since epoch -> java.time.Instant (micros preserved). + return DateTimeUtil.timestamptzFromMicros((Long) icebergValue).toInstant(); } } else if (icebergValue instanceof Integer && type.isLogicalType(SqlTypes.DATE.getIdentifier())) { return DateTimeUtil.dateFromDays((Integer) icebergValue); - } else if (icebergValue instanceof OffsetDateTime - && type.isLogicalType(SqlTypes.DATETIME.getIdentifier())) { - return ((OffsetDateTime) icebergValue) - .withOffsetSameInstant(ZoneOffset.UTC) - .toLocalDateTime(); + } else if (icebergValue instanceof OffsetDateTime) { + OffsetDateTime odt = (OffsetDateTime) icebergValue; + if (type.isLogicalType(SqlTypes.DATETIME.getIdentifier())) { + return odt.withOffsetSameInstant(ZoneOffset.UTC).toLocalDateTime(); + } else if (type.isLogicalType(Timestamp.IDENTIFIER)) { + return odt.toInstant(); + } } // LocalDateTime, LocalDate, LocalTime return icebergValue; diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/IncrementalScanSource.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/IncrementalScanSource.java index 324eb8172760..98870095e171 100644 --- a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/IncrementalScanSource.java +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/IncrementalScanSource.java @@ -68,7 +68,9 @@ public PCollection expand(PBegin input) { .setCoder(KvCoder.of(ReadTaskDescriptor.getCoder(), ReadTask.getCoder())) .apply(Redistribute.arbitrarily()) .apply("Read Rows From Tasks", ParDo.of(new ReadFromTasks(scanConfig))) - .setRowSchema(IcebergUtils.icebergSchemaToBeamSchema(scanConfig.getProjectedSchema())); + .setRowSchema( + IcebergUtils.icebergSchemaToBeamSchema( + scanConfig.getProjectedSchema(), scanConfig.getUpdateCompatibilityVersion())); } /** Continuously watches for new snapshots. */ diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/ReadFromTasks.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/ReadFromTasks.java index 71114437731c..438e2de464d6 100644 --- a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/ReadFromTasks.java +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/ReadFromTasks.java @@ -69,7 +69,9 @@ public void process( return; } FileScanTask task = fileScanTasks.get((int) l); - Schema beamSchema = IcebergUtils.icebergSchemaToBeamSchema(scanConfig.getProjectedSchema()); + Schema beamSchema = + IcebergUtils.icebergSchemaToBeamSchema( + scanConfig.getProjectedSchema(), scanConfig.getUpdateCompatibilityVersion()); try (CloseableIterable reader = ReadUtils.createReader(task, table, scanConfig)) { for (Record record : reader) { diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/ScanSource.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/ScanSource.java index c407ef8d3e2d..d8c1780c5db3 100644 --- a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/ScanSource.java +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/ScanSource.java @@ -116,7 +116,9 @@ public void populateDisplayData(DisplayData.Builder builder) { @Override public Coder getOutputCoder() { - return RowCoder.of(IcebergUtils.icebergSchemaToBeamSchema(scanConfig.getProjectedSchema())); + return RowCoder.of( + IcebergUtils.icebergSchemaToBeamSchema( + scanConfig.getProjectedSchema(), scanConfig.getUpdateCompatibilityVersion())); } @Override diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/ScanTaskReader.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/ScanTaskReader.java index c9ad372a0751..c6ddd0a7e250 100644 --- a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/ScanTaskReader.java +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/ScanTaskReader.java @@ -66,7 +66,10 @@ class ScanTaskReader extends BoundedSource.BoundedReader { public ScanTaskReader(ScanTaskSource source) { this.source = source; - this.beamSchema = icebergSchemaToBeamSchema(source.getScanConfig().getProjectedSchema()); + this.beamSchema = + icebergSchemaToBeamSchema( + source.getScanConfig().getProjectedSchema(), + source.getScanConfig().getUpdateCompatibilityVersion()); } @Override diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/IcebergIOReadTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/IcebergIOReadTest.java index edd261458168..7920fc37c84b 100644 --- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/IcebergIOReadTest.java +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/IcebergIOReadTest.java @@ -32,7 +32,9 @@ import java.io.File; import java.io.IOException; +import java.time.OffsetDateTime; import java.util.Arrays; +import java.util.Collections; import java.util.List; import java.util.Map; import java.util.Objects; @@ -41,8 +43,11 @@ import java.util.stream.Stream; import org.apache.avro.generic.GenericData; import org.apache.avro.generic.GenericRecord; +import org.apache.beam.sdk.Pipeline; import org.apache.beam.sdk.io.iceberg.IcebergIO.ReadRows.StartingStrategy; +import org.apache.beam.sdk.options.StreamingOptions; import org.apache.beam.sdk.schemas.Schema; +import org.apache.beam.sdk.schemas.logicaltypes.Timestamp; import org.apache.beam.sdk.testing.PAssert; import org.apache.beam.sdk.testing.TestPipeline; import org.apache.beam.sdk.transforms.DoFn; @@ -83,6 +88,7 @@ import org.apache.parquet.avro.AvroParquetWriter; import org.apache.parquet.hadoop.ParquetWriter; import org.checkerframework.checker.nullness.qual.Nullable; +import org.joda.time.DateTime; import org.joda.time.Duration; import org.junit.ClassRule; import org.junit.Rule; @@ -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 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 outputLegacy = testPipeline2.apply(read).apply(new PrintRow()); + PAssert.that(outputLegacy).containsInAnyOrder(expectedLegacyRow); + testPipeline2.run().waitUntilFinish(); + } + public void runWithStartingStrategy(@Nullable StartingStrategy strategy, boolean streaming) throws IOException { assumeTrue(useIncrementalScan); diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/IcebergUtilsTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/IcebergUtilsTest.java index 3da31ecc2061..7e707717f3cf 100644 --- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/IcebergUtilsTest.java +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/IcebergUtilsTest.java @@ -42,6 +42,7 @@ import org.apache.beam.sdk.schemas.logicaltypes.FixedPrecisionNumeric; import org.apache.beam.sdk.schemas.logicaltypes.FixedString; import org.apache.beam.sdk.schemas.logicaltypes.SqlTypes; +import org.apache.beam.sdk.schemas.logicaltypes.Timestamp; import org.apache.beam.sdk.schemas.logicaltypes.UuidLogicalType; import org.apache.beam.sdk.schemas.logicaltypes.VariableBytes; import org.apache.beam.sdk.schemas.logicaltypes.VariableString; @@ -232,6 +233,13 @@ public void testTimestampWithZone() { OffsetDateTime offsetDateTime = OffsetDateTime.parse(val); LocalDateTime localDateTime = offsetDateTime.withOffsetSameInstant(ZoneOffset.UTC).toLocalDateTime(); + // Timestamp.MICROS + checkRowValueToRecordValue( + Schema.FieldType.logicalType(Timestamp.MICROS), + offsetDateTime.toInstant(), + Types.TimestampType.withZone(), + offsetDateTime.withOffsetSameInstant(ZoneOffset.UTC)); + // SqlTypes.DATETIME checkRowValueToRecordValue( Schema.FieldType.logicalType(SqlTypes.DATETIME), @@ -426,6 +434,24 @@ public void testTimestampWithZone() { OffsetDateTime offsetDateTime = OffsetDateTime.parse(timestamp); LocalDateTime localDateTime = offsetDateTime.withOffsetSameInstant(ZoneOffset.UTC).toLocalDateTime(); + + // Timestamp.MICROS + checkRecordValueToRowValue( + Types.TimestampType.withZone(), + offsetDateTime, + Schema.FieldType.logicalType(Timestamp.MICROS), + offsetDateTime.toInstant()); + checkRecordValueToRowValue( + Types.TimestampType.withZone(), + DateTimeUtil.microsFromTimestamptz(offsetDateTime), + Schema.FieldType.logicalType(Timestamp.MICROS), + offsetDateTime.toInstant()); + checkRecordValueToRowValue( + Types.TimestampType.withZone(), + timestamp, + Schema.FieldType.logicalType(Timestamp.MICROS), + offsetDateTime.toInstant()); + // SqlTypes.DATETIME checkRecordValueToRowValue( Types.TimestampType.withZone(), @@ -458,6 +484,21 @@ public void testTimestampWithZone() { Types.TimestampType.withZone(), timestamp, Schema.FieldType.DATETIME, dateTime); } + @Test + public void testUpdateCompatibilityVersionGatesTimestamptzMapping() { + org.apache.iceberg.Schema icebergSchema = + new org.apache.iceberg.Schema(required(0, "ts", Types.TimestampType.withZone())); + + // A pinned, older update-compatibility version keeps the legacy DATETIME mapping on read. + Schema pinnedOld = IcebergUtils.icebergSchemaToBeamSchema(icebergSchema, "2.50.0"); + assertEquals(Schema.FieldType.DATETIME, pinnedOld.getField("ts").getType()); + + // An unset (null) or future update-compatibility version uses the new micros mapping. + Schema unpinned = IcebergUtils.icebergSchemaToBeamSchema(icebergSchema, null); + assertEquals( + Schema.FieldType.logicalType(Timestamp.MICROS), unpinned.getField("ts").getType()); + } + @Test public void testFixed() {} @@ -868,7 +909,7 @@ public void testMapBeamFieldTypeToIcebergFieldType() { .addNullableStringField("str") .addNullableBooleanField("bool") .addByteArrayField("bytes") - .addDateTimeField("datetime_tz") + .addLogicalTypeField("datetime_tz", Timestamp.MICROS) .addLogicalTypeField("datetime", SqlTypes.DATETIME) .addLogicalTypeField("time", SqlTypes.TIME) .addLogicalTypeField("date", SqlTypes.DATE) diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/IcebergWriteSchemaTransformProviderTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/IcebergWriteSchemaTransformProviderTest.java index c5fc5a6b6fe7..5a7aa11e10a9 100644 --- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/IcebergWriteSchemaTransformProviderTest.java +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/IcebergWriteSchemaTransformProviderTest.java @@ -38,6 +38,8 @@ import java.util.UUID; import org.apache.beam.sdk.Pipeline; import org.apache.beam.sdk.managed.Managed; +import org.apache.beam.sdk.options.PipelineOptions; +import org.apache.beam.sdk.options.StreamingOptions; import org.apache.beam.sdk.schemas.Schema; import org.apache.beam.sdk.schemas.logicaltypes.SqlTypes; import org.apache.beam.sdk.testing.PAssert; @@ -484,7 +486,12 @@ public void writePartitionedData(boolean autosharding) { .satisfies(new VerifyOutputs(Collections.singletonList(identifier), "append")); testPipeline.run().waitUntilFinish(); - Pipeline p = Pipeline.create(TestPipeline.testingPipelineOptions()); + // This table has timestamptz columns written as joda DateTime. Pin an older update + // compatibility version so the read keeps the legacy DATETIME mapping and matches the written + // rows + PipelineOptions readOptions = TestPipeline.testingPipelineOptions(); + readOptions.as(StreamingOptions.class).setUpdateCompatibilityVersion("2.75.0"); + Pipeline p = Pipeline.create(readOptions); PCollection readRows = p.apply(Managed.read(Managed.ICEBERG).withConfig(config)).getSinglePCollection(); PAssert.that(readRows).containsInAnyOrder(rows); @@ -554,7 +561,9 @@ public void testWriteCreateTableWithPartitionSpec() { .satisfies(new VerifyOutputs(Collections.singletonList(identifier), "append")); testPipeline.run().waitUntilFinish(); - Pipeline p = Pipeline.create(TestPipeline.testingPipelineOptions()); + PipelineOptions readOptions = TestPipeline.testingPipelineOptions(); + readOptions.as(StreamingOptions.class).setUpdateCompatibilityVersion("2.75.0"); + Pipeline p = Pipeline.create(readOptions); PCollection readRows = p.apply(Managed.read(Managed.ICEBERG).withConfig(config)).getSinglePCollection(); PAssert.that(readRows).containsInAnyOrder(rows); diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/catalog/IcebergCatalogBaseIT.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/catalog/IcebergCatalogBaseIT.java index f0c7ae925df7..36f50b6e7da0 100644 --- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/catalog/IcebergCatalogBaseIT.java +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/catalog/IcebergCatalogBaseIT.java @@ -59,6 +59,7 @@ import org.apache.beam.sdk.managed.Managed; import org.apache.beam.sdk.schemas.Schema; import org.apache.beam.sdk.schemas.logicaltypes.SqlTypes; +import org.apache.beam.sdk.schemas.logicaltypes.Timestamp; import org.apache.beam.sdk.testing.PAssert; import org.apache.beam.sdk.testing.TestPipeline; import org.apache.beam.sdk.transforms.Create; @@ -106,8 +107,6 @@ import org.apache.iceberg.util.DateTimeUtil; import org.apache.iceberg.util.PartitionUtil; import org.checkerframework.checker.nullness.qual.Nullable; -import org.joda.time.DateTime; -import org.joda.time.DateTimeZone; import org.joda.time.Duration; import org.joda.time.Instant; import org.joda.time.LocalDate; @@ -272,7 +271,7 @@ public void cleanUp() throws Exception { .addArrayField("arr_long", Schema.FieldType.INT64) .addNullableRowField("nullable_row", NESTED_ROW_SCHEMA) .addNullableInt64Field("nullable_long") - .addDateTimeField("datetime_tz") + .addLogicalTypeField("datetime_tz", Timestamp.MICROS) .addLogicalTypeField("datetime", SqlTypes.DATETIME) .addLogicalTypeField("date", SqlTypes.DATE) .addLogicalTypeField("time", SqlTypes.TIME) @@ -309,7 +308,7 @@ public Row apply(Long num) { .addValue(LongStream.range(0, num % 10).boxed().collect(Collectors.toList())) .addValue(num % 2 == 0 ? null : nestedRow) .addValue(num) - .addValue(new DateTime(timestampMillis).withZone(DateTimeZone.forOffsetHours(4))) + .addValue(DateTimeUtil.timestamptzFromMicros(timestampMillis * 1000).toInstant()) .addValue(DateTimeUtil.timestampFromMicros(timestampMillis * 1000)) .addValue(DateTimeUtil.dateFromDays(Integer.parseInt(strNum))) .addValue(DateTimeUtil.timeFromMicros(num)) From fa43293bb4afe5d4f34fec29bb82e35cdf02376d Mon Sep 17 00:00:00 2001 From: Ahmed Abualsaud Date: Wed, 15 Jul 2026 14:02:40 -0400 Subject: [PATCH 2/7] changes and trigger ITs --- .github/trigger_files/IO_Iceberg_Integration_Tests.json | 2 +- CHANGES.md | 4 ++++ 2 files changed, 5 insertions(+), 1 deletion(-) diff --git a/.github/trigger_files/IO_Iceberg_Integration_Tests.json b/.github/trigger_files/IO_Iceberg_Integration_Tests.json index 7ab7bcd9a9c6..37dd25bf9029 100644 --- a/.github/trigger_files/IO_Iceberg_Integration_Tests.json +++ b/.github/trigger_files/IO_Iceberg_Integration_Tests.json @@ -1,4 +1,4 @@ { "comment": "Modify this file in a trivial way to cause this test suite to run.", - "modification": 2 + "modification": 3 } diff --git a/CHANGES.md b/CHANGES.md index a6fd20ad34dd..86027753f25f 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -74,6 +74,10 @@ ## Breaking Changes * (Python) Removed `google-perftools` from the SDK container images. Users who wish to use `--profiler_agent=tcmalloc` should install google-perftools APT package in their custom container images separately ([#39323](https://github.com/apache/beam/issues/39323)). +* [IcebergIO] Reading a `timestamptz` column will now return a `Timestamp.MICROS` Beam logical type, equivalent to a + `java.time.Instant` object, in order to preserve microseconds (the old Beam `Schema.FieldType#DATETIME` primitive type + truncates past milliseconds). Use pipeline option `--updateCompatibilityVersion=2.75.0` (or any older version) to + keep the old behavior. ## Deprecations From 88064c773711982097d9a76fd2f7d4e46cdfacff Mon Sep 17 00:00:00 2001 From: Ahmed Abualsaud Date: Wed, 15 Jul 2026 14:17:45 -0400 Subject: [PATCH 3/7] address comments --- .../java/org/apache/beam/sdk/io/iceberg/IcebergUtils.java | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/IcebergUtils.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/IcebergUtils.java index 0e7210de8d6c..3154557b17ee 100644 --- a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/IcebergUtils.java +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/IcebergUtils.java @@ -46,7 +46,6 @@ import org.apache.beam.sdk.values.PCollection; import org.apache.beam.sdk.values.Row; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting; -import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Strings; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableList; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap; import org.apache.iceberg.catalog.TableIdentifier; @@ -109,7 +108,8 @@ private static Schema.FieldType icebergTypeToBeamFieldType( if (ts.shouldAdjustToUTC()) { // timestamptz. The micros-precision Timestamp logical type preserves microseconds, while // the legacy DATETIME (joda) mapping truncates to millis. Gated for update compatibility. - if (!Strings.isNullOrEmpty(updateCompatibilityVersion) + if (updateCompatibilityVersion != null + && !updateCompatibilityVersion.isEmpty() && TransformUpgrader.compareVersions(updateCompatibilityVersion, "2.76.0") < 0) { return Schema.FieldType.DATETIME; } @@ -640,7 +640,7 @@ private static Object getLogicalTypeValue(Object icebergValue, Schema.FieldType } else if (type.isLogicalType(SqlTypes.DATETIME.getIdentifier())) { return LocalDateTime.parse(strValue); } else if (type.isLogicalType(Timestamp.IDENTIFIER)) { - return java.time.Instant.parse(strValue); + return OffsetDateTime.parse(strValue).toInstant(); } } else if (icebergValue instanceof Long) { if (type.isLogicalType(SqlTypes.TIME.getIdentifier())) { From 14d0e04b57d60f3182666962a08596ce4aaf2c04 Mon Sep 17 00:00:00 2001 From: Ahmed Abualsaud Date: Wed, 15 Jul 2026 14:59:32 -0400 Subject: [PATCH 4/7] format changes --- CHANGES.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/CHANGES.md b/CHANGES.md index 86027753f25f..aac63eff9bbd 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -2511,4 +2511,4 @@ Schema Options, it will be removed in version `2.23.0`. ([BEAM-9704](https://iss ## Highlights -- For versions 2.19.0 and older release notes are available on [Apache Beam Blog](https://beam.apache.org/blog/). +- For versions 2.19.0 and older release notes are available on [Apache Beam Blog](https://beam.apache.org/blog/). \ No newline at end of file From 2ada632f8df6486db04d9c126ad1f4898a3d31a6 Mon Sep 17 00:00:00 2001 From: Ahmed Abualsaud Date: Wed, 15 Jul 2026 15:20:24 -0400 Subject: [PATCH 5/7] add link --- CHANGES.md | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/CHANGES.md b/CHANGES.md index aac63eff9bbd..4406abacfcab 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -77,7 +77,7 @@ * [IcebergIO] Reading a `timestamptz` column will now return a `Timestamp.MICROS` Beam logical type, equivalent to a `java.time.Instant` object, in order to preserve microseconds (the old Beam `Schema.FieldType#DATETIME` primitive type truncates past milliseconds). Use pipeline option `--updateCompatibilityVersion=2.75.0` (or any older version) to - keep the old behavior. + keep the old behavior ([#39344](https://github.com/apache/beam/issues/39344)). ## Deprecations @@ -2511,4 +2511,4 @@ Schema Options, it will be removed in version `2.23.0`. ([BEAM-9704](https://iss ## Highlights -- For versions 2.19.0 and older release notes are available on [Apache Beam Blog](https://beam.apache.org/blog/). \ No newline at end of file +- For versions 2.19.0 and older release notes are available on [Apache Beam Blog](https://beam.apache.org/blog/). From 69a4d11b242c8242535f5caed469ff3a67b2e70b Mon Sep 17 00:00:00 2001 From: Ahmed Abualsaud Date: Wed, 15 Jul 2026 20:58:28 -0500 Subject: [PATCH 6/7] tighten BQ conversion logic --- .../sdk/io/gcp/bigquery/BigQueryUtils.java | 9 ++++- .../io/gcp/bigquery/BigQueryUtilsTest.java | 33 ++++++++++++------- .../catalog/BigQueryMetastoreCatalogIT.java | 3 ++ .../iceberg/catalog/IcebergCatalogBaseIT.java | 4 +-- 4 files changed, 35 insertions(+), 14 deletions(-) diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryUtils.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryUtils.java index d9805a6f4e06..d0ce05bdf8fb 100644 --- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryUtils.java +++ b/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 { + BigDecimal bd = new BigDecimal(jsonBQString); + long seconds = bd.longValue(); + long nanos = bd.remainder(BigDecimal.ONE).movePointRight(9).longValue(); + return java.time.Instant.ofEpochSecond(seconds, nanos); + } catch (NumberFormatException e) { + return VAR_PRECISION_FORMATTER.parse(jsonBQString, java.time.Instant::from); + } } } diff --git a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryUtilsTest.java b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryUtilsTest.java index b50e8448698a..2092d9188b8c 100644 --- a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryUtilsTest.java +++ b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryUtilsTest.java @@ -41,6 +41,7 @@ import java.time.LocalDate; import java.time.LocalDateTime; import java.time.LocalTime; +import java.time.OffsetDateTime; import java.util.Arrays; import java.util.Base64; import java.util.Collections; @@ -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"; + java.time.Instant instant = OffsetDateTime.parse(parsableTimestamp).toInstant(); + long seconds = instant.getEpochSecond(); + long micros = instant.getNano() / 1000; // BQ stores in micros precision + String value = seconds + "." + micros; + + List testRows = + Arrays.asList(new TableRow().set("ts", timestamp), new TableRow().set("ts", value)); + + for (TableRow row : testRows) { + Row beamRow = BigQueryUtils.toBeamRow(schema, row); + + 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()); + } } @Test diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/catalog/BigQueryMetastoreCatalogIT.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/catalog/BigQueryMetastoreCatalogIT.java index eb3ebfbf5219..2637b78665e1 100644 --- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/catalog/BigQueryMetastoreCatalogIT.java +++ b/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") .build(), new Configuration()); } @@ -115,6 +116,8 @@ public void testWriteToPartitionedAndValidateWithBQQuery() .map(tr -> BigQueryUtils.toBeamRow(BEAM_SCHEMA, tr)) .collect(Collectors.toList()); + System.out.println("returned: " + beamRows); + System.out.println("expected: " + inputRows); assertThat(beamRows, containsInAnyOrder(inputRows.toArray())); String queryByPartition = diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/catalog/IcebergCatalogBaseIT.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/catalog/IcebergCatalogBaseIT.java index 36f50b6e7da0..7a3cfc9f6bbc 100644 --- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/catalog/IcebergCatalogBaseIT.java +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/catalog/IcebergCatalogBaseIT.java @@ -308,8 +308,8 @@ public Row apply(Long num) { .addValue(LongStream.range(0, num % 10).boxed().collect(Collectors.toList())) .addValue(num % 2 == 0 ? null : nestedRow) .addValue(num) - .addValue(DateTimeUtil.timestamptzFromMicros(timestampMillis * 1000).toInstant()) - .addValue(DateTimeUtil.timestampFromMicros(timestampMillis * 1000)) + .addValue(DateTimeUtil.timestamptzFromMicros(timestampMillis * 1000 + 123456789).toInstant()) + .addValue(DateTimeUtil.timestampFromMicros(timestampMillis * 1000 + 123456789)) .addValue(DateTimeUtil.dateFromDays(Integer.parseInt(strNum))) .addValue(DateTimeUtil.timeFromMicros(num)) .build(); From d18d8f849cfe0e23aa1553a4854d047828019082 Mon Sep 17 00:00:00 2001 From: Ahmed Abualsaud Date: Wed, 15 Jul 2026 23:01:05 -0400 Subject: [PATCH 7/7] spotless --- .../beam/sdk/io/iceberg/catalog/IcebergCatalogBaseIT.java | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/catalog/IcebergCatalogBaseIT.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/catalog/IcebergCatalogBaseIT.java index 7a3cfc9f6bbc..5c28f0192a61 100644 --- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/catalog/IcebergCatalogBaseIT.java +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/catalog/IcebergCatalogBaseIT.java @@ -308,7 +308,9 @@ public Row apply(Long num) { .addValue(LongStream.range(0, num % 10).boxed().collect(Collectors.toList())) .addValue(num % 2 == 0 ? null : nestedRow) .addValue(num) - .addValue(DateTimeUtil.timestamptzFromMicros(timestampMillis * 1000 + 123456789).toInstant()) + .addValue( + DateTimeUtil.timestamptzFromMicros(timestampMillis * 1000 + 123456789) + .toInstant()) .addValue(DateTimeUtil.timestampFromMicros(timestampMillis * 1000 + 123456789)) .addValue(DateTimeUtil.dateFromDays(Integer.parseInt(strNum))) .addValue(DateTimeUtil.timeFromMicros(num))