[pull] master from apache:master

.github/trigger_files/beam_PostCommit_Python_Xlang_Gcp_Direct.json

@@ -1,4 +1,4 @@
 {
   "comment": "Modify this file in a trivial way to cause this test suite to run",
-  "modification": 17
+  "modification": 1
 }

.github/trigger_files/beam_PostCommit_Python_Xlang_IO_Direct.json

@@ -1,4 +1,4 @@
 {
   "comment": "Modify this file in a trivial way to cause this test suite to run",
-  "modification": 7
+  "modification": 1
 }

.github/workflows/beam_Infrastructure_AuditUnmanagedKeys.yml

@@ -64,6 +64,11 @@ jobs:
       env:
         GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
         GITHUB_REPOSITORY: ${{ github.repository }}
+        SMTP_SERVER: smtp.gmail.com
+        SMTP_PORT: 465
+        EMAIL_ADDRESS: ${{ secrets.ISSUE_REPORT_SENDER_EMAIL_ADDRESS }}
+        EMAIL_PASSWORD: ${{ secrets.ISSUE_REPORT_SENDER_EMAIL_PASSWORD }}
+        EMAIL_RECIPIENT: "dev@beam.apache.org"
       run: python account_keys.py --action announce
 
 

.github/workflows/build_release_candidate.yml

@@ -316,7 +316,7 @@ jobs:
         # settings.xml file
         run: rm ~/.m2/settings.xml || true
       - name: Login to Docker Hub
-        uses: docker/login-action@371161bbe7024a29a25c5e19bfcbc0804fe9ad2c
+        uses: docker/login-action@dbcb813823bdd20940b903addbd779551569679f
         with:
           username: ${{ secrets.DOCKERHUB_USER }}
           password: ${{ secrets.DOCKERHUB_TOKEN }}

.github/workflows/finalize_release.yml

@@ -44,7 +44,7 @@ jobs:
     runs-on: [self-hosted, ubuntu-24.04, main]
     steps:
     - name: Login to Docker Hub
-      uses: docker/login-action@371161bbe7024a29a25c5e19bfcbc0804fe9ad2c
+      uses: docker/login-action@dbcb813823bdd20940b903addbd779551569679f
       with:
         username: ${{ secrets.DOCKERHUB_USER }}
         password: ${{ secrets.DOCKERHUB_TOKEN }}

CHANGES.md

@@ -73,6 +73,9 @@
 * (Java) Supported acknowledge mode for JmsIO ([#39253](https://github.com/apache/beam/issues/39253)).
 * (Python) Staged files directory is now automatically added to `sys.path` on the Python SDK worker at startup. This makes Python files provided via the '--files_to_stage' pipeline option importable in the pipeline code and makes it easier to initialize Python SDK harness at startup via the `--beam_plugins` pipeline option. For more information, see the [Staging Individual Files](https://beam.apache.org/documentation/sdks/python-pipeline-dependencies/#staging-files) section of the dependency management docs. This behavior can be disabled by passing the '--experiments=no_staged_dir_in_sys_path' pipeline option ([#39431](https://github.com/apache/beam/issues/39431)).
 * (Python) Added `equal_to_approx`, an `assert_that` matcher that compares numeric pipeline outputs with a configurable tolerance ([#18028](https://github.com/apache/beam/issues/18028)).
+* (Python) `Timestamp` now supports variable subsecond precision, up to nanoseconds. The portable
+  `beam:logical_type:timestamp:v1` logical type now maps to Python's `Timestamp` ([#39344](https://github.com/apache/beam/issues/39344)).
+* X feature added (Java/Python) ([#X](https://github.com/apache/beam/issues/X)).
 
 ## Breaking Changes
 

model/fn-execution/src/main/resources/org/apache/beam/model/fnexecution/v1/standard_coders.yaml

@@ -626,3 +626,34 @@ examples:
   "\u0001\u0003\u0061\u0062\u0063" : "abc"
   "\u0001\u000a\u006d\u006f\u0072\u0065\u0020\u0062\u0079\u0074\u0065\u0073" : "more bytes"
   "\u0000" : null
+
+---
+
+coder:
+  urn: "beam:coder:row:v1"
+  # f_timestamp: logical(timestamp:v1, precision=3, representation row<seconds:int64, subseconds:int16>)
+  payload: "\x0a\x88\x01\x0a\x0bf_timestamp\x1ay:w\x0a\x1ebeam:logical_type:timestamp:v1\x1aK2I\x0aG\x0a\x0d\x0a\x07seconds\x1a\x02\x10\x04\x0a\x10\x0a\x0asubseconds\x1a\x02\x10\x02\x12$5f2fbf6c-8b8e-4a30-a9be-e0f65f0f79c3\"\x02\x10\x03*\x04\x0a\x02\x18\x03\x12$9c7c1b6c-2a5e-4c53-9b32-91b71c1f8ac3"
+examples:
+  "\x01\x00\x02\x00\xb6\x95\xd5\xf9\x05\x03\xe7": {f_timestamp: {seconds: 1597328054, subseconds: 999}}
+  "\x01\x00\x02\x00\xfe\xff\xff\xff\xff\xff\xff\xff\xff\x01\x01\xf4": {f_timestamp: {seconds: -2, subseconds: 500}}
+  "\x01\x00\x02\x00\xb6\x95\xd5\xf9\x05\x00\x00": {f_timestamp: {seconds: 1597328054, subseconds: 0}}
+
+---
+
+coder:
+  urn: "beam:coder:row:v1"
+  # f_timestamp: logical(timestamp:v1, precision=6, representation row<seconds:int64, subseconds:int32>)
+  payload: "\x0a\x88\x01\x0a\x0bf_timestamp\x1ay:w\x0a\x1ebeam:logical_type:timestamp:v1\x1aK2I\x0aG\x0a\x0d\x0a\x07seconds\x1a\x02\x10\x04\x0a\x10\x0a\x0asubseconds\x1a\x02\x10\x03\x12$b8b3faf8-63f8-4b40-9f4d-3c1e83d4e07b\"\x02\x10\x03*\x04\x0a\x02\x18\x06\x12$e0a72e10-b214-4fd2-a4dc-fca0e7f7d924"
+examples:
+  "\x01\x00\x02\x00\xb6\x95\xd5\xf9\x05\xc0\xc4\x07": {f_timestamp: {seconds: 1597328054, subseconds: 123456}}
+  "\x01\x00\x02\x00\xfe\xff\xff\xff\xff\xff\xff\xff\xff\x01\xa0\xc2\x1e": {f_timestamp: {seconds: -2, subseconds: 500000}}
+
+---
+
+coder:
+  urn: "beam:coder:row:v1"
+  # f_timestamp: logical(timestamp:v1, precision=9, representation row<seconds:int64, subseconds:int32>)
+  payload: "\x0a\x88\x01\x0a\x0bf_timestamp\x1ay:w\x0a\x1ebeam:logical_type:timestamp:v1\x1aK2I\x0aG\x0a\x0d\x0a\x07seconds\x1a\x02\x10\x04\x0a\x10\x0a\x0asubseconds\x1a\x02\x10\x03\x12$3d6f7b81-2f13-4a41-8be3-4d8cd8f30f42\"\x02\x10\x03*\x04\x0a\x02\x18\x09\x12$cf03a3c1-4a3a-4f77-8e2b-6e5b19c19f27"
+examples:
+  "\x01\x00\x02\x00\xb6\x95\xd5\xf9\x05\x95\x9a\xef:": {f_timestamp: {seconds: 1597328054, subseconds: 123456789}}
+  "\x01\x00\x02\x00\xfe\xff\xff\xff\xff\xff\xff\xff\xff\x01\xff\x93\xeb\xdc\x03": {f_timestamp: {seconds: -2, subseconds: 999999999}}

model/pipeline/src/main/proto/org/apache/beam/model/pipeline/v1/schema.proto

@@ -234,6 +234,19 @@ message LogicalTypes {
     //     since the epoch.
     DATE = 8 [(org.apache.beam.model.pipeline.v1.beam_urn) =
       "beam:logical_type:date:v1"];
+
+    // A URN for Timestamp type
+    //   - Representation type: ROW<seconds: INT64, subseconds: INT16> for
+    //     precision < 5, otherwise ROW<seconds: INT64, subseconds: INT32>
+    //   - Argument type: INT32.
+    //     A timestamp without a timezone, represented as seconds since the
+    //     epoch plus a non-negative fraction of a second. The argument is
+    //     the precision: the number of decimal digits of the subseconds
+    //     field (e.g. 3 for milliseconds, 6 for micros, 9 for nanos).
+    //     subseconds stays non-negative even for timestamps before the epoch,
+    //     e.g. -1.5s at precision 6 is {seconds: -2, subseconds: 500000}.
+    TIMESTAMP = 9 [(org.apache.beam.model.pipeline.v1.beam_urn) =
+      "beam:logical_type:timestamp:v1"];
   }
 }
 

runners/core-java/src/main/java/org/apache/beam/runners/core/GroupAlsoByWindowViaWindowSetNewDoFn.java

@@ -109,7 +109,7 @@ public void processElement(ProcessContext c) throws Exception {
             reduceFn,
             c.getPipelineOptions());
 
-    reduceFnRunner.processElements(keyedWorkItem.elementsIterable());
+    reduceFnRunner.processElements(keyedWorkItem);
     reduceFnRunner.onTimers(keyedWorkItem.timersIterable());
     reduceFnRunner.persist();
   }

runners/core-java/src/main/java/org/apache/beam/runners/core/KeyedWorkItem.java

@@ -35,4 +35,13 @@ public interface KeyedWorkItem<K, ElemT> {
 
   /** Returns an iterable containing the elements. */
   Iterable<WindowedValue<ElemT>> elementsIterable();
+
+  /**
+   * Returns an iterable containing windowed values without guaranteeing element payload decoding.
+   * Useful for lightweight inspection of windowing metadata without payload deserialization
+   * overhead.
+   */
+  default Iterable<WindowedValue<?>> elementWindowsIterable() {
+    return (Iterable) elementsIterable();
+  }
 }

runners/core-java/src/main/java/org/apache/beam/runners/core/ReduceFnRunner.java

@@ -60,6 +60,7 @@
 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.collect.FluentIterable;
 import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableSet;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Iterables;
 import org.checkerframework.checker.nullness.qual.Nullable;
 import org.joda.time.Duration;
 import org.joda.time.Instant;
@@ -361,13 +362,24 @@ private Collection<W> windowsThatShouldFire(Set<W> windows) throws Exception {
    *       setting holds, and invoking {@link ReduceFn#onTrigger}.
    * </ol>
    */
+  public void processElements(KeyedWorkItem<?, InputT> keyedWorkItem) throws Exception {
+    processElementsInternal(
+        keyedWorkItem.elementWindowsIterable(), keyedWorkItem.elementsIterable());
+  }
+
   public void processElements(Iterable<WindowedValue<InputT>> values) throws Exception {
-    if (!values.iterator().hasNext()) {
+    processElementsInternal(values, values);
+  }
+
+  private void processElementsInternal(
+      Iterable<? extends WindowedValue<?>> elementWindows, Iterable<WindowedValue<InputT>> values)
+      throws Exception {
+    if (Iterables.isEmpty(elementWindows)) {
       return;
     }
 
     // Determine all the windows for elements.
-    Set<W> windows = collectWindows(values);
+    Set<W> windows = collectWindows(elementWindows);
     // If an incoming element introduces a new window, attempt to merge it into an existing
     // window eagerly.
     Map<W, W> windowToMergeResult = mergeWindows(windows);
@@ -426,7 +438,7 @@ public void persist() {
   }
 
   /** Extract the windows associated with the values. */
-  private Set<W> collectWindows(Iterable<WindowedValue<InputT>> values) throws Exception {
+  private Set<W> collectWindows(Iterable<? extends WindowedValue<?>> values) throws Exception {
     Set<W> windows = new HashSet<>();
     for (WindowedValue<?> value : values) {
       for (BoundedWindow untypedWindow : value.getWindows()) {

runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/DataflowOutputCounter.java

@@ -18,12 +18,14 @@
 package org.apache.beam.runners.dataflow.worker;
 
 import org.apache.beam.runners.core.ElementByteSizeObservable;
+import org.apache.beam.runners.core.KeyedWorkItem;
 import org.apache.beam.runners.dataflow.worker.counters.Counter;
 import org.apache.beam.runners.dataflow.worker.counters.CounterFactory;
 import org.apache.beam.runners.dataflow.worker.counters.CounterName;
 import org.apache.beam.runners.dataflow.worker.counters.NameContext;
 import org.apache.beam.runners.dataflow.worker.util.common.worker.ElementCounter;
 import org.apache.beam.runners.dataflow.worker.util.common.worker.OutputObjectAndByteCounter;
+import org.apache.beam.sdk.annotations.Internal;
 import org.apache.beam.sdk.values.WindowedValue;
 import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting;
 
@@ -33,6 +35,7 @@
 @SuppressWarnings({
   "nullness" // TODO(https://github.com/apache/beam/issues/20497)
 })
+@Internal
 public class DataflowOutputCounter implements ElementCounter {
   /** Number of logical element and single window pairs that were processed. */
   private static final String ELEMENT_COUNTER_NAME = "-ElementCount";
@@ -41,20 +44,36 @@ public class DataflowOutputCounter implements ElementCounter {
 
   private OutputObjectAndByteCounter objectAndByteCounter;
   private Counter<Long, ?> elementCount;
+  private final boolean isStreaming;
 
-  public DataflowOutputCounter(
-      String outputName, CounterFactory counterFactory, NameContext nameContext) {
-    this(outputName, null, counterFactory, nameContext);
+  public static DataflowOutputCounter create(
+      String outputName,
+      ElementByteSizeObservable<?> elementByteSizeObservable,
+      CounterFactory counterFactory,
+      NameContext nameContext,
+      boolean isStreaming) {
+    return new DataflowOutputCounter(
+        outputName, elementByteSizeObservable, counterFactory, nameContext, isStreaming);
+  }
+
+  public static DataflowOutputCounter create(
+      String outputName,
+      CounterFactory counterFactory,
+      NameContext nameContext,
+      boolean isStreaming) {
+    return new DataflowOutputCounter(outputName, null, counterFactory, nameContext, isStreaming);
   }
 
-  public DataflowOutputCounter(
+  private DataflowOutputCounter(
       String outputName,
       ElementByteSizeObservable<?> elementByteSizeObservable,
       CounterFactory counterFactory,
-      NameContext nameContext) {
-    objectAndByteCounter =
+      NameContext nameContext,
+      boolean isStreaming) {
+    this.isStreaming = isStreaming;
+    this.objectAndByteCounter =
         new OutputObjectAndByteCounter(elementByteSizeObservable, counterFactory, nameContext);
-    objectAndByteCounter.countMeanByte(outputName + MEAN_BYTE_COUNTER_NAME);
+    this.objectAndByteCounter.countMeanByte(outputName + MEAN_BYTE_COUNTER_NAME);
     createElementCounter(counterFactory, outputName + ELEMENT_COUNTER_NAME);
   }
 
@@ -63,15 +82,42 @@ public void update(Object elem) throws Exception {
     objectAndByteCounter.update(elem);
     long windowsSize = ((WindowedValue<?>) elem).getWindows().size();
     if (windowsSize == 0) {
-      // GroupingShuffleReader produces ValueInEmptyWindows.
-      // For now, we count the element at least once to keep the current counter
-      // behavior.
-      elementCount.addValue(1L);
+      updateEmptyWindows((WindowedValue<?>) elem);
     } else {
+      // Standard WindowedValue.
       elementCount.addValue(windowsSize);
     }
   }
 
+  private void updateEmptyWindows(WindowedValue<?> elem) {
+    if (isStreaming) {
+      Object value = elem.getValue();
+      if (value instanceof KeyedWorkItem<?, ?>) {
+        // KeyedWorkItem wrapped in ValueInEmptyWindows
+        // (e.g. WindowingWindmillReader for Streaming GBK)
+        KeyedWorkItem<?, ?> keyedWorkItem = (KeyedWorkItem<?, ?>) value;
+        long totalElementCount = 0;
+        // Iterate through elementWindowsIterable and ignore timers in KeyedWorkItem.
+        // Uses lightweight metadata-only iteration without payload deserialization overhead.
+        for (WindowedValue<?> element : keyedWorkItem.elementWindowsIterable()) {
+          long elementWindowsSize = element.getWindows().size();
+          // Fan out for windows.
+          totalElementCount += (elementWindowsSize == 0 ? 1L : elementWindowsSize);
+        }
+        elementCount.addValue(totalElementCount);
+      } else {
+        // NOTE: in streaming mode, this should not normally happen.
+        // Counting as 1 element serves as a fallback to maintain counter behavior without failing
+        // execution.
+        elementCount.addValue(1L);
+      }
+    } else {
+      // Non-KeyedWorkItem wrapped in ValueInEmptyWindows
+      // (e.g. GroupingShuffleReader KV output for Batch GBK)
+      elementCount.addValue(1L);
+    }
+  }
+
   @Override
   public void finishLazyUpdate(Object elem) {
     objectAndByteCounter.finishLazyUpdate(elem);

runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/IntrinsicMapTaskExecutorFactory.java

@@ -63,6 +63,7 @@
 import org.apache.beam.sdk.coders.KvCoder;
 import org.apache.beam.sdk.fn.IdGenerator;
 import org.apache.beam.sdk.options.PipelineOptions;
+import org.apache.beam.sdk.options.StreamingOptions;
 import org.apache.beam.sdk.util.common.ElementByteSizeObserver;
 import org.apache.beam.sdk.values.TupleTag;
 import org.apache.beam.sdk.values.WindowedValues.WindowedValueCoder;
@@ -102,8 +103,9 @@ public DataflowMapTaskExecutor create(
       IdGenerator idGenerator) {
 
     // Swap out all the InstructionOutput nodes with OutputReceiver nodes
+    boolean isStreaming = options.as(StreamingOptions.class).isStreaming();
     Networks.replaceDirectedNetworkNodes(
-        network, createOutputReceiversTransform(stageName, counterSet));
+        network, createOutputReceiversTransform(stageName, counterSet, isStreaming));
 
     // Swap out all the ParallelInstruction nodes with Operation nodes. While updating the network,
     // we keep track of
@@ -345,7 +347,7 @@ OperationNode createFlattenOperation(
    * Returns a function which can convert {@link InstructionOutput}s into {@link OutputReceiver}s.
    */
   static Function<Node, Node> createOutputReceiversTransform(
-      final String stageName, final CounterFactory counterFactory) {
+      final String stageName, final CounterFactory counterFactory, final boolean isStreaming) {
     return new TypeSafeNodeFunction<InstructionOutputNode>(InstructionOutputNode.class) {
       @Override
       public Node typedApply(InstructionOutputNode input) {
@@ -355,15 +357,16 @@ public Node typedApply(InstructionOutputNode input) {
             CloudObjects.coderFromCloudObject(CloudObject.fromSpec(cloudOutput.getCodec()));
 
         ElementCounter outputCounter =
-            new DataflowOutputCounter(
+            DataflowOutputCounter.create(
                 cloudOutput.getName(),
                 new ElementByteSizeObservableCoder<>(coder),
                 counterFactory,
                 NameContext.create(
                     stageName,
                     cloudOutput.getOriginalName(),
                     cloudOutput.getSystemName(),
-                    cloudOutput.getName()));
+                    cloudOutput.getName()),
+                isStreaming);
         outputReceiver.addOutputCounter(outputCounter);
 
         return OutputReceiverNode.create(outputReceiver, coder, input.getPcollectionId());

runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/SimpleParDoFnHelpers.java

@@ -190,9 +190,10 @@ public <TagT> void output(TupleTag<TagT> tag, WindowedValue<TagT> output) {
               // doesn't today.)
               OutputReceiver undeclaredReceiver = new OutputReceiver();
 
+              boolean isStreaming = options.as(StreamingOptions.class).isStreaming();
               ElementCounter outputCounter =
-                  new DataflowOutputCounter(
-                      outputName, counterFactory, stepContext.getNameContext());
+                  DataflowOutputCounter.create(
+                      outputName, counterFactory, stepContext.getNameContext(), isStreaming);
               undeclaredReceiver.addOutputCounter(outputCounter);
               undeclaredOutputs.put(tag, undeclaredReceiver);
               receiver = undeclaredReceiver;

runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingGroupAlsoByWindowViaWindowSetFn.java

@@ -93,7 +93,7 @@ public void processElement(
             reduceFn,
             options);
 
-    reduceFnRunner.processElements(keyedWorkItem.elementsIterable());
+    reduceFnRunner.processElements(keyedWorkItem);
     reduceFnRunner.onTimers(keyedWorkItem.timersIterable());
     reduceFnRunner.persist();
   }