nbali · Oct 2, 2026 · Oct 14, 2026 · Nov 1, 2026 · Nov 1, 2026
diff --git a/CHANGES.md b/CHANGES.md
@@ -70,6 +70,7 @@
 
 * X feature added (Java/Python) ([#X](https://github.com/apache/beam/issues/X)).
 * (Python) Added support for Protobuf 7.x ([#40580](https://github.com/apache/beam/issues/40580)).
+* FirestoreIO batch writes now ramp up from a single pipeline-wide start time instead of each worker's first write, as DatastoreIO does. This adds a side input to the write transforms; set `--updateCompatibilityVersion` to 2.78.0 or earlier to keep the previous pipeline shape (Java) ([#27170](https://github.com/apache/beam/issues/27170)).
 
 ## Breaking Changes
 

diff --git a/...google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/firestore/FirestoreV1.java b/...google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/firestore/FirestoreV1.java
@@ -56,18 +56,23 @@
 import org.apache.beam.sdk.io.gcp.firestore.FirestoreV1ReadFn.RunQueryFn;
 import org.apache.beam.sdk.io.gcp.firestore.FirestoreV1WriteFn.BatchWriteFnWithDeadLetterQueue;
 import org.apache.beam.sdk.io.gcp.firestore.FirestoreV1WriteFn.BatchWriteFnWithSummary;
+import org.apache.beam.sdk.options.StreamingOptions;
+import org.apache.beam.sdk.transforms.Create;
 import org.apache.beam.sdk.transforms.DoFn;
 import org.apache.beam.sdk.transforms.MapElements;
 import org.apache.beam.sdk.transforms.PTransform;
 import org.apache.beam.sdk.transforms.ParDo;
 import org.apache.beam.sdk.transforms.Reshuffle;
 import org.apache.beam.sdk.transforms.SimpleFunction;
+import org.apache.beam.sdk.transforms.View;
 import org.apache.beam.sdk.transforms.display.DisplayData;
 import org.apache.beam.sdk.transforms.display.HasDisplayData;
 import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.PCollectionView;
 import org.apache.beam.sdk.values.PDone;
 import org.apache.beam.sdk.values.PInput;
 import org.apache.beam.sdk.values.POutput;
+import org.apache.beam.sdk.values.TypeDescriptor;
 import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting;
 import org.checkerframework.checker.nullness.qual.Nullable;
 import org.joda.time.Instant;
@@ -230,6 +235,31 @@ public final class FirestoreV1 {
 
   private FirestoreV1() {}
 
+  /**
+   * Whether {@code updateCompatibilityVersion} requests the batch write pipeline shape from before
+   * the pipeline-wide ramp-up start side input was introduced.
+   */
+  private static boolean useBatchWriteExpand_2_78_0(PCollection<?> input) {
+    return StreamingOptions.updateCompatibilityVersionLessThan(
+        input.getPipeline().getOptions(), "2.79.0");
+  }
+
+  /**
+   * Creates a side input holding a single pipeline-wide instant, from which the write ramp-up
+   * budget of every worker grows, rather than each worker starting its own ramp-up from its first
+   * write. Follows the approach of {@code DatastoreV1.Mutate}.
+   */
+  private static PCollectionView<Instant> createRampUpStartView(
+      PCollection<?> input, JodaClock clock) {
+    return input
+        .getPipeline()
+        .apply("Create ramp-up start", Create.of("rampUpStart"))
+        .apply(
+            "Generate ramp-up start timestamp",
+            MapElements.into(TypeDescriptor.of(Instant.class)).via(ignored -> clock.instant()))
+        .apply("Ramp-up start view", View.asSingleton());
+  }
+
   /**
    * The class returned by this method provides the ability to create {@link PTransform PTransforms}
    * for read operations available in the Firestore V1 API provided by {@link
@@ -1485,6 +1515,26 @@ public BatchWriteWithSummary(
     @Override
     public PCollection<WriteSuccessSummary> expand(
         PCollection<com.google.firestore.v1.Write> input) {
+      if (useBatchWriteExpand_2_78_0(input)) {
+        return expand_2_78_0(input);
+      }
+      PCollectionView<Instant> rampUpStartView = createRampUpStartView(input, clock);
+      return input.apply(
+          "batchWrite",
+          ParDo.of(
+                  new BatchWriteFnWithSummary(
+                      clock,
+                      firestoreStatefulComponentFactory,
+                      rpcQosOptions,
+                      CounterFactory.DEFAULT,
+                      projectId,
+                      databaseId,
+                      rampUpStartView))
+              .withSideInputs(rampUpStartView));
+    }
+
+    private PCollection<WriteSuccessSummary> expand_2_78_0(
+        PCollection<com.google.firestore.v1.Write> input) {
       return input.apply(
           "batchWrite",
           ParDo.of(
@@ -1494,7 +1544,8 @@ public PCollection<WriteSuccessSummary> expand(
                   rpcQosOptions,
                   CounterFactory.DEFAULT,
                   projectId,
-                  databaseId)));
+                  databaseId,
+                  null)));
     }
 
     @Override
@@ -1618,6 +1669,26 @@ private BatchWriteWithDeadLetterQueue(
 
     @Override
     public PCollection<WriteFailure> expand(PCollection<com.google.firestore.v1.Write> input) {
+      if (useBatchWriteExpand_2_78_0(input)) {
+        return expand_2_78_0(input);
+      }
+      PCollectionView<Instant> rampUpStartView = createRampUpStartView(input, clock);
+      return input.apply(
+          "batchWrite",
+          ParDo.of(
+                  new BatchWriteFnWithDeadLetterQueue(
+                      clock,
+                      firestoreStatefulComponentFactory,
+                      rpcQosOptions,
+                      CounterFactory.DEFAULT,
+                      projectId,
+                      databaseId,
+                      rampUpStartView))
+              .withSideInputs(rampUpStartView));
+    }
+
+    private PCollection<WriteFailure> expand_2_78_0(
+        PCollection<com.google.firestore.v1.Write> input) {
       return input.apply(
           "batchWrite",
           ParDo.of(
@@ -1627,7 +1698,8 @@ public PCollection<WriteFailure> expand(PCollection<com.google.firestore.v1.Writ
                   rpcQosOptions,
                   CounterFactory.DEFAULT,
                   projectId,
-                  databaseId)));
+                  databaseId,
+                  null)));
     }
 
     @Override

diff --git a/...cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/firestore/FirestoreV1WriteFn.java b/...cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/firestore/FirestoreV1WriteFn.java
@@ -53,6 +53,7 @@
 import org.apache.beam.sdk.util.BackOffUtils;
 import org.apache.beam.sdk.util.Preconditions;
 import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.PCollectionView;
 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.ImmutableList;
 import org.checkerframework.checker.nullness.qual.NonNull;
@@ -75,14 +76,16 @@ static final class BatchWriteFnWithSummary extends BaseBatchWriteFn<WriteSuccess
         RpcQosOptions rpcQosOptions,
         CounterFactory counterFactory,
         @Nullable String projectId,
-        @Nullable String databaseId) {
+        @Nullable String databaseId,
+        @Nullable PCollectionView<Instant> rampUpStartView) {
       super(
           clock,
           firestoreStatefulComponentFactory,
           rpcQosOptions,
           counterFactory,
           projectId,
-          databaseId);
+          databaseId,
+          rampUpStartView);
     }
 
     @Override
@@ -112,14 +115,16 @@ static final class BatchWriteFnWithDeadLetterQueue extends BaseBatchWriteFn<Writ
         RpcQosOptions rpcQosOptions,
         CounterFactory counterFactory,
         @Nullable String projectId,
-        @Nullable String databaseId) {
+        @Nullable String databaseId,
+        @Nullable PCollectionView<Instant> rampUpStartView) {
       super(
           clock,
           firestoreStatefulComponentFactory,
           rpcQosOptions,
           counterFactory,
           projectId,
-          databaseId);
+          databaseId,
+          rampUpStartView);
     }
 
     @Override
@@ -175,6 +180,7 @@ abstract static class BaseBatchWriteFn<OutT> extends ExplicitlyWindowedFirestore
     private transient DatabaseRootName databaseRootName;
     private final @Nullable String configuredProjectId;
     private final @Nullable String configuredDatabaseId;
+    private final @Nullable PCollectionView<Instant> rampUpStartView;
 
     @VisibleForTesting
     transient Queue<@NonNull WriteElement> writes = new PriorityQueue<>(WriteElement.COMPARATOR);
@@ -190,14 +196,16 @@ abstract static class BaseBatchWriteFn<OutT> extends ExplicitlyWindowedFirestore
         RpcQosOptions rpcQosOptions,
         CounterFactory counterFactory,
         @Nullable String configuredProjectId,
-        @Nullable String configuredDatabaseId) {
+        @Nullable String configuredDatabaseId,
+        @Nullable PCollectionView<Instant> rampUpStartView) {
       this.clock = clock;
       this.firestoreStatefulComponentFactory = firestoreStatefulComponentFactory;
       this.rpcQosOptions = rpcQosOptions;
       this.counterFactory = counterFactory;
       this.rpcAttemptContext = V1FnRpcAttemptContext.BatchWrite;
       this.configuredProjectId = configuredProjectId;
       this.configuredDatabaseId = configuredDatabaseId;
+      this.rampUpStartView = rampUpStartView;
     }
 
     @Override
@@ -265,6 +273,9 @@ public void processElement(ProcessContext context, BoundedWindow window) throws
           "nullness") // error checker is configured to treat any method not explicitly annotated as
       // @Nullable as non-null, this includes Objects.requireNonNull
       Write write = requireNonNull(context.element(), "context.element() must be non null");
+      if (rampUpStartView != null) {
+        rpcQos.setRampUpStart(context.sideInput(rampUpStartView));
+      }
       ProcessContextAdapter<OutT> contextAdapter = new ProcessContextAdapter<>(context);
       int serializedSize = write.getSerializedSize();
       boolean tooLarge = rpcQos.bytesOverLimit(serializedSize);

diff --git a/...a/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/firestore/RpcQos.java b/...a/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/firestore/RpcQos.java
@@ -98,6 +98,15 @@ interface RpcQos {
    */
   RpcWriteAttempt newWriteAttempt(Context context);
 
+  /**
+   * Set the instant from which the write ramp-up budget grows.
+   *
+   * <p>If never called, the ramp-up starts from the first write attempted by this instance.
+   *
+   * @param rampUpStart The instant the ramp-up is considered to have started.
+   */
+  void setRampUpStart(Instant rampUpStart);
+
   /**
    * Check if a request is over the max allowed number of bytes.
    *

diff --git a/.../google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/firestore/RpcQosImpl.java b/.../google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/firestore/RpcQosImpl.java
@@ -125,6 +125,11 @@ final class RpcQosImpl implements RpcQos {
     computeCounters = (Context c) -> O11y.create(c, counterFactory, filteringDistributionFactory);
   }
 
+  @Override
+  public void setRampUpStart(Instant rampUpStart) {
+    writeRampUp.setFirstInstant(rampUpStart);
+  }
+
   @Override
   public RpcWriteAttemptImpl newWriteAttempt(Context context) {
     return new RpcWriteAttemptImpl(
@@ -633,6 +638,10 @@ static final class WriteRampUp {
           distributionFactory.get(RpcQos.class.getName(), "qos_rampUp_availableWriteCountBudget");
     }
 
+    void setFirstInstant(Instant instant) {
+      firstInstant = Optional.of(instant);
+    }
+
     int getAvailableWriteCountBudget(Instant instant) {
       if (!firstInstant.isPresent()) {
         firstInstant = Optional.of(instant);

diff --git a/.../test/java/org/apache/beam/sdk/io/gcp/firestore/FirestoreV1BatchWriteRampUpStartTest.java b/.../test/java/org/apache/beam/sdk/io/gcp/firestore/FirestoreV1BatchWriteRampUpStartTest.java
@@ -0,0 +1,99 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.io.gcp.firestore;
+
+import static org.junit.Assert.assertEquals;
+
+import com.google.firestore.v1.Write;
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.beam.sdk.Pipeline;
+import org.apache.beam.sdk.extensions.protobuf.ProtoCoder;
+import org.apache.beam.sdk.options.PipelineOptions;
+import org.apache.beam.sdk.options.PipelineOptionsFactory;
+import org.apache.beam.sdk.options.StreamingOptions;
+import org.apache.beam.sdk.runners.TransformHierarchy;
+import org.apache.beam.sdk.transforms.Create;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.PCollectionView;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+@RunWith(JUnit4.class)
+public final class FirestoreV1BatchWriteRampUpStartTest {
+
+  @Test
+  public void batchWrite_usesPipelineWideRampUpStart() {
+    assertEquals(1, batchWriteSideInputs(PipelineOptionsFactory.create(), false).size());
+  }
+
+  @Test
+  public void batchWriteWithDeadLetterQueue_usesPipelineWideRampUpStart() {
+    assertEquals(1, batchWriteSideInputs(PipelineOptionsFactory.create(), true).size());
+  }
+
+  @Test
+  public void batchWrite_updateCompatibilityVersion_2_78_0_keepsPreviousPipelineShape() {
+    assertEquals(0, batchWriteSideInputs(withUpdateCompatibilityVersion("2.78.0"), false).size());
+  }
+
+  @Test
+  public void batchWriteWithDeadLetterQueue_updateCompatibilityVersion_2_78_0_keepsPreviousShape() {
+    assertEquals(0, batchWriteSideInputs(withUpdateCompatibilityVersion("2.78.0"), true).size());
+  }
+
+  @Test
+  public void batchWrite_updateCompatibilityVersion_2_79_0_usesPipelineWideRampUpStart() {
+    assertEquals(1, batchWriteSideInputs(withUpdateCompatibilityVersion("2.79.0"), false).size());
+  }
+
+  private static PipelineOptions withUpdateCompatibilityVersion(String version) {
+    StreamingOptions options = PipelineOptionsFactory.as(StreamingOptions.class);
+    options.setUpdateCompatibilityVersion(version);
+    return options;
+  }
+
+  private static List<PCollectionView<?>> batchWriteSideInputs(
+      PipelineOptions options, boolean withDeadLetterQueue) {
+    Pipeline pipeline = Pipeline.create(options);
+    PCollection<Write> writes = pipeline.apply(Create.empty(ProtoCoder.of(Write.class)));
+    if (withDeadLetterQueue) {
+      writes.apply(FirestoreIO.v1().write().batchWrite().withDeadLetterQueue().build());
+    } else {
+      writes.apply(FirestoreIO.v1().write().batchWrite().build());
+    }
+
+    List<PCollectionView<?>> sideInputs = new ArrayList<>();
+    pipeline.traverseTopologically(
+        new Pipeline.PipelineVisitor.Defaults() {
+          @Override
+          public CompositeBehavior enterCompositeTransform(TransformHierarchy.Node node) {
+            PTransform<?, ?> transform = node.getTransform();
+            if (transform instanceof ParDo.SingleOutput
+                && node.getFullName().endsWith("/batchWrite")) {
+              sideInputs.addAll(((ParDo.SingleOutput<?, ?>) transform).getSideInputs().values());
+            }
+            return CompositeBehavior.ENTER_TRANSFORM;
+          }
+        });
+    return sideInputs;
+  }
+}
diff --git a/.../org/apache/beam/sdk/io/gcp/firestore/FirestoreV1FnBatchWriteWithDeadLetterQueueTest.java b/.../org/apache/beam/sdk/io/gcp/firestore/FirestoreV1FnBatchWriteWithDeadLetterQueueTest.java
@@ -223,6 +223,6 @@ protected BatchWriteFnWithDeadLetterQueue getFn(
       CounterFactory counterFactory,
       DistributionFactory distributionFactory) {
     return new BatchWriteFnWithDeadLetterQueue(
-        clock, ff, rpcQosOptions, counterFactory, null, null);
+        clock, ff, rpcQosOptions, counterFactory, null, null, null);
   }
 }