nbali commented on code in PR #40391:
URL: https://github.com/apache/beam/pull/40391#discussion_r4434413668
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/firestore/FirestoreV1.java:
##########
@@ -230,6 +235,29 @@ public final class FirestoreV1 {
private FirestoreV1() {}
+ /**
+ * 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}.
+ *
+ * <p>Returns {@code null} when {@code updateCompatibilityVersion} requests
a pipeline shape from
+ * before this side input was introduced.
+ */
+ private static @Nullable PCollectionView<Instant> createRampUpStartView(
+ PCollection<?> input, JodaClock clock) {
+ if (StreamingOptions.updateCompatibilityVersionLessThan(
Review Comment:
Thanks, done in 53a7ca9. Each batch write transform keeps the previous
expansion unchanged in `expand_2_78_0`, and `expand` delegates to it in its
first lines when `updateCompatibilityVersion` is 2.78.0 or earlier, as
`Reshuffle` does. The new path no longer has a nullable side input.
I named it `_2_78_0` rather than `_2_77` because 2.78.0 was cut without this
change, so the new shape starts with 2.79.0. I also merged master, which now
includes #40390.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
