Update Triggers to new shouldFire() based semantics

User-facing trigger expressions are unchanged in syntax or semantics.

The high-level change to the internal-facing API is as follows:

 - onElement no longer returns a TriggerResult, now void
 - onMerge no longer returns a MergeResult, now void
 - onTimer no longer exists in Trigger (it exists in the runner only)
 - shouldFire says whether to fire; the runner controls when to ask
 - onFire transitions/resets a trigger for sequential firings

This also addresses a particular semantic issue where OrFinally did not
fire as soon as expected.

----Release Notes----
 - The "finally" branch of OrFinally now fires as soon as it is
   satisfied; before it would often take much longer due to seeing
   only the prefix of each work unit that satisfied the "main" branch.

[]
-------------
Created by MOE: https://github.com/google/moe
MOE_MIGRATED_REVID=112809409

Author: kennknowles

sdk/src/main/java/com/google/cloud/dataflow/sdk/transforms/windowing/AfterAll.java

@@ -49,59 +49,25 @@ public static <W extends BoundedWindow> OnceTrigger<W> of(
     return new AfterAll<W>(Arrays.<Trigger<W>>asList(triggers));
   }
 
-  private TriggerResult result(TriggerContext c) {
-    // If all children have finished, then they must have each fired at least once.
-    if (c.trigger().areAllSubtriggersFinished()) {
-      return TriggerResult.FIRE_AND_FINISH;
-    }
-
-    return TriggerResult.CONTINUE;
-  }
-
   @Override
-  public TriggerResult onElement(OnElementContext c) throws Exception {
+  public void onElement(OnElementContext c) throws Exception {
     for (ExecutableTrigger<W> subTrigger : c.trigger().unfinishedSubTriggers()) {
       // Since subTriggers are all OnceTriggers, they must either CONTINUE or FIRE_AND_FINISH.
       // invokeElement will automatically mark the finish bit if they return FIRE_AND_FINISH.
-      subTrigger.invokeElement(c);
+      subTrigger.invokeOnElement(c);
     }
-
-    return result(c);
   }
 
   @Override
-  public MergeResult onMerge(OnMergeContext c) throws Exception {
-    // CONTINUE if merging returns CONTINUE for at least one sub-trigger
-    // ALREADY_FINISHED if merging returns ALREADY_FINISHED for all sub-triggers and this
-    // trigger itself was already finished in some window.
-    // FIRE_AND_FINISH otherwise: It means this trigger is ready to fire (because all subtriggers
-    // are satisfied) but has never fired as a whole.
-    boolean anyContinue = false;
-    boolean alreadyFinished = true;
+  public void onMerge(OnMergeContext c) throws Exception {
     for (ExecutableTrigger<W> subTrigger : c.trigger().subTriggers()) {
-      MergeResult result = subTrigger.invokeMerge(c);
-      anyContinue |= !result.isFire() && !result.isFinish();
-      alreadyFinished &= !result.isFire() && result.isFinish();
+      subTrigger.invokeOnMerge(c);
     }
-
-    if (anyContinue) {
-      return MergeResult.CONTINUE;
-    } else if (alreadyFinished && c.trigger().finishedInAnyMergingWindow()) {
-      return MergeResult.ALREADY_FINISHED;
-    } else {
-      return MergeResult.FIRE_AND_FINISH;
+    boolean allFinished = true;
+    for (ExecutableTrigger<W> subTrigger1 : c.trigger().subTriggers()) {
+      allFinished &= c.forTrigger(subTrigger1).trigger().isFinished();
     }
-  }
-
-  @Override
-  public TriggerResult onTimer(OnTimerContext c) throws Exception {
-    for (ExecutableTrigger<W> subTrigger : c.trigger().unfinishedSubTriggers()) {
-      // Since subTriggers are all OnceTriggers, they must either CONTINUE or FIRE_AND_FINISH.
-      // invokeTimer will automatically mark the finish bit if they return FIRE_AND_FINISH.
-      subTrigger.invokeTimer(c);
-    }
-
-    return result(c);
+    c.trigger().setFinished(allFinished);
   }
 
   @Override
@@ -121,4 +87,31 @@ public Instant getWatermarkThatGuaranteesFiring(W window) {
   public OnceTrigger<W> getContinuationTrigger(List<Trigger<W>> continuationTriggers) {
     return new AfterAll<W>(continuationTriggers);
   }
+
+  /**
+   * {@inheritDoc}
+   *
+   * @return {@code true} if all subtriggers return {@code true}.
+   */
+  @Override
+  public boolean shouldFire(TriggerContext context) throws Exception {
+    for (ExecutableTrigger<W> subtrigger : context.trigger().subTriggers()) {
+      if (!context.forTrigger(subtrigger).trigger().isFinished()
+          && !subtrigger.invokeShouldFire(context)) {
+        return false;
+      }
+    }
+    return true;
+  }
+
+  /**
+   * Invokes {@link #onFire} for all subtriggers, eliding redundant calls to {@link #shouldFire}
+   * because they all must be ready to fire.
+   */
+  @Override
+  public void onOnlyFiring(TriggerContext context) throws Exception {
+    for (ExecutableTrigger<W> subtrigger : context.trigger().subTriggers()) {
+      subtrigger.invokeOnFire(context);
+    }
+  }
 }

sdk/src/main/java/com/google/cloud/dataflow/sdk/transforms/windowing/AfterDelayFromFirstElement.java

@@ -0,0 +1,127 @@
+/*
+ * Copyright (C) 2015 Google Inc.
+ *
+ * Licensed 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 com.google.cloud.dataflow.sdk.transforms.windowing;
+
+import com.google.cloud.dataflow.sdk.transforms.SerializableFunction;
+import com.google.cloud.dataflow.sdk.util.ReduceFn.MergingStateContext;
+import com.google.cloud.dataflow.sdk.util.ReduceFn.StateContext;
+import com.google.cloud.dataflow.sdk.util.TimeDomain;
+import com.google.cloud.dataflow.sdk.util.state.CombiningValueState;
+
+import org.joda.time.Instant;
+
+import java.util.List;
+
+import javax.annotation.Nullable;
+
+/**
+ * A base class for triggers that happen after a processing time delay from the arrival
+ * of the first element in a pane.
+ */
+abstract class AfterDelayFromFirstElement<W extends BoundedWindow> extends TimeTrigger<W> {
+
+  /**
+   * To complete an implementation, return the desired time from the TriggerContext.
+   */
+  @Nullable
+  public abstract Instant getCurrentTime(Trigger<W>.TriggerContext context);
+
+  private final TimeDomain timeDomain;
+
+  public AfterDelayFromFirstElement(
+      TimeDomain timeDomain, List<SerializableFunction<Instant, Instant>> timestampMappers) {
+    super(timestampMappers);
+    this.timeDomain = timeDomain;
+  }
+
+  private Instant getTargetTimestamp(OnElementContext c) {
+    return computeTargetTimestamp(c.currentProcessingTime());
+  }
+
+  @Override
+  public void prefetchOnElement(StateContext state) {
+    state.access(DELAYED_UNTIL_TAG).get();
+  }
+
+  @Override
+  public void onElement(OnElementContext c) throws Exception {
+    CombiningValueState<Instant, Instant> delayUntilState = c.state().access(DELAYED_UNTIL_TAG);
+    Instant oldDelayUntil = delayUntilState.get().read();
+
+    // Since processing time can only advance, resulting in target wake-up times we would
+    // ignore anyhow, we don't bother with it if it is already set.
+    if (oldDelayUntil != null) {
+      return;
+    }
+
+    Instant targetTimestamp = getTargetTimestamp(c);
+    delayUntilState.add(targetTimestamp);
+    c.setTimer(targetTimestamp, timeDomain);
+  }
+
+  @Override
+  public void prefetchOnMerge(MergingStateContext state) {
+    state.mergingAccess(DELAYED_UNTIL_TAG).get();
+  }
+
+  @Override
+  public void onMerge(OnMergeContext c) throws Exception {
+    // If the trigger is already finished, there is no way it will become re-activated
+    if (c.trigger().isFinished()) {
+      return;
+    }
+
+    // Determine the earliest point across all the windows, and delay to that.
+    CombiningValueState<Instant, Instant> mergingDelays =
+        c.state().mergingAccess(DELAYED_UNTIL_TAG);
+
+    Instant earliestTargetTime = mergingDelays.get().read();
+    if (earliestTargetTime != null) {
+      mergingDelays.clear();
+      mergingDelays.add(earliestTargetTime);
+      c.setTimer(earliestTargetTime, timeDomain);
+    }
+  }
+
+  @Override
+  public void prefetchShouldFire(StateContext state) {
+    state.access(DELAYED_UNTIL_TAG).get();
+  }
+
+  @Override
+  public void clear(TriggerContext c) throws Exception {
+    c.state().access(DELAYED_UNTIL_TAG).clear();
+  }
+
+  @Override
+  public Instant getWatermarkThatGuaranteesFiring(W window) {
+    return BoundedWindow.TIMESTAMP_MAX_VALUE;
+  }
+
+  @Override
+  public boolean shouldFire(Trigger<W>.TriggerContext context) throws Exception {
+    Instant delayedUntil = context.state().access(DELAYED_UNTIL_TAG).get().read();
+    return delayedUntil != null
+        && getCurrentTime(context) != null
+        && getCurrentTime(context).isAfter(delayedUntil);
+  }
+
+  @Override
+  protected void onOnlyFiring(Trigger<W>.TriggerContext context) throws Exception {
+    clear(context);
+  }
+}

sdk/src/main/java/com/google/cloud/dataflow/sdk/transforms/windowing/AfterEach.java

@@ -16,14 +16,14 @@
 
 package com.google.cloud.dataflow.sdk.transforms.windowing;
 
+import static com.google.common.base.Preconditions.checkArgument;
+
 import com.google.cloud.dataflow.sdk.annotations.Experimental;
 import com.google.cloud.dataflow.sdk.util.ExecutableTrigger;
-import com.google.common.base.Preconditions;
 
 import org.joda.time.Instant;
 
 import java.util.Arrays;
-import java.util.Iterator;
 import java.util.List;
 
 /**
@@ -50,7 +50,7 @@ public class AfterEach<W extends BoundedWindow> extends Trigger<W> {
 
   private AfterEach(List<Trigger<W>> subTriggers) {
     super(subTriggers);
-    Preconditions.checkArgument(subTriggers.size() > 1);
+    checkArgument(subTriggers.size() > 1);
   }
 
   /**
@@ -61,94 +61,75 @@ public static <W extends BoundedWindow> Trigger<W> inOrder(Trigger<W>... trigger
     return new AfterEach<W>(Arrays.<Trigger<W>>asList(triggers));
   }
 
-  private TriggerResult result(TriggerContext c, TriggerResult subResult)
-      throws Exception {
-    if (subResult.isFire()) {
-      return c.trigger().areAllSubtriggersFinished()
-          ? TriggerResult.FIRE_AND_FINISH : TriggerResult.FIRE;
+  @Override
+  public void onElement(OnElementContext c) throws Exception {
+    if (!c.trigger().isMerging()) {
+      // If merges are not possible, we need only run the first unfinished subtrigger
+      c.trigger().firstUnfinishedSubTrigger().invokeOnElement(c);
     } else {
-      return TriggerResult.CONTINUE;
+      // If merges are possible, we need to run all subtriggers in parallel
+      for (ExecutableTrigger<W> subTrigger :  c.trigger().subTriggers()) {
+        // Even if the subTrigger is done, it may be revived via merging and must have
+        // adequate state.
+        subTrigger.invokeOnElement(c);
+      }
     }
   }
 
   @Override
-  public TriggerResult onElement(OnElementContext c) throws Exception {
-    Iterator<ExecutableTrigger<W>> iterator = c.trigger().unfinishedSubTriggers().iterator();
-
-    // If all the sub-triggers have finished, we should have already finished, so we know there is
-    // at least one unfinished trigger.
-    TriggerResult firstResult = iterator.next().invokeElement(c);
-
-    // If onMerge might be called, we need to make sure we have proper state for future triggers.
-    if (c.trigger().isMerging()) {
-      if (firstResult.isFire()) {
-        // If we're firing, clear out all of the later subtriggers, since we don't want to pollute
-        // their state.
-        resetRemaining(c, iterator);
+  public void onMerge(OnMergeContext context) throws Exception {
+    // If merging makes a subtrigger no-longer-finished, it will automatically
+    // begin participating in shouldFire and onFire appropriately.
+
+    // All the following triggers are retroactively "not started" but that is
+    // also automatic because they are cleared whenever this trigger
+    // fires.
+    boolean priorTriggersAllFinished = true;
+    for (ExecutableTrigger<W> subTrigger : context.trigger().subTriggers()) {
+      if (priorTriggersAllFinished) {
+        subTrigger.invokeOnMerge(context);
+        priorTriggersAllFinished &= context.forTrigger(subTrigger).trigger().isFinished();
       } else {
-        // Otherwise, iterate over all of them to build up some state.
-        while (iterator.hasNext()) {
-          iterator.next().invokeElement(c);
-        }
+        subTrigger.invokeClear(context);
       }
     }
-
-    return result(c, firstResult);
+    updateFinishedState(context);
   }
 
   @Override
-  public MergeResult onMerge(OnMergeContext c) throws Exception {
-    // Iterate over the sub-triggers to identify the "current" sub-trigger.
-    Iterator<ExecutableTrigger<W>> iterator = c.trigger().subTriggers().iterator();
-    while (iterator.hasNext()) {
-      ExecutableTrigger<W> subTrigger = iterator.next();
-
-      MergeResult mergeResult = subTrigger.invokeMerge(c);
-
-      if (MergeResult.CONTINUE.equals(mergeResult)) {
-        resetRemaining(c, iterator);
-        return MergeResult.CONTINUE;
-      } else if (MergeResult.FIRE.equals(mergeResult)) {
-        resetRemaining(c, iterator);
-        return MergeResult.FIRE;
-      } else if (MergeResult.FIRE_AND_FINISH.equals(mergeResult)) {
-        resetRemaining(c, iterator);
-        return c.trigger().areAllSubtriggersFinished()
-            ? MergeResult.FIRE_AND_FINISH : MergeResult.FIRE;
-      }
-    }
-
-    // If we get here, all the merges indicated they were finished, which means there was at least
-    // one merged window in which the triggers had all already finished. Given that, this AfterEach
-    // would have already finished in that window as well. Since the window was still in the window
-    // set for merging, we can return FINISHED (because we were finished in that window) and we also
-    // know that there must be another trigger (parent or sibling) which hasn't finished yet, which
-    // will FIRE, CONTINUE, or FIRE_AND_FINISH.
-    return MergeResult.ALREADY_FINISHED;
+  public Instant getWatermarkThatGuaranteesFiring(W window) {
+    // This trigger will fire at least once when the first trigger in the sequence
+    // fires at least once.
+    return subTriggers.get(0).getWatermarkThatGuaranteesFiring(window);
   }
 
-  private void resetRemaining(
-      TriggerContext c, Iterator<ExecutableTrigger<W>> triggers) throws Exception {
-    while (triggers.hasNext()) {
-      c.forTrigger(triggers.next()).trigger().resetTree();
-    }
+  @Override
+  public Trigger<W> getContinuationTrigger(List<Trigger<W>> continuationTriggers) {
+    return Repeatedly.forever(new AfterFirst<W>(continuationTriggers));
   }
 
   @Override
-  public TriggerResult onTimer(OnTimerContext c) throws Exception {
-    // Only deliver to the currently active subtrigger
-    return result(c, c.trigger().firstUnfinishedSubTrigger().invokeTimer(c));
+  public boolean shouldFire(Trigger<W>.TriggerContext context) throws Exception {
+    ExecutableTrigger<W> firstUnfinished = context.trigger().firstUnfinishedSubTrigger();
+    return firstUnfinished.invokeShouldFire(context);
   }
 
   @Override
-  public Instant getWatermarkThatGuaranteesFiring(W window) {
-    // This trigger will fire at least once when the first trigger in the sequence
-    // fires at least once.
-    return subTriggers.get(0).getWatermarkThatGuaranteesFiring(window);
+  public void onFire(Trigger<W>.TriggerContext context) throws Exception {
+    context.trigger().firstUnfinishedSubTrigger().invokeOnFire(context);
+
+    // Reset all subtriggers if in a merging context; any may be revived by merging so they are
+    // all run in parallel for each pending pane.
+    if (context.trigger().isMerging()) {
+      for (ExecutableTrigger<W> subTrigger : context.trigger().subTriggers()) {
+        subTrigger.invokeClear(context);
+      }
+    }
+
+    updateFinishedState(context);
   }
 
-  @Override
-  public Trigger<W> getContinuationTrigger(List<Trigger<W>> continuationTriggers) {
-    return Repeatedly.forever(new AfterFirst<W>(continuationTriggers));
+  private void updateFinishedState(TriggerContext context) {
+    context.trigger().setFinished(context.trigger().firstUnfinishedSubTrigger() == null);
   }
 }