Makes side inputs per window.

This is a backwards-incompatible change.
     - sideInput() can no longer be called from startBundle/finishBundle.
     - Calls to sideInput() now return values only in a specific window corresponding to the window of the main input element, not the whole side input PCollectionView

----Release Notes----
- sideInput() can no longer be called from startBundle/finishBundle.
- Calls to sideInput() now return values only in a specific window corresponding to the window of the main input element, not the whole side input PCollectionView.  For PCollections and side inputs that are both windowed by GlobalWindows, this is identical to the old behavior

[]
-------------
Created by MOE: http://code.google.com/p/moe-java
MOE_MIGRATED_REVID=89356757

Author: dpmills

examples/src/main/java/com/google/cloud/dataflow/examples/TopWikipediaSessions.java

@@ -126,7 +126,7 @@ public PCollection<List<KV<String, Long>>> apply(PCollection<KV<String, Long>> s
                     public int compare(KV<String, Long> o1, KV<String, Long> o2) {
                       return Long.compare(o1.getValue(), o2.getValue());
                     }
-                  }));
+                  }).withoutDefaults());
     }
   }
 

sdk/src/main/java/com/google/cloud/dataflow/sdk/testing/DataflowAssert.java

@@ -25,6 +25,8 @@
 import com.google.cloud.dataflow.sdk.transforms.ParDo;
 import com.google.cloud.dataflow.sdk.transforms.SerializableFunction;
 import com.google.cloud.dataflow.sdk.transforms.View;
+import com.google.cloud.dataflow.sdk.transforms.windowing.GlobalWindows;
+import com.google.cloud.dataflow.sdk.transforms.windowing.Window;
 import com.google.cloud.dataflow.sdk.values.PCollection;
 import com.google.cloud.dataflow.sdk.values.PCollectionView;
 import com.google.common.base.Optional;
@@ -80,7 +82,8 @@ private DataflowAssert() {}
    * {@link PCollection PCollection<T>}.
    */
   public static <T> IterableAssert<T> that(PCollection<T> actual) {
-    return new IterableAssert<>(actual.apply(View.<T>asIterable()))
+    return
+        new IterableAssert<>(inGlobalWindows(actual).apply(View.<T>asIterable()))
         .setCoder(actual.getCoder());
   }
 
@@ -101,7 +104,7 @@ public static <T> IterableAssert<T> thatSingletonIterable(PCollection<Iterable<T
         + " single Coder<T> to apply to the elements.");
     }
 
-    return new IterableAssert<>(actual.apply(View.<Iterable<T>>asSingleton()))
+    return new IterableAssert<>(inGlobalWindows(actual).apply(View.<Iterable<T>>asSingleton()))
         .setCoder(tCoder);
   }
 
@@ -118,7 +121,7 @@ public static <T> IterableAssert<T> thatIterable(PCollectionView<Iterable<T>> ac
    * {@code PCollection PCollection<T>}, which must be a singleton.
    */
   public static <T> SingletonAssert<T> thatSingleton(PCollection<T> actual) {
-    return new SingletonAssert<>(actual.apply(View.<T>asSingleton()))
+    return new SingletonAssert<>(inGlobalWindows(actual).apply(View.<T>asSingleton()))
         .setCoder(actual.getCoder());
   }
 
@@ -344,6 +347,16 @@ public SingletonAssert<T> is(T expectedValue) {
 
   ////////////////////////////////////////////////////////////////////////
 
+  /**
+   * Returns a new PCollection equivalent to the input, but with all elements
+   * in the GlobalWindow.  Preserves ordering if the input is ordered.
+   */
+  private static <T> PCollection<T> inGlobalWindows(PCollection<T> input) {
+    return input
+        .apply(Window.<T>into(new GlobalWindows()))
+        .setOrdered(input.isOrdered());
+  }
+
   /**
    * An assertion checker that takes a single {@link PCollectionView PCollectionView&lt;A&gt;}
    * and an assertion over {@code A}, and checks it within a dataflow pipeline.

sdk/src/main/java/com/google/cloud/dataflow/sdk/transforms/Combine.java

@@ -59,14 +59,15 @@ public class Combine {
   /**
    * Returns a {@link Globally Combine.Globally} {@code PTransform}
    * that uses the given {@code SerializableFunction} to combine all
-   * the elements of the input {@code PCollection} into a singleton
-   * {@code PCollection} value.  The types of the input elements and the
-   * output value must be the same.
+   * the elements in each window of the input {@code PCollection} into a
+   * single value in the output {@code PCollection}.  The types of the input
+   * elements and the output elements must be the same.
    *
-   * <p>If the input {@code PCollection} is empty, the ouput will contain a the
-   * default value of the combining function if the input is windowed into
-   * the {@link GlobalWindows}; otherwise, the output will be empty.  Note: this
-   * behavior is subject to change.
+   * <p> If the input {@code PCollection} is windowed into {@link GlobalWindows},
+   * a default value in the {@link GlobalWindow} will be output if the input
+   * {@code PCollection} is empty.  To use this with inputs with other windowing,
+   * either {@link Globally#withoutDefaults} or {@link Globally#asSingletonView}
+   * must be called.
    *
    * <p> See {@link Globally Combine.Globally} for more information.
    */
@@ -77,21 +78,22 @@ public static <V> Globally<V, V> globally(
 
   /**
    * Returns a {@link Globally Combine.Globally} {@code PTransform}
-   * that uses the given {@code CombineFn} to combine all the elements
-   * of the input {@code PCollection} into a singleton {@code PCollection}
-   * value.  The types of the input elements and the output value can
-   * differ.
+   * that uses the given {@code SerializableFunction} to combine all
+   * the elements in each window of the input {@code PCollection} into a
+   * single value in the output {@code PCollection}.  The types of the input
+   * elements and the output elements can differ
    *
-   * If the input {@code PCollection} is empty, the ouput will contain a the
-   * default value of the combining function if the input is windowed into
-   * the {@link GlobalWindows}; otherwise, the output will be empty.  Note: this
-   * behavior is subject to change.
+   * <p> If the input {@code PCollection} is windowed into {@link GlobalWindows},
+   * a default value in the {@link GlobalWindow} will be output if the input
+   * {@code PCollection} is empty.  To use this with inputs with other windowing,
+   * either {@link Globally#withoutDefaults} or {@link Globally#asSingletonView}
+   * must be called.
    *
    * <p> See {@link Globally Combine.Globally} for more information.
    */
   public static <VI, VO> Globally<VI, VO> globally(
       CombineFn<? super VI, ?, VO> fn) {
-    return new Globally<>(fn);
+    return new Globally<>(fn, true);
   }
 
   /**
@@ -1053,10 +1055,9 @@ public Coder<VO> getDefaultOutputCoder(
 
   /**
    * {@code Combine.Globally<VI, VO>} takes a {@code PCollection<VI>}
-   * and returns a {@code PCollection<VO>} whose single element is the result of
-   * combining all the elements of the input {@code PCollection},
-   * using a specified}
-   * {@link CombineFn CombineFn&lt;VI, VA, VO&gt;}.  It is common
+   * and returns a {@code PCollection<VO>} whose elements are the result of
+   * combining all the elements in each window of the input {@code PCollection},
+   * using a specified {@link CombineFn CombineFn<VI, VA, VO>}.  It is common
    * for {@code VI == VO}, but not required.  Common combining
    * functions include sums, mins, maxes, and averages of numbers,
    * conjunctions and disjunctions of booleans, statistical
@@ -1074,6 +1075,11 @@ public Coder<VO> getDefaultOutputCoder(
    * intermediate results combined further, in an arbitrary tree
    * reduction pattern, until a single result value is produced.
    *
+   * <p> If the input {@code PCollection} is windowed into {@link GlobalWindows},
+   * a default value in the {@link GlobalWindow} will be output if the input
+   * {@code PCollection} is empty.  To use this with inputs with other windowing,
+   * either {@link #withoutDefaults} or {@link #asSingletonView} must be called.
+   *
    * <p> By default, the {@code Coder} of the output {@code PValue<VO>}
    * is inferred from the concrete type of the
    * {@code CombineFn<VI, VA, VO>}'s output type {@code VO}.
@@ -1090,9 +1096,36 @@ public static class Globally<VI, VO>
       extends PTransform<PCollection<VI>, PCollection<VO>> {
 
     private final CombineFn<? super VI, ?, VO> fn;
+    private final boolean insertDefault;
 
-    private Globally(CombineFn<? super VI, ?, VO> fn) {
+    private Globally(CombineFn<? super VI, ?, VO> fn, boolean insertDefault) {
       this.fn = fn;
+      this.insertDefault = insertDefault;
+    }
+
+    @Override
+    @SuppressWarnings("unchecked")
+    public Globally<VI, VO> withName(String name) {
+      return (Globally<VI, VO>) super.withName(name);
+    }
+
+    /**
+     * Returns a {@link PTransform} that produces a {@code PCollectionView}
+     * whose elements are the result of combining elements per-window in
+     * the input {@code PCollection}.  If a value is requested from the view
+     * for a window that is not present, the result of calling the {@code CombineFn}
+     * on empty input will returned.
+     */
+    public GloballyAsSingletonView<VI, VO> asSingletonView() {
+      return new GloballyAsSingletonView<>(fn, insertDefault);
+    }
+
+    /**
+     * Returns a {@link PTransform} identical to this, but that does not attempt to
+     * provide a default value in the case of empty input.
+     */
+    public Globally<VI, VO> withoutDefaults() {
+      return new Globally<>(fn, false);
     }
 
     @Override
@@ -1103,7 +1136,13 @@ public PCollection<VO> apply(PCollection<VI> input) {
           .apply(Combine.<Void, VI, VO>perKey(fn.<Void>asKeyedFn()))
           .apply(Values.<VO>create());
 
-      if (input.getWindowFn().isCompatible(new GlobalWindows())) {
+      if (insertDefault) {
+        if (!output.getWindowFn().isCompatible(new GlobalWindows())) {
+          throw new IllegalStateException(
+              "Attempted to add default value to PCollection not windowed by GlobalWindows. "
+              + "Instead, use Combine.globally().withoutDefaults() or "
+              + "Combine.globally().asSingletonView().");
+        }
         return insertDefaultValueIfEmpty(output);
       } else {
         return output;
@@ -1117,15 +1156,15 @@ private PCollection<VO> insertDefaultValueIfEmpty(PCollection<VO> maybeEmpty) {
           .apply(Create.of((Void) null)).setCoder(VoidCoder.of())
           .apply(ParDo.of(
               new DoFn<Void, VO>() {
-                  @Override
-                  public void processElement(DoFn<Void, VO>.ProcessContext c) {
-                    Iterator<VO> combined = c.sideInput(maybeEmptyView).iterator();
-                    if (combined.hasNext()) {
-                      c.output(combined.next());
-                    } else {
-                      c.output(fn.apply(Collections.<VI>emptyList()));
-                    }
+                @Override
+                public void processElement(DoFn<Void, VO>.ProcessContext c) {
+                  Iterator<VO> combined = c.sideInput(maybeEmptyView).iterator();
+                  if (combined.hasNext()) {
+                    c.output(combined.next());
+                  } else {
+                    c.output(fn.apply(Collections.<VI>emptyList()));
                   }
+                }
               }).withSideInputs(maybeEmptyView))
           .setCoder(maybeEmpty.getCoder());
     }
@@ -1136,6 +1175,81 @@ protected String getKindString() {
     }
   }
 
+  /**
+   * {@code Combine.GloballyAsSingletonView<VI, VO>} takes a {@code PCollection<VI>}
+   * and returns a {@code PCollectionView<VO>} whose elements are the result of
+   * combining all the elements in each window of the input {@code PCollection},
+   * using a specified {@link CombineFn CombineFn<VI, VA, VO>}. It is common for
+   * {@code VI == VO}, but not required. Common combining
+   * functions include sums, mins, maxes, and averages of numbers,
+   * conjunctions and disjunctions of booleans, statistical
+   * aggregations, etc.
+   *
+   * <p> Example of use:
+   * <pre> {@code
+   * PCollection<Integer> pc = ...;
+   * PCollection<Integer> sum = pc.apply(
+   *     Combine.globally(new Sum.SumIntegerFn()));
+   * } </pre>
+   *
+   * <p> Combining can happen in parallel, with different subsets of the
+   * input {@code PCollection} being combined separately, and their
+   * intermediate results combined further, in an arbitrary tree
+   * reduction pattern, until a single result value is produced.
+   *
+   * <p> If a value is requested from the view for a window that is not present
+   * and {@code insertDefault} is true, the result of calling the {@code CombineFn}
+   * on empty input will returned. If {@code insertDefault} is false, an
+   * exception will be thrown instead.
+   *
+   * <p> By default, the {@code Coder} of the output {@code PValue<VO>}
+   * is inferred from the concrete type of the
+   * {@code CombineFn<VI, VA, VO>}'s output type {@code VO}.
+   *
+   * <p> See also {@link #perKey}/{@link PerKey Combine.PerKey} and
+   * {@link #groupedValues}/{@link GroupedValues Combine.GroupedValues},
+   * which are useful for combining values associated with each key in
+   * a {@code PCollection} of {@code KV}s.
+   *
+   * @param <VI> type of input values
+   * @param <VO> type of output values
+   */
+  public static class GloballyAsSingletonView<VI, VO>
+      extends PTransform<PCollection<VI>, PCollectionView<VO>> {
+
+    private final CombineFn<? super VI, ?, VO> fn;
+    private final boolean insertDefault;
+
+    private GloballyAsSingletonView(CombineFn<? super VI, ?, VO> fn, boolean insertDefault) {
+      this.fn = fn;
+      this.insertDefault = insertDefault;
+    }
+
+    @Override
+    @SuppressWarnings("unchecked")
+    public GloballyAsSingletonView<VI, VO> withName(String name) {
+      return (GloballyAsSingletonView<VI, VO>) super.withName(name);
+    }
+
+    @Override
+    public PCollectionView<VO> apply(PCollection<VI> input) {
+      PCollection<VO> combined = input
+          .apply(Combine.globally(fn).withoutDefaults());
+      if (insertDefault) {
+        return combined
+            .apply(View.<VO>asSingleton().withDefaultValue(
+                fn.apply(Collections.<VI>emptyList())));
+      } else {
+        return combined.apply(View.<VO>asSingleton());
+      }
+    }
+
+    @Override
+    protected String getKindString() {
+      return "Combine.GloballyAsSingletonView";
+    }
+  }
+
   /**
    * Converts a {@link SerializableFunction} from {@code Iterable<V>}s
    * to {@code V}s into a simple {@link CombineFn} over {@code V}s.

sdk/src/main/java/com/google/cloud/dataflow/sdk/transforms/DoFn.java

@@ -66,14 +66,6 @@ public abstract class Context {
      */
     public abstract PipelineOptions getPipelineOptions();
 
-    /**
-     * Returns the value of the side input.
-     *
-     * @throws IllegalArgumentException if this is not a side input
-     * @see ParDo#withSideInputs
-     */
-    public abstract <T> T sideInput(PCollectionView<T> view);
-
     /**
      * Adds the given element to the main output {@code PCollection}.
      *
@@ -207,6 +199,19 @@ public abstract class ProcessContext extends Context {
      */
     public abstract I element();
 
+    /**
+     * Returns the value of the side input for the window corresponding to the
+     * window of the main input element.
+     *
+     * <p> See
+     * {@link com.google.cloud.dataflow.sdk.transforms.windowing.WindowFn#getSideInputWindow}
+     * for how this corresponding window is determined.
+     *
+     * @throws IllegalArgumentException if this is not a side input
+     * @see ParDo#withSideInputs
+     */
+    public abstract <T> T sideInput(PCollectionView<T> view);
+
     /**
      * Returns this {@code DoFn}'s state associated with the input
      * element's key.  This state can be used by the {@code DoFn} to

sdk/src/main/java/com/google/cloud/dataflow/sdk/transforms/Top.java

@@ -80,10 +80,9 @@ public class Top {
    * {@code KV}s and return the top values associated with each key.
    */
   public static <T, C extends Comparator<T> & Serializable>
-      PTransform<PCollection<T>, PCollection<List<T>>> of(int count, C compareFn) {
+      Combine.Globally<T, List<T>> of(int count, C compareFn) {
     return Combine.globally(new TopCombineFn<>(count, compareFn))
         .withName("Top");
-
   }
 
   /**
@@ -121,7 +120,7 @@ PTransform<PCollection<T>, PCollection<List<T>>> of(int count, C compareFn) {
    * {@code KV}s and return the top values associated with each key.
    */
   public static <T extends Comparable<T>>
-      PTransform<PCollection<T>, PCollection<List<T>>> smallest(int count) {
+      Combine.Globally<T, List<T>> smallest(int count) {
     return Combine.globally(new TopCombineFn<>(count, new Smallest<T>()))
         .withName("Top.Smallest");
   }
@@ -161,7 +160,7 @@ PTransform<PCollection<T>, PCollection<List<T>>> smallest(int count) {
    * {@code KV}s and return the top values associated with each key.
    */
   public static <T extends Comparable<T>>
-      PTransform<PCollection<T>, PCollection<List<T>>> largest(int count) {
+      Combine.Globally<T, List<T>> largest(int count) {
     return Combine.globally(new TopCombineFn<>(count, new Largest<T>()))
         .withName("Top.Largest");
   }