[GitHub] [beam] boyuanzz commented on a change in pull request #12710: [BEAM-10303] Refactor split logic to reuse common logic.

[GitBox <gi...@apache.org>]

boyuanzz commented on a change in pull request #12710:
URL: https://github.com/apache/beam/pull/12710#discussion_r482267362



##########
File path: sdks/java/harness/src/main/java/org/apache/beam/fn/harness/FnApiDoFnRunner.java
##########
@@ -1546,50 +1459,32 @@ public Object restriction() {
               currentWindowIndex,
               stopWindowIndex,
               stopWindowIndex,
-              splitResult,
+              elementSplitResult,
               watermarkAndState);
     }
-    return KV.of(windowedSplitResult, newWindowStopIndex);
+    return SplitResultsWithStopIndex.of(
+        windowedSplitResult, downstreamSplitResult, newWindowStopIndex);
   }
 
-  private HandlesSplits.SplitResult trySplitForElementAndRestriction(
-      double fractionOfRemainder, Duration resumeDelay) {
-    KV<Instant, WatermarkEstimatorStateT> watermarkAndState;
-    WindowedSplitResult windowedSplitResult = null;
-    synchronized (splitLock) {
-      // There is nothing to split if we are between element and restriction processing calls.
-      if (currentTracker == null) {
-        return null;
-      }
-      // Make sure to get the output watermark before we split to ensure that the lower bound
-      // applies to the residual.
-      watermarkAndState = currentWatermarkEstimator.getWatermarkAndState();
-      KV<WindowedSplitResult, Integer> splitResult =
-          trySplitForProcess(
-              currentElement,
-              currentRestriction,
-              currentWindow,
-              currentWindows,
-              currentWatermarkEstimatorState,
-              fractionOfRemainder,
-              currentTracker,
-              watermarkAndState,
-              windowCurrentIndex,
-              windowStopIndex);
-      if (splitResult == null) {
-        return null;
-      }
-      windowStopIndex = splitResult.getValue();
-      windowedSplitResult =
-          calculateRestrictionSize(
-              splitResult.getKey(),
-              PTransformTranslation.SPLITTABLE_PROCESS_SIZED_ELEMENTS_AND_RESTRICTIONS_URN
-                  + "/GetSize");
-    }
-
+  @VisibleForTesting
+  static <WatermarkEstimatorStateT> HandlesSplits.SplitResult constructSplitResult(
+      WindowedSplitResult windowedSplitResult,
+      HandlesSplits.SplitResult downstreamElementSplit,
+      Coder fullInputCoder,
+      Instant initialWatermark,
+      KV<Instant, WatermarkEstimatorStateT> watermarkAndState,
+      String pTransformId,
+      String mainInputId,
+      Collection<String> outputIds,
+      Duration resumeDelay) {
+    // The element split cannot from both windowedSplitResult and downstreamElementSplit.
+    checkArgument(

Review comment:
       Yeah both `windowedSplitResult.getResidualSplitRoot` and `downstreamElementSplit` can be null. The only illegal state here is both of them are not null.




----------------------------------------------------------------
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.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org