[FLINK-3521] Make Iterable part of method signature for WindowFunction

This closes apache#1723

flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/ml/IncrementalLearningSkeleton.java

@@ -167,7 +167,7 @@ public Watermark checkAndGetNextWatermark(Integer lastElement, long extractedTim
 	/**
 	 * Builds up-to-date partial models on new training data.
 	 */
-	public static class PartialModelBuilder implements AllWindowFunction<Iterable<Integer>, Double[], TimeWindow> {
+	public static class PartialModelBuilder implements AllWindowFunction<Integer, Double[], TimeWindow> {
 		private static final long serialVersionUID = 1L;
 
 		protected Double[] buildPartialModel(Iterable<Integer> values) {

flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/windowing/GroupedProcessingTimeWindowExample.java

@@ -104,7 +104,7 @@ public Key getKey(Type value) {
 		}
 	}
 
-	public static class SummingWindowFunction implements WindowFunction<Iterable<Tuple2<Long, Long>>, Tuple2<Long, Long>, Long, Window> {
+	public static class SummingWindowFunction implements WindowFunction<Tuple2<Long, Long>, Tuple2<Long, Long>, Long, Window> {
 
 		@Override
 		public void apply(Long key, Window window, Iterable<Tuple2<Long, Long>> values, Collector<Tuple2<Long, Long>> out) {

flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/AllWindowedStream.java

@@ -26,7 +26,6 @@
 import org.apache.flink.api.common.functions.RichFunction;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.api.java.Utils;
-import org.apache.flink.api.java.typeutils.GenericTypeInfo;
 import org.apache.flink.api.java.typeutils.TypeExtractor;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 import org.apache.flink.streaming.api.functions.aggregation.AggregationFunction;
@@ -222,11 +221,10 @@ public AllWindowedStream<T, W> evictor(Evictor<? super T, ? super W> evictor) {
 	 * @param function The window function.
 	 * @return The data stream that is the result of applying the window function to the window.
 	 */
-	public <R> SingleOutputStreamOperator<R, ?> apply(AllWindowFunction<Iterable<T>, R, W> function) {
+	public <R> SingleOutputStreamOperator<R, ?> apply(AllWindowFunction<T, R, W> function) {
 		@SuppressWarnings("unchecked, rawtypes")
-		TypeInformation<Iterable<T>> iterTypeInfo = new GenericTypeInfo<>((Class) Iterable.class);
 		TypeInformation<R> resultType = TypeExtractor.getUnaryOperatorReturnType(
-				function, AllWindowFunction.class, true, true, iterTypeInfo, null, false);
+				function, AllWindowFunction.class, true, true, getInputType(), null, false);
 
 		return apply(function, resultType);
 	}
@@ -242,7 +240,7 @@ public AllWindowedStream<T, W> evictor(Evictor<? super T, ? super W> evictor) {
 	 * @param function The window function.
 	 * @return The data stream that is the result of applying the window function to the window.
 	 */
-	public <R> SingleOutputStreamOperator<R, ?> apply(AllWindowFunction<Iterable<T>, R, W> function, TypeInformation<R> resultType) {
+	public <R> SingleOutputStreamOperator<R, ?> apply(AllWindowFunction<T, R, W> function, TypeInformation<R> resultType) {
 		//clean the closure
 		function = input.getExecutionEnvironment().clean(function);
 

flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/CoGroupedStreams.java

@@ -556,7 +556,7 @@ public KEY getKey(TaggedUnion<T1, T2> value) throws Exception{
 
 	private static class CoGroupWindowFunction<T1, T2, T, KEY, W extends Window>
 			extends WrappingFunction<CoGroupFunction<T1, T2, T>>
-			implements WindowFunction<Iterable<TaggedUnion<T1, T2>>, T, KEY, W> {
+			implements WindowFunction<TaggedUnion<T1, T2>, T, KEY, W> {
 
 		private static final long serialVersionUID = 1L;
 

flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/WindowedStream.java

@@ -30,7 +30,6 @@
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.api.java.Utils;
 import org.apache.flink.api.java.functions.KeySelector;
-import org.apache.flink.api.java.typeutils.GenericTypeInfo;
 import org.apache.flink.api.java.typeutils.TypeExtractor;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 import org.apache.flink.streaming.api.functions.aggregation.AggregationFunction;
@@ -52,6 +51,8 @@
 import org.apache.flink.streaming.runtime.operators.windowing.AccumulatingProcessingTimeWindowOperator;
 import org.apache.flink.streaming.runtime.operators.windowing.AggregatingProcessingTimeWindowOperator;
 import org.apache.flink.streaming.runtime.operators.windowing.EvictingWindowOperator;
+import org.apache.flink.streaming.runtime.operators.windowing.functions.InternalIterableWindowFunction;
+import org.apache.flink.streaming.runtime.operators.windowing.functions.InternalSingleValueWindowFunction;
 import org.apache.flink.streaming.runtime.operators.windowing.WindowOperator;
 import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
 import org.apache.flink.streaming.runtime.streamrecord.StreamRecordSerializer;
@@ -218,11 +219,9 @@ public WindowedStream<T, K, W> evictor(Evictor<? super T, ? super W> evictor) {
 	 * @param function The window function.
 	 * @return The data stream that is the result of applying the window function to the window.
 	 */
-	public <R> SingleOutputStreamOperator<R, ?> apply(WindowFunction<Iterable<T>, R, K, W> function) {
-		@SuppressWarnings("unchecked, rawtypes")
-		TypeInformation<Iterable<T>> iterTypeInfo = new GenericTypeInfo<>((Class) Iterable.class);
+	public <R> SingleOutputStreamOperator<R, ?> apply(WindowFunction<T, R, K, W> function) {
 		TypeInformation<R> resultType = TypeExtractor.getUnaryOperatorReturnType(
-				function, WindowFunction.class, true, true, iterTypeInfo, null, false);
+				function, WindowFunction.class, true, true, getInputType(), null, false);
 
 		return apply(function, resultType);
 	}
@@ -240,7 +239,7 @@ public WindowedStream<T, K, W> evictor(Evictor<? super T, ? super W> evictor) {
 	 * @param resultType Type information for the result type of the window function
 	 * @return The data stream that is the result of applying the window function to the window.
 	 */
-	public <R> SingleOutputStreamOperator<R, ?> apply(WindowFunction<Iterable<T>, R, K, W> function, TypeInformation<R> resultType) {
+	public <R> SingleOutputStreamOperator<R, ?> apply(WindowFunction<T, R, K, W> function, TypeInformation<R> resultType) {
 
 		//clean the closure
 		function = input.getExecutionEnvironment().clean(function);
@@ -270,7 +269,7 @@ public WindowedStream<T, K, W> evictor(Evictor<? super T, ? super W> evictor) {
 					keySel,
 					input.getKeyType().createSerializer(getExecutionEnvironment().getConfig()),
 					stateDesc,
-					function,
+					new InternalIterableWindowFunction<>(function),
 					trigger,
 					evictor);
 
@@ -285,7 +284,7 @@ public WindowedStream<T, K, W> evictor(Evictor<? super T, ? super W> evictor) {
 					keySel,
 					input.getKeyType().createSerializer(getExecutionEnvironment().getConfig()),
 					stateDesc,
-					function,
+					new InternalIterableWindowFunction<>(function),
 					trigger);
 		}
 
@@ -350,13 +349,13 @@ public WindowedStream<T, K, W> evictor(Evictor<? super T, ? super W> evictor) {
 			opName = "TriggerWindow(" + windowAssigner + ", " + stateDesc + ", " + trigger + ", " + evictor + ", " + udfName + ")";
 
 			operator = new EvictingWindowOperator<>(windowAssigner,
-				windowAssigner.getWindowSerializer(getExecutionEnvironment().getConfig()),
-				keySel,
-				input.getKeyType().createSerializer(getExecutionEnvironment().getConfig()),
-				stateDesc,
-				new ReduceApplyWindowFunction<>(reduceFunction, function),
-				trigger,
-				evictor);
+					windowAssigner.getWindowSerializer(getExecutionEnvironment().getConfig()),
+					keySel,
+					input.getKeyType().createSerializer(getExecutionEnvironment().getConfig()),
+					stateDesc,
+					new InternalIterableWindowFunction<>(new ReduceApplyWindowFunction<>(reduceFunction, function)),
+					trigger,
+					evictor);
 
 		} else {
 			ReducingStateDescriptor<T> stateDesc = new ReducingStateDescriptor<>("window-contents",
@@ -370,7 +369,7 @@ public WindowedStream<T, K, W> evictor(Evictor<? super T, ? super W> evictor) {
 					keySel,
 					input.getKeyType().createSerializer(getExecutionEnvironment().getConfig()),
 					stateDesc,
-					function,
+					new InternalSingleValueWindowFunction<>(function),
 					trigger);
 		}
 
@@ -441,7 +440,7 @@ public WindowedStream<T, K, W> evictor(Evictor<? super T, ? super W> evictor) {
 				keySel,
 				input.getKeyType().createSerializer(getExecutionEnvironment().getConfig()),
 				stateDesc,
-				new FoldApplyWindowFunction<>(initialValue, foldFunction, function),
+				new InternalIterableWindowFunction<>(new FoldApplyWindowFunction<>(initialValue, foldFunction, function)),
 				trigger,
 				evictor);
 
@@ -458,7 +457,7 @@ public WindowedStream<T, K, W> evictor(Evictor<? super T, ? super W> evictor) {
 				keySel,
 				input.getKeyType().createSerializer(getExecutionEnvironment().getConfig()),
 				stateDesc,
-				function,
+				new InternalSingleValueWindowFunction<>(function),
 				trigger);
 		}
 
@@ -694,7 +693,7 @@ public WindowedStream<T, K, W> evictor(Evictor<? super T, ? super W> evictor) {
 			}
 			else if (function instanceof WindowFunction) {
 				@SuppressWarnings("unchecked")
-				WindowFunction<Iterable<T>, R, K, TimeWindow> wf = (WindowFunction<Iterable<T>, R, K, TimeWindow>) function;
+				WindowFunction<T, R, K, TimeWindow> wf = (WindowFunction<T, R, K, TimeWindow>) function;
 
 				OneInputStreamOperator<T, R> op = new AccumulatingProcessingTimeWindowOperator<>(
 						wf, input.getKeySelector(),
@@ -726,7 +725,7 @@ else if (function instanceof WindowFunction) {
 			}
 			else if (function instanceof WindowFunction) {
 				@SuppressWarnings("unchecked")
-				WindowFunction<Iterable<T>, R, K, TimeWindow> wf = (WindowFunction<Iterable<T>, R, K, TimeWindow>) function;
+				WindowFunction<T, R, K, TimeWindow> wf = (WindowFunction<T, R, K, TimeWindow>) function;
 
 				OneInputStreamOperator<T, R> op = new AccumulatingProcessingTimeWindowOperator<>(
 						wf, input.getKeySelector(),

flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/AllWindowFunction.java

@@ -30,9 +30,10 @@
  *
  * @param <IN> The type of the input value.
  * @param <OUT> The type of the output value.
+ * @param <W> The type of {@code Window} that this window function can be applied on.
  */
 @Public
-public interface AllWindowFunction<IN, OUT,  W extends Window> extends Function, Serializable {
+public interface AllWindowFunction<IN, OUT, W extends Window> extends Function, Serializable {
 
 	/**
 	 * Evaluates the window and outputs none or several elements.
@@ -43,5 +44,5 @@ public interface AllWindowFunction<IN, OUT,  W extends Window> extends Function,
 	 *
 	 * @throws Exception The function may throw exceptions to fail the program and trigger recovery.
 	 */
-	void apply(W window, IN values, Collector<OUT> out) throws Exception;
+	void apply(W window, Iterable<IN> values, Collector<OUT> out) throws Exception;
 }

flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/FoldApplyAllWindowFunction.java

@@ -33,11 +33,12 @@
 import java.io.ByteArrayInputStream;
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
+import java.util.Collections;
 
 @Internal
 public class FoldApplyAllWindowFunction<W extends Window, T, ACC>
 	extends WrappingFunction<AllWindowFunction<ACC, ACC, W>>
-	implements AllWindowFunction<Iterable<T>, ACC, W>, OutputTypeConfigurable<ACC> {
+	implements AllWindowFunction<T, ACC, W>, OutputTypeConfigurable<ACC> {
 
 	private static final long serialVersionUID = 1L;
 
@@ -75,7 +76,7 @@ public void apply(W window, Iterable<T> values, Collector<ACC> out) throws Excep
 			result = foldFunction.fold(result, val);
 		}
 
-		wrappedFunction.apply(window, result, out);
+		wrappedFunction.apply(window, Collections.singletonList(result), out);
 	}
 
 	@Override

flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/FoldApplyWindowFunction.java

@@ -33,11 +33,12 @@
 import java.io.ByteArrayInputStream;
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
+import java.util.Collections;
 
 @Internal
 public class FoldApplyWindowFunction<K, W extends Window, T, ACC>
 	extends WrappingFunction<WindowFunction<ACC, ACC, K, W>>
-	implements WindowFunction<Iterable<T>, ACC, K, W>, OutputTypeConfigurable<ACC> {
+	implements WindowFunction<T, ACC, K, W>, OutputTypeConfigurable<ACC> {
 
 	private static final long serialVersionUID = 1L;
 
@@ -75,7 +76,7 @@ public void apply(K key, W window, Iterable<T> values, Collector<ACC> out) throw
 			result = foldFunction.fold(result, val);
 		}
 
-		wrappedFunction.apply(key, window, result, out);
+		wrappedFunction.apply(key, window, Collections.singletonList(result), out);
 	}
 
 	@Override

flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/PassThroughAllWindowFunction.java

@@ -27,7 +27,9 @@ public class PassThroughAllWindowFunction<W extends Window, T> implements AllWin
 	private static final long serialVersionUID = 1L;
 
 	@Override
-	public void apply(W window, T input, Collector<T> out) throws Exception {
-		out.collect(input);
+	public void apply(W window, Iterable<T> input, Collector<T> out) throws Exception {
+		for (T in: input) {
+			out.collect(in);
+		}
 	}
 }

flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/PassThroughWindowFunction.java

@@ -27,7 +27,9 @@ public class PassThroughWindowFunction<K, W extends Window, T> implements Window
 	private static final long serialVersionUID = 1L;
 
 	@Override
-	public void apply(K k, W window, T input, Collector<T> out) throws Exception {
-		out.collect(input);
+	public void apply(K k, W window, Iterable<T> input, Collector<T> out) throws Exception {
+		for (T in: input) {
+			out.collect(in);
+		}
 	}
 }

flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/ReduceApplyAllWindowFunction.java

@@ -23,10 +23,12 @@
 import org.apache.flink.streaming.api.windowing.windows.Window;
 import org.apache.flink.util.Collector;
 
+import java.util.Collections;
+
 @Internal
 public class ReduceApplyAllWindowFunction<W extends Window, T, R>
 	extends WrappingFunction<AllWindowFunction<T, R, W>>
-	implements AllWindowFunction<Iterable<T>, R, W> {
+	implements AllWindowFunction<T, R, W> {
 
 	private static final long serialVersionUID = 1L;
 
@@ -51,6 +53,6 @@ public void apply(W window, Iterable<T> input, Collector<R> out) throws Exceptio
 				curr = reduceFunction.reduce(curr, val);
 			}
 		}
-		windowFunction.apply(window, curr, out);
+		windowFunction.apply(window, Collections.singletonList(curr), out);
 	}
 }

flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/ReduceApplyWindowFunction.java

@@ -23,10 +23,12 @@
 import org.apache.flink.streaming.api.windowing.windows.Window;
 import org.apache.flink.util.Collector;
 
+import java.util.Collections;
+
 @Internal
 public class ReduceApplyWindowFunction<K, W extends Window, T, R>
 	extends WrappingFunction<WindowFunction<T, R, K, W>>
-	implements WindowFunction<Iterable<T>, R, K, W> {
+	implements WindowFunction<T, R, K, W> {
 
 	private static final long serialVersionUID = 1L;
 
@@ -51,6 +53,6 @@ public void apply(K k, W window, Iterable<T> input, Collector<R> out) throws Exc
 				curr = reduceFunction.reduce(curr, val);
 			}
 		}
-		windowFunction.apply(k, window, curr, out);
+		windowFunction.apply(k, window, Collections.singletonList(curr), out);
 	}
 }

flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/ReduceIterableAllWindowFunction.java

@@ -23,7 +23,7 @@
 import org.apache.flink.util.Collector;
 
 @Internal
-public class ReduceIterableAllWindowFunction<W extends Window, T> implements AllWindowFunction<Iterable<T>, T, W> {
+public class ReduceIterableAllWindowFunction<W extends Window, T> implements AllWindowFunction<T, T, W> {
 	private static final long serialVersionUID = 1L;
 
 	private final ReduceFunction<T> reduceFunction;

flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/ReduceIterableWindowFunction.java

@@ -23,7 +23,7 @@
 import org.apache.flink.util.Collector;
 
 @Internal
-public class ReduceIterableWindowFunction<K, W extends Window, T> implements WindowFunction<Iterable<T>, T, K, W> {
+public class ReduceIterableWindowFunction<K, W extends Window, T> implements WindowFunction<T, T, K, W> {
 	private static final long serialVersionUID = 1L;
 
 	private final ReduceFunction<T> reduceFunction;

flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/WindowFunction.java

@@ -31,6 +31,7 @@
  * @param <IN> The type of the input value.
  * @param <OUT> The type of the output value.
  * @param <KEY> The type of the key.
+ * @param <W> The type of {@code Window} that this window function can be applied on.
  */
 @Public
 public interface WindowFunction<IN, OUT, KEY, W extends Window> extends Function, Serializable {
@@ -45,5 +46,5 @@ public interface WindowFunction<IN, OUT, KEY, W extends Window> extends Function
 	 * 
 	 * @throws Exception The function may throw exceptions to fail the program and trigger recovery. 
 	 */
-	void apply(KEY key, W window, IN input, Collector<OUT> out) throws Exception;
+	void apply(KEY key, W window, Iterable<IN> input, Collector<OUT> out) throws Exception;
 }

flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/AccumulatingKeyedTimePanes.java

@@ -36,15 +36,15 @@ public class AccumulatingKeyedTimePanes<Type, Key, Result> extends AbstractKeyed
 
 	private final KeyMap.LazyFactory<ArrayList<Type>> listFactory = getListFactory();
 
-	private final WindowFunction<Iterable<Type>, Result, Key, Window> function;
+	private final WindowFunction<Type, Result, Key, Window> function;
 
 	/**
 	 * IMPORTANT: This value needs to start at one, so it is fresher than the value that new entries have (zero) */
 	private long evaluationPass = 1L;   
 
 	// ------------------------------------------------------------------------
 
-	public AccumulatingKeyedTimePanes(KeySelector<Type, Key> keySelector, WindowFunction<Iterable<Type>, Result, Key, Window> function) {
+	public AccumulatingKeyedTimePanes(KeySelector<Type, Key> keySelector, WindowFunction<Type, Result, Key, Window> function) {
 		this.keySelector = keySelector;
 		this.function = function;
 	}
@@ -86,7 +86,7 @@ public void evaluateWindow(Collector<Result> out, TimeWindow window,
 
 	static final class WindowFunctionTraversal<Key, Type, Result> implements KeyMap.TraversalEvaluator<Key, ArrayList<Type>> {
 
-		private final WindowFunction<Iterable<Type>, Result, Key, Window> function;
+		private final WindowFunction<Type, Result, Key, Window> function;
 
 		private final UnionIterator<Type> unionIterator;
 
@@ -99,7 +99,7 @@ static final class WindowFunctionTraversal<Key, Type, Result> implements KeyMap.
 		private Key currentKey;
 
 
-		WindowFunctionTraversal(WindowFunction<Iterable<Type>, Result, Key, Window> function, TimeWindow window,
+		WindowFunctionTraversal(WindowFunction<Type, Result, Key, Window> function, TimeWindow window,
 								Collector<Result> out, AbstractStreamOperator<Result> contextOperator) {
 			this.function = function;
 			this.out = out;