Eliminate WindowedValue.getPane() in preparation for making it a user-facing interface

Author: kennknowles

examples/java/src/main/java/org/apache/beam/examples/KafkaStreaming.java

@@ -300,7 +300,7 @@ public void processElement(ProcessContext c, IntervalWindow w) throws Exception
       String startTime = w.start().toString(dateTimeFormatter);
       String endTime = w.end().toString(dateTimeFormatter);
 
-      PaneInfo.Timing timing = c.pane().getTiming();
+      PaneInfo.Timing timing = c.paneInfo().getTiming();
 
       switch (timing) {
         case EARLY:

examples/java/src/main/java/org/apache/beam/examples/complete/game/LeaderBoard.java

@@ -155,7 +155,7 @@ public interface Options extends ExampleOptions, StreamingOptions {
     tableConfigure.put(
         "timing",
         new WriteWindowedToBigQuery.FieldInfo<>(
-            "STRING", (c, w) -> c.pane().getTiming().toString()));
+            "STRING", (c, w) -> c.paneInfo().getTiming().toString()));
     return tableConfigure;
   }
 

examples/java/src/main/java/org/apache/beam/examples/cookbook/TriggerExample.java

@@ -401,9 +401,9 @@ public void processElement(ProcessContext c, BoundedWindow window) throws Except
               .set("total_flow", Integer.parseInt(values[0]))
               .set("number_of_records", Long.parseLong(values[1]))
               .set("window", window.toString())
-              .set("isFirst", c.pane().isFirst())
-              .set("isLast", c.pane().isLast())
-              .set("timing", c.pane().getTiming().toString())
+              .set("isFirst", c.paneInfo().isFirst())
+              .set("isLast", c.paneInfo().isLast())
+              .set("timing", c.paneInfo().getTiming().toString())
               .set("event_time", c.timestamp().toString())
               .set("processing_time", Instant.now().toString());
       c.output(row);

examples/java/src/test/java/org/apache/beam/examples/complete/game/LeaderBoardTest.java

@@ -99,7 +99,7 @@ public void testTeamScoresOnTime() {
             .addElements(
                 event(TestUser.RED_ONE, 1, Duration.standardMinutes(4)),
                 event(TestUser.BLUE_ONE, 2, Duration.standardSeconds(270)))
-            // The window should close and emit an ON_TIME pane
+            // The window should close and emit an ON_TIME paneInfo
             .advanceWatermarkToInfinity();
 
     PCollection<KV<String, Integer>> teamScores =
@@ -137,7 +137,8 @@ public void testTeamScoresSpeculative() {
             // Some additional time passes and we get a speculative pane for the red team
             .advanceProcessingTime(Duration.standardMinutes(12))
             .addElements(event(TestUser.BLUE_TWO, 3, Duration.standardSeconds(22)))
-            // More time passes and a speculative pane containing a refined value for the blue pane
+            // More time passes and a speculative pane containing a refined value for the blue
+            // paneInfo
             // is
             // emitted
             .advanceProcessingTime(Duration.standardMinutes(10))
@@ -155,7 +156,7 @@ public void testTeamScoresSpeculative() {
     String blueTeam = TestUser.BLUE_ONE.getTeam();
     String redTeam = TestUser.RED_ONE.getTeam();
     IntervalWindow window = new IntervalWindow(baseTime, TEAM_WINDOW_DURATION);
-    // The window contains speculative panes alongside the on-time pane
+    // The window contains speculative panes alongside the on-time paneInfo
     PAssert.that(teamScores)
         .inWindow(window)
         .containsInAnyOrder(
@@ -190,7 +191,7 @@ public void testTeamScoresUnobservablyLate() {
             .advanceWatermarkTo(
                 baseTime.plus(TEAM_WINDOW_DURATION).minus(Duration.standardMinutes(1)))
             // These events are late, but the window hasn't closed yet, so the elements are in the
-            // on-time pane
+            // on-time paneInfo
             .addElements(
                 event(TestUser.RED_TWO, 2, Duration.ZERO),
                 event(TestUser.RED_TWO, 5, Duration.standardMinutes(1)),
@@ -235,7 +236,7 @@ public void testTeamScoresObservablyLate() {
                 event(TestUser.RED_ONE, 4, Duration.standardMinutes(2)),
                 event(TestUser.BLUE_ONE, 3, Duration.standardMinutes(5)))
             .advanceWatermarkTo(firstWindowCloses.minus(Duration.standardMinutes(1)))
-            // These events are late but should still appear in a late pane
+            // These events are late but should still appear in a late paneInfo
             .addElements(
                 event(TestUser.RED_TWO, 2, Duration.ZERO),
                 event(TestUser.RED_TWO, 5, Duration.standardMinutes(1)),
@@ -244,7 +245,7 @@ public void testTeamScoresObservablyLate() {
             // has
             // not yet closed because the watermark has not advanced
             .advanceProcessingTime(Duration.standardMinutes(12))
-            // These elements should appear in the final pane
+            // These elements should appear in the final paneInfo
             .addElements(
                 event(TestUser.RED_TWO, 9, Duration.standardMinutes(1)),
                 event(TestUser.RED_TWO, 1, Duration.standardMinutes(3)))

runners/core-java/src/main/java/org/apache/beam/runners/core/LateDataDroppingDoFnRunner.java

@@ -142,7 +142,7 @@ public <K, InputT> Iterable<WindowedValue<InputT>> filter(
           } else {
             nonLateElements.add(
                 WindowedValues.of(
-                    element.getValue(), element.getTimestamp(), window, element.getPane()));
+                    element.getValue(), element.getTimestamp(), window, element.getPaneInfo()));
           }
         }
       }

runners/core-java/src/main/java/org/apache/beam/runners/core/OutputAndTimeBoundedSplittableProcessElementInvoker.java

@@ -379,8 +379,8 @@ public Instant timestamp() {
     }
 
     @Override
-    public PaneInfo pane() {
-      return element.getPane();
+    public PaneInfo paneInfo() {
+      return element.getPaneInfo();
     }
 
     @Override
@@ -395,7 +395,7 @@ public void output(OutputT output) {
 
     @Override
     public void outputWithTimestamp(OutputT value, Instant timestamp) {
-      outputWindowedValue(value, timestamp, element.getWindows(), element.getPane());
+      outputWindowedValue(value, timestamp, element.getWindows(), element.getPaneInfo());
     }
 
     @Override
@@ -419,7 +419,7 @@ public <T> void output(TupleTag<T> tag, T value) {
     @Override
     public <T> void outputWithTimestamp(TupleTag<T> tag, T value, Instant timestamp) {
       outputReceiver.output(
-          tag, WindowedValues.of(value, timestamp, element.getWindows(), element.getPane()));
+          tag, WindowedValues.of(value, timestamp, element.getWindows(), element.getPaneInfo()));
     }
 
     @Override

runners/core-java/src/main/java/org/apache/beam/runners/core/PaneInfoTracker.java

@@ -48,7 +48,7 @@ public PaneInfoTracker(TimerInternals timerInternals) {
 
   @VisibleForTesting
   static final StateTag<ValueState<PaneInfo>> PANE_INFO_TAG =
-      StateTags.makeSystemTagInternal(StateTags.value("pane", PaneInfoCoder.INSTANCE));
+      StateTags.makeSystemTagInternal(StateTags.value("paneInfo", PaneInfoCoder.INSTANCE));
 
   public void clear(StateAccessor<?> state) {
     state.access(PANE_INFO_TAG).clear();

runners/core-java/src/main/java/org/apache/beam/runners/core/ReduceFnContextFactory.java

@@ -103,8 +103,8 @@ public ReduceFn<K, InputT, OutputT, W>.ProcessValueContext forValue(
   }
 
   public ReduceFn<K, InputT, OutputT, W>.OnTriggerContext forTrigger(
-      W window, PaneInfo pane, StateStyle style, OnTriggerCallbacks<OutputT> callbacks) {
-    return new OnTriggerContextImpl(stateAccessor(window, style), pane, callbacks);
+      W window, PaneInfo paneInfo, StateStyle style, OnTriggerCallbacks<OutputT> callbacks) {
+    return new OnTriggerContextImpl(stateAccessor(window, style), paneInfo, callbacks);
   }
 
   public ReduceFn<K, InputT, OutputT, W>.OnMergeContext forMerge(
@@ -402,15 +402,15 @@ public Timers timers() {
 
   private class OnTriggerContextImpl extends ReduceFn<K, InputT, OutputT, W>.OnTriggerContext {
     private final StateAccessorImpl<K, W> state;
-    private final PaneInfo pane;
+    private final PaneInfo paneInfo;
     private final OnTriggerCallbacks<OutputT> callbacks;
     private final TimersImpl timers;
 
     private OnTriggerContextImpl(
-        StateAccessorImpl<K, W> state, PaneInfo pane, OnTriggerCallbacks<OutputT> callbacks) {
+        StateAccessorImpl<K, W> state, PaneInfo paneInfo, OnTriggerCallbacks<OutputT> callbacks) {
       reduceFn.super();
       this.state = state;
-      this.pane = pane;
+      this.paneInfo = paneInfo;
       this.callbacks = callbacks;
       this.timers = new TimersImpl(state.namespace());
     }
@@ -437,7 +437,7 @@ public StateAccessor<K> state() {
 
     @Override
     public PaneInfo paneInfo() {
-      return pane;
+      return paneInfo;
     }
 
     @Override

runners/core-java/src/main/java/org/apache/beam/runners/core/ReduceFnRunner.java

@@ -1037,28 +1037,28 @@ private void prefetchOnTrigger(
     }
 
     // Calculate the pane info.
-    final PaneInfo pane = paneInfoTracker.getNextPaneInfo(directContext, isFinished).read();
+    final PaneInfo paneInfo = paneInfoTracker.getNextPaneInfo(directContext, isFinished).read();
 
     // Only emit a pane if it has data or empty panes are observable.
-    if (needToEmit(isEmpty, isFinished, pane.getTiming())) {
+    if (needToEmit(isEmpty, isFinished, paneInfo.getTiming())) {
       // Run reduceFn.onTrigger method.
       final List<W> windows = Collections.singletonList(directContext.window());
       ReduceFn<K, InputT, OutputT, W>.OnTriggerContext renamedTriggerContext =
           contextFactory.forTrigger(
               directContext.window(),
-              pane,
+              paneInfo,
               StateStyle.RENAMED,
               toOutput -> {
                 // We're going to output panes, so commit the (now used) PaneInfo.
                 // This is unnecessary if the trigger isFinished since the saved
                 // state will be immediately deleted.
                 if (!isFinished) {
-                  paneInfoTracker.storeCurrentPaneInfo(directContext, pane);
+                  paneInfoTracker.storeCurrentPaneInfo(directContext, paneInfo);
                 }
 
                 // Output the actual value.
                 outputter.output(
-                    WindowedValues.of(KV.of(key, toOutput), outputTimestamp, windows, pane));
+                    WindowedValues.of(KV.of(key, toOutput), outputTimestamp, windows, paneInfo));
               });
 
       reduceFn.onTrigger(renamedTriggerContext);

runners/core-java/src/main/java/org/apache/beam/runners/core/SimpleDoFnRunner.java

@@ -404,8 +404,8 @@ public <T> T sideInput(PCollectionView<T> view) {
     }
 
     @Override
-    public PaneInfo pane() {
-      return elem.getPane();
+    public PaneInfo paneInfo() {
+      return elem.getPaneInfo();
     }
 
     @Override
@@ -437,7 +437,7 @@ public <T> void output(TupleTag<T> tag, T output) {
     public <T> void outputWithTimestamp(TupleTag<T> tag, T output, Instant timestamp) {
       checkNotNull(tag, "Tag passed to outputWithTimestamp cannot be null");
       checkTimestamp(elem.getTimestamp(), timestamp);
-      outputWindowedValue(tag, output, timestamp, elem.getWindows(), elem.getPane());
+      outputWindowedValue(tag, output, timestamp, elem.getWindows(), elem.getPaneInfo());
     }
 
     @Override
@@ -467,7 +467,7 @@ public BoundedWindow window() {
 
     @Override
     public PaneInfo paneInfo(DoFn<InputT, OutputT> doFn) {
-      return pane();
+      return paneInfo();
     }
 
     @Override