[pull] master from apache:master

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(

sdks/java/io/delta/src/main/java/org/apache/beam/sdk/io/delta/DeltaIO.java

@@ -168,7 +168,9 @@ public PCollection<Row> 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();
     }

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<Long> 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<Row> output =
+        PCollectionRowTuple.empty(readPipeline)
+            .apply(new DeltaCdcReadSchemaTransformProvider().from(config))
+            .get(DeltaCdcReadSchemaTransformProvider.OUTPUT_TAG);
+
+    PCollection<String> 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");

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;

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();

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 = {}

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'

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)

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,

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

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

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':

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):