You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by re...@apache.org on 2025/02/12 21:35:03 UTC
(beam) branch master updated: Merge pull request #33973: Fix timestamp propagation to successful-loads PCollection.
This is an automated email from the ASF dual-hosted git repository.
reuvenlax pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new 27f9fe9d889 Merge pull request #33973: Fix timestamp propagation to successful-loads PCollection.
27f9fe9d889 is described below
commit 27f9fe9d8897a1ed5dcf57bb065b6ac511338248
Author: Reuven Lax <re...@google.com>
AuthorDate: Wed Feb 12 13:34:58 2025 -0800
Merge pull request #33973: Fix timestamp propagation to successful-loads PCollection.
---
.../main/java/org/apache/beam/sdk/io/gcp/bigquery/BatchLoads.java | 7 +++++--
1 file changed, 5 insertions(+), 2 deletions(-)
diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BatchLoads.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BatchLoads.java
index 8f97ef1f8c5..c35b99fbfc3 100644
--- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BatchLoads.java
+++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BatchLoads.java
@@ -67,6 +67,7 @@ import org.apache.beam.sdk.transforms.windowing.AfterProcessingTime;
import org.apache.beam.sdk.transforms.windowing.DefaultTrigger;
import org.apache.beam.sdk.transforms.windowing.GlobalWindows;
import org.apache.beam.sdk.transforms.windowing.Repeatedly;
+import org.apache.beam.sdk.transforms.windowing.TimestampCombiner;
import org.apache.beam.sdk.transforms.windowing.Window;
import org.apache.beam.sdk.util.Preconditions;
import org.apache.beam.sdk.values.KV;
@@ -340,7 +341,8 @@ class BatchLoads<DestinationT, ElementT>
AfterProcessingTime.pastFirstElementInPane()
.plusDelayOf(triggeringFrequency),
AfterPane.elementCountAtLeast(FILE_TRIGGERING_RECORD_COUNT))))
- .discardingFiredPanes());
+ .discardingFiredPanes()
+ .withTimestampCombiner(TimestampCombiner.EARLIEST));
results = writeStaticallyShardedFiles(inputInGlobalWindow, tempFilePrefixView);
} else {
// In the case of dynamic sharding, however, we use a default trigger since the transform
@@ -364,7 +366,8 @@ class BatchLoads<DestinationT, ElementT>
Repeatedly.forever(
AfterProcessingTime.pastFirstElementInPane()
.plusDelayOf(triggeringFrequency)))
- .discardingFiredPanes());
+ .discardingFiredPanes()
+ .withTimestampCombiner(TimestampCombiner.EARLIEST));
TupleTag<KV<ShardedKey<DestinationT>, WritePartition.Result>> multiPartitionsTag =
new TupleTag<>("multiPartitionsTag");