From 7dda2cdb72d036c07d14713a22ccf7e8077b2a01 Mon Sep 17 00:00:00 2001 From: feefs Date: Tue, 11 Aug 2026 10:13:31 -0700 Subject: [PATCH 1/4] Enable Apache Iceberg REST Metrics Reporting for Lakehouse (#39650) --- .../test/java/org/apache/beam/sdk/io/iceberg/AddFilesIT.java | 3 +-- .../apache/beam/sdk/io/iceberg/catalog/RESTCatalogBLMSIT.java | 1 - sdks/python/apache_beam/transforms/managed_iceberg_it_test.py | 3 +-- 3 files changed, 2 insertions(+), 5 deletions(-) diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesIT.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesIT.java index 54c9cb8dc93f..528d6453eac2 100644 --- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesIT.java +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesIT.java @@ -126,8 +126,7 @@ public class AddFilesIT { "warehouse", WAREHOUSE, "header.x-goog-user-project", PROJECT, "rest.auth.type", "google", - "io-impl", "org.apache.iceberg.gcp.gcs.GCSFileIO", - "rest-metrics-reporting-enabled", "false"); + "io-impl", "org.apache.iceberg.gcp.gcs.GCSFileIO"); private Storage storage; private PubsubClient pubsub; private Notification notification; diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/catalog/RESTCatalogBLMSIT.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/catalog/RESTCatalogBLMSIT.java index c16df763333f..6934bf13f21c 100644 --- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/catalog/RESTCatalogBLMSIT.java +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/catalog/RESTCatalogBLMSIT.java @@ -41,7 +41,6 @@ public static void setup() { .put("uri", "https://biglake.googleapis.com/iceberg/v1/restcatalog") .put("warehouse", BIGLAKE_WAREHOUSE) .put("header.x-goog-user-project", OPTIONS.getProject()) - .put("rest-metrics-reporting-enabled", "false") .put("io-impl", "org.apache.iceberg.gcp.gcs.GCSFileIO") .put("rest.auth.type", "org.apache.iceberg.gcp.auth.GoogleAuthManager") .build(); diff --git a/sdks/python/apache_beam/transforms/managed_iceberg_it_test.py b/sdks/python/apache_beam/transforms/managed_iceberg_it_test.py index 23d19c504970..b381b1960f38 100644 --- a/sdks/python/apache_beam/transforms/managed_iceberg_it_test.py +++ b/sdks/python/apache_beam/transforms/managed_iceberg_it_test.py @@ -63,8 +63,7 @@ def test_write_read_pipeline(self): 'header.x-goog-user-project': 'apache-beam-testing', 'rest.auth.type': 'google', 'io-impl': 'org.apache.iceberg.gcp.gcs.GCSFileIO', - 'header.X-Iceberg-Access-Delegation': 'vended-credentials', - 'rest-metrics-reporting-enabled': 'false' + 'header.X-Iceberg-Access-Delegation': 'vended-credentials' } iceberg_config = { "table": "test_iceberg_write_read.test_" + uuid.uuid4().hex, From ce45298a609064e6d5d4264088c2e658a028a22a Mon Sep 17 00:00:00 2001 From: Chamikara Jayalath Date: Tue, 11 Aug 2026 10:41:51 -0700 Subject: [PATCH 2/4] Fixes to delta CDC read (#39713) --- .../beam/sdk/io/delta/DeltaCDCSourceDoFn.java | 24 +++-- .../org/apache/beam/sdk/io/delta/DeltaIO.java | 4 +- .../apache/beam/sdk/io/delta/DeltaIOTest.java | 98 +++++++++++++++++++ 3 files changed, 115 insertions(+), 11 deletions(-) diff --git a/sdks/java/io/delta/src/main/java/org/apache/beam/sdk/io/delta/DeltaCDCSourceDoFn.java b/sdks/java/io/delta/src/main/java/org/apache/beam/sdk/io/delta/DeltaCDCSourceDoFn.java index 414402429c38..97d9c914a086 100644 --- a/sdks/java/io/delta/src/main/java/org/apache/beam/sdk/io/delta/DeltaCDCSourceDoFn.java +++ b/sdks/java/io/delta/src/main/java/org/apache/beam/sdk/io/delta/DeltaCDCSourceDoFn.java @@ -231,7 +231,7 @@ public ColumnarBatch next() { "Field " + DeltaIO.CHANGE_TYPE_COLUMN + " must not be null."); } ValueKind kind = getValueKind(changeType); - Row publicRow = projectRow(beamRow, publicBeamSchema); + Row publicRow = projectRow(beamRow, publicBeamSchema, task); out.builder(publicRow).setValueKind(kind).output(); } } @@ -240,14 +240,18 @@ public ColumnarBatch next() { } } - private static Row projectRow(Row row, Schema targetSchema) { - if (row.getSchema().equals(targetSchema)) { - // We can return the original Row since schemas are the same. - return row; - } + private static Row projectRow(Row row, Schema targetSchema, DeltaCDCReadTask task) { Row.Builder builder = Row.withSchema(targetSchema); for (Schema.Field field : targetSchema.getFields()) { - builder.addValue(row.getValue(field.getName())); + Object value = row.getValue(field.getName()); + if (value == null) { + if (field.getName().equals(DeltaIO.COMMIT_VERSION_COLUMN)) { + value = task.getVersion(); + } else if (field.getName().equals(DeltaIO.COMMIT_TIMESTAMP_COLUMN)) { + value = new org.joda.time.Instant(task.getTimestamp()); + } + } + builder.addValue(value); } return builder.build(); } @@ -271,9 +275,9 @@ private static ValueKind getValueKind(String changeType) { private static StructType appendCDFColumns(StructType schema) { return schema - .add(DeltaIO.CHANGE_TYPE_COLUMN, StringType.STRING, false) - .add(DeltaIO.COMMIT_VERSION_COLUMN, LongType.LONG, false) - .add(DeltaIO.COMMIT_TIMESTAMP_COLUMN, TimestampType.TIMESTAMP, false); + .add(DeltaIO.CHANGE_TYPE_COLUMN, StringType.STRING, true) + .add(DeltaIO.COMMIT_VERSION_COLUMN, LongType.LONG, true) + .add(DeltaIO.COMMIT_TIMESTAMP_COLUMN, TimestampType.TIMESTAMP, true); } private ColumnarBatch appendConstantCDFColumns( diff --git a/sdks/java/io/delta/src/main/java/org/apache/beam/sdk/io/delta/DeltaIO.java b/sdks/java/io/delta/src/main/java/org/apache/beam/sdk/io/delta/DeltaIO.java index 8057332ddce4..3a53b5c76202 100644 --- a/sdks/java/io/delta/src/main/java/org/apache/beam/sdk/io/delta/DeltaIO.java +++ b/sdks/java/io/delta/src/main/java/org/apache/beam/sdk/io/delta/DeltaIO.java @@ -168,7 +168,9 @@ public PCollection expand(PBegin input) { static Schema convertToBeamSchema(StructType deltaSchema) { Schema.Builder builder = Schema.builder(); for (StructField field : deltaSchema.fields()) { - builder.addField(field.getName(), convertToBeamFieldType(field.getDataType())); + builder.addField( + Schema.Field.of(field.getName(), convertToBeamFieldType(field.getDataType())) + .withNullable(field.isNullable())); } return builder.build(); } diff --git a/sdks/java/io/delta/src/test/java/org/apache/beam/sdk/io/delta/DeltaIOTest.java b/sdks/java/io/delta/src/test/java/org/apache/beam/sdk/io/delta/DeltaIOTest.java index b9b2a604a4b0..0db0aef9e080 100644 --- a/sdks/java/io/delta/src/test/java/org/apache/beam/sdk/io/delta/DeltaIOTest.java +++ b/sdks/java/io/delta/src/test/java/org/apache/beam/sdk/io/delta/DeltaIOTest.java @@ -451,6 +451,26 @@ public void testConvertToBeamSchema() { org.junit.Assert.assertEquals(expectedSchema, actualSchema); } + @Test + public void testConvertToBeamSchemaPreservesNullability() { + StructType deltaSchema = + new StructType( + java.util.Arrays.asList( + new StructField("nullable_string", StringType.STRING, true), + new StructField("non_nullable_integer", IntegerType.INTEGER, false))); + + Schema expectedSchema = + Schema.builder() + .addField( + Schema.Field.of("nullable_string", Schema.FieldType.STRING).withNullable(true)) + .addField( + Schema.Field.of("non_nullable_integer", Schema.FieldType.INT32).withNullable(false)) + .build(); + + Schema actualSchema = DeltaIO.ReadRows.convertToBeamSchema(deltaSchema); + org.junit.Assert.assertEquals(expectedSchema, actualSchema); + } + @Test public void testDeltaReadTaskTracker() { java.util.List sizes = java.util.Arrays.asList(100L, 200L, 300L); @@ -1090,6 +1110,84 @@ public void testReadChangesWithMetadataColumns() throws Exception { readPipeline.run().waitUntilFinish(); } + @Test + public void testReadChangesWithMissingMetadataColumns() throws Exception { + File tableDir = tempFolder.newFolder("delta-table-changes-missing-metadata"); + Engine engine = DefaultEngine.create(new org.apache.hadoop.conf.Configuration()); + + // 1. Write parquet files for Version 0 (insert-only commit) + Schema tableSchema = Schema.builder().addField("name", Schema.FieldType.STRING).build(); + Row tableRow1 = Row.withSchema(tableSchema).addValues("row-1").build(); + Row tableRow2 = Row.withSchema(tableSchema).addValues("row-2").build(); + StructType deltaSchema = new StructType().add("name", StringType.STRING); + + DeltaWriteTestUtils.writeAppendCommit( + engine, + tableDir.getAbsolutePath(), + 0L, + 100000000000L, + deltaSchema, + java.util.Arrays.asList(tableRow1, tableRow2)); + + // 2. Write cdc parquet file for Version 1 (commit with cdc actions), but OMIT version and + // timestamp columns! + Schema cdcWriteSchema = + Schema.builder() + .addField("name", Schema.FieldType.STRING) + .addField(DeltaIO.CHANGE_TYPE_COLUMN, Schema.FieldType.STRING) + .build(); + StructType cdcWriteDeltaSchema = + new StructType() + .add("name", StringType.STRING) + .add(DeltaIO.CHANGE_TYPE_COLUMN, StringType.STRING); + + Row cdcRow1 = Row.withSchema(cdcWriteSchema).addValues("row-1", "update_preimage").build(); + Row cdcRow2 = + Row.withSchema(cdcWriteSchema).addValues("row-1-updated", "update_postimage").build(); + Row cdcRow3 = Row.withSchema(cdcWriteSchema).addValues("row-2", "delete").build(); + + DeltaWriteTestUtils.writeCdcCommit( + engine, + tableDir.getAbsolutePath(), + 1L, + 200000000000L, + deltaSchema, + null, + null, + java.util.Arrays.asList(cdcRow1, cdcRow2, cdcRow3), + cdcWriteDeltaSchema); + + // 3. Read CDF data from table requesting metadata columns + DeltaCdcReadSchemaTransformProvider.Configuration config = + DeltaCdcReadSchemaTransformProvider.Configuration.builder() + .setTable(tableDir.getAbsolutePath()) + .setStartVersion(0L) + .setIncludeMetadataColumns( + java.util.Arrays.asList( + DeltaIO.CHANGE_TYPE_COLUMN, + DeltaIO.COMMIT_VERSION_COLUMN, + DeltaIO.COMMIT_TIMESTAMP_COLUMN)) + .build(); + + PCollection output = + PCollectionRowTuple.empty(readPipeline) + .apply(new DeltaCdcReadSchemaTransformProvider().from(config)) + .get(DeltaCdcReadSchemaTransformProvider.OUTPUT_TAG); + + PCollection formattedOutput = + output.apply("Format Row with Metadata", ParDo.of(new FormatRowWithMetadata())); + + PAssert.that(formattedOutput) + .containsInAnyOrder( + "row-1:insert:v0:t100000000000", + "row-2:insert:v0:t100000000000", + "row-1:update_preimage:v1:t200000000000", + "row-1-updated:update_postimage:v1:t200000000000", + "row-2:delete:v1:t200000000000"); + + readPipeline.run().waitUntilFinish(); + } + @Test public void testReadChangesWithSubsetOfMetadataColumns() throws Exception { File tableDir = tempFolder.newFolder("delta-table-changes-subset-metadata"); From e344ec03fb7b3bdac74471d7356a687905da6008 Mon Sep 17 00:00:00 2001 From: claudevdm <33973061+claudevdm@users.noreply.github.com> Date: Tue, 11 Aug 2026 14:03:06 -0400 Subject: [PATCH 3/4] Python timestamp fixes. (#39722) --- sdks/python/apache_beam/io/gcp/pubsub_test.py | 42 +++++++++++++++++++ .../runners/direct/transform_evaluator.py | 5 +++ sdks/python/apache_beam/typehints/schemas.py | 10 ++++- .../apache_beam/typehints/schemas_test.py | 10 +++++ sdks/python/apache_beam/utils/timestamp.py | 5 ++- .../apache_beam/utils/timestamp_test.py | 17 ++++++++ 6 files changed, 86 insertions(+), 3 deletions(-) diff --git a/sdks/python/apache_beam/io/gcp/pubsub_test.py b/sdks/python/apache_beam/io/gcp/pubsub_test.py index 050a69aff6cf..5a8f2bf617b8 100644 --- a/sdks/python/apache_beam/io/gcp/pubsub_test.py +++ b/sdks/python/apache_beam/io/gcp/pubsub_test.py @@ -25,6 +25,7 @@ import hamcrest as hc import mock +import pytest import apache_beam as beam from apache_beam import Pipeline @@ -712,6 +713,47 @@ def test_read_messages_timestamp_attribute_rfc3339_success(self, mock_pubsub): mock_pubsub.return_value.close.assert_not_called() + @pytest.mark.timeout(60) + def test_read_messages_timestamp_attribute_sub_micro_rfc3339( + self, mock_pubsub): + # Publishers may emit 7-9 fractional digits. Sub-microsecond digits + # must be truncated when the attribute is parsed; element timestamps + # are limited to microsecond resolution and messages are acked before + # the bundle is output. + data = b'data' + attributes = {'time': '2018-03-12T13:37:01.2345678Z'} + publish_time_secs = 1337000000 + publish_time_nanos = 133700000 + ack_id = 'ack_id' + pull_response = test_utils.create_pull_response([ + test_utils.PullResponseMessage( + data, attributes, publish_time_secs, publish_time_nanos, ack_id) + ]) + expected_elements = [ + TestWindowedValue( + PubsubMessage(data, attributes), + timestamp.Timestamp(1520861821, micros=234567), + [window.GlobalWindow()]), + ] + mock_pubsub.return_value.pull.return_value = pull_response + + options = PipelineOptions([]) + options.view_as(StandardOptions).streaming = True + with TestPipeline(options=options) as p: + pcoll = ( + p + | ReadFromPubSub( + 'projects/fakeprj/topics/a_topic', + None, + None, + with_attributes=True, + timestamp_attribute='time')) + assert_that(pcoll, equal_to(expected_elements), reify_windows=True) + mock_pubsub.return_value.acknowledge.assert_has_calls( + [mock.call(subscription=mock.ANY, ack_ids=[ack_id])]) + + mock_pubsub.return_value.close.assert_not_called() + def test_read_messages_timestamp_attribute_missing(self, mock_pubsub): data = b'data' attributes = {} diff --git a/sdks/python/apache_beam/runners/direct/transform_evaluator.py b/sdks/python/apache_beam/runners/direct/transform_evaluator.py index 6702ec3362bc..d60c840e8c74 100644 --- a/sdks/python/apache_beam/runners/direct/transform_evaluator.py +++ b/sdks/python/apache_beam/runners/direct/transform_evaluator.py @@ -724,6 +724,11 @@ def _get_element(message): timestamp = Timestamp.from_rfc3339(rfc3339_or_milli) except ValueError as e: raise ValueError('Bad timestamp value: %s' % e) + if timestamp.precision() > Timestamp.MICROS_PRECISION: + # Element timestamps are limited to microsecond resolution, so + # ignore sub-microsecond digits, as the Dataflow service does. + timestamp = timestamp.to_precision( + Timestamp.MICROS_PRECISION, allow_lossy_conversion=True) else: if message.publish_time is None: raise ValueError('No publish time present in message: %s' % message) diff --git a/sdks/python/apache_beam/typehints/schemas.py b/sdks/python/apache_beam/typehints/schemas.py index 2fd3c22e1e58..80bee60ec952 100644 --- a/sdks/python/apache_beam/typehints/schemas.py +++ b/sdks/python/apache_beam/typehints/schemas.py @@ -1025,7 +1025,12 @@ class ParameterizedTimestamp(LogicalType[Timestamp, Timestamp to this logical type, re-register using :func:`~LogicalType.register_logical_type(ParameterizedTimestamp)`. """ - def __init__(self, precision: int = Timestamp.MICROS_PRECISION) -> None: + def __init__(self, precision: Optional[int] = None) -> None: + if precision is None: + # A timestamp:v1 proto without its precision argument is malformed; + # decoding at a guessed precision would silently misscale subseconds. + raise ValueError( + 'beam:logical_type:timestamp:v1 requires a precision argument.') # The argument arrives as np.int32 when decoded from a schema proto. precision = int(precision) if not 0 <= precision <= Timestamp.NANOS_PRECISION: @@ -1077,7 +1082,8 @@ def argument(self): @classmethod def _from_typing(cls, typ): - return cls() + # A bare Timestamp typehint has no precision; default to micros. + return cls(Timestamp.MICROS_PRECISION) @LogicalType._register_internal diff --git a/sdks/python/apache_beam/typehints/schemas_test.py b/sdks/python/apache_beam/typehints/schemas_test.py index 327fe7947ca5..5e66a491090d 100644 --- a/sdks/python/apache_beam/typehints/schemas_test.py +++ b/sdks/python/apache_beam/typehints/schemas_test.py @@ -879,6 +879,16 @@ def test_to_representation_type_guards_against_precision_loss(self): representation = logical_type.to_representation_type(millis_value) self.assertEqual(representation.subseconds, 500000) + def test_from_runner_api_rejects_missing_argument(self): + # A proto without the precision argument must be rejected; guessing a + # default precision would silently misscale subseconds. + proto = schema_pb2.LogicalType( + urn=common_urns.timestamp.urn, + representation=typing_to_runner_api( + schemas.ParameterizedTimestampShortRepresentation)) + with self.assertRaises(ValueError): + schemas.LogicalType.from_runner_api(proto) + class HypothesisTest(unittest.TestCase): # There is considerable variablility in runtime for this test, disable diff --git a/sdks/python/apache_beam/utils/timestamp.py b/sdks/python/apache_beam/utils/timestamp.py index 2953541b42f4..925467044f34 100644 --- a/sdks/python/apache_beam/utils/timestamp.py +++ b/sdks/python/apache_beam/utils/timestamp.py @@ -166,7 +166,10 @@ def from_utc_datetime(cls, dt: datetime.datetime) -> 'Timestamp': if dt.tzinfo != pytz.utc and dt.tzinfo != datetime.timezone.utc: raise ValueError('dt not in UTC: %s' % dt) duration = dt - cls._epoch_datetime_utc() - return Timestamp(duration.total_seconds()) + # Avoid total_seconds(): its float result can be off by a microsecond. + return Timestamp( + seconds=duration.days * 86400 + duration.seconds, + micros=duration.microseconds) @classmethod def from_rfc3339(cls, rfc3339: str) -> 'Timestamp': diff --git a/sdks/python/apache_beam/utils/timestamp_test.py b/sdks/python/apache_beam/utils/timestamp_test.py index e1d120da471f..ec1c36046523 100644 --- a/sdks/python/apache_beam/utils/timestamp_test.py +++ b/sdks/python/apache_beam/utils/timestamp_test.py @@ -369,6 +369,23 @@ def test_duration_arithmetic_guard_rail(self): with self.assertRaises(ValueError): _ = ts % Duration(seconds=1) + def test_from_rfc3339_fraction_is_exact(self): + # Expected values are integers taken from the string, never + # Timestamp(float): both sides would share the same lossy float path. + # Seconds just above a power of two maximize float error. + for rfc, want_sec, want_sub, want_p in [ + ('2038-01-19T03:14:08.510215590Z', 2147483648, 510215590, 9), + ('2004-01-10T13:37:04.611178002Z', 1073741824, 611178002, 9), + ('1970-01-01T00:00:00.1252641Z', 0, 1252641, 7), + ('1970-01-01T00:00:00.254229935Z', 0, 254229935, 9), + ('1969-12-31T23:59:59.746939251Z', -1, 746939251, 9), + ('9999-12-31T23:59:59.389694109Z', 253402300799, 389694109, 9), + ]: + ts = Timestamp.from_rfc3339(rfc) + self.assertEqual((ts.seconds(), ts.subseconds(), ts.precision()), + (want_sec, want_sub, want_p), + rfc) + class DurationTest(unittest.TestCase): def test_of(self): From 524036fb5236fed0f5065c6dc3f45e72402506a8 Mon Sep 17 00:00:00 2001 From: Ashwin S Date: Tue, 11 Aug 2026 14:15:05 -0400 Subject: [PATCH 4/4] bump FnAPI container to beam-master-20260811 (#39721) Co-authored-by: Ashwin Sampathkumar --- sdks/python/apache_beam/runners/dataflow/internal/names.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/sdks/python/apache_beam/runners/dataflow/internal/names.py b/sdks/python/apache_beam/runners/dataflow/internal/names.py index 656eeb7b04b9..bf764a680827 100644 --- a/sdks/python/apache_beam/runners/dataflow/internal/names.py +++ b/sdks/python/apache_beam/runners/dataflow/internal/names.py @@ -35,6 +35,6 @@ # Update this tag whenever there is a change that # requires changes to SDK harness container or SDK harness launcher. -BEAM_DEV_SDK_CONTAINER_TAG = 'beam-master-20260803' +BEAM_DEV_SDK_CONTAINER_TAG = 'beam-master-20260811' DATAFLOW_CONTAINER_IMAGE_REPOSITORY = 'gcr.io/cloud-dataflow/v1beta3'