kennknowles commented on code in PR #40391:
URL: https://github.com/apache/beam/pull/40391#discussion_r4350203937
##########
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:
The way we should use this flag is that we should clone the prior `expand`
method to `expand_2_77(...)`.
- The `expand_2_77(...)` method will always have an unchanged topology.
- The new `expand` delegates to `expand_2_77` in its first line(s) and
after that has the new, totally separate, topology.
- This eliminate the "action at a distance" of this return value being null
causing the graph topology to be altered latter.
Check other uses of this pipeline option for examples. Particularly a simple
case I can think of is `Reshuffle`.
They can share sub-transforms or helper functions, but the forking based on
update compatibility should happen at the highest level. This is a case where
redundancy is a benefit, and pushing the cyclomatic complexity to the earliest
possible point is the way to do it.
--
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]
