package com.dataartisans.flink.dataflow.translation.functions;

import com.google.cloud.dataflow.sdk.values.KV;
import org.apache.flink.api.common.functions.GroupReduceFunction;
import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.List;

/**
 * Flink {@link org.apache.flink.api.common.functions.GroupReduceFunction} for executing a
 * {@link com.google.cloud.dataflow.sdk.transforms.GroupByKey} operation. This reads the input
 * {@link com.google.cloud.dataflow.sdk.values.KV} elements, extracts the key and collects
 * the values in a {@code List}.
 */
public class FlinkKeyedListAggregationFunction<K,V> implements GroupReduceFunction<KV<K, V>, KV<K, Iterable<V>>> {

	@Override
	public void reduce(Iterable<KV<K, V>> values, Collector<KV<K, Iterable<V>>> out) throws Exception {
		K k = null;
		List<V> result = new ArrayList<V>();
		for (KV<K, V> kv : values) {
			k = kv.getKey();
			result.add(kv.getValue());
		}
		out.collect(KV.of(k, (Iterable<V>) result));
	}

}

@@ -240,16 +240,43 @@
 ayList;%0A
+import java.util.Iterator;%0A
 import j
@@ -843,193 +843,959 @@
 %7B%0A%09%09
-K k = null;%0A%09%09List%3CV%3E result = new ArrayList%3CV%3E();%0A%09%09for (KV%3CK, V%3E kv : values) %7B%0A%09%09%09k = kv.getKey();%0A%09%09%09result.add(kv.getValue());%0A%09%09%7D%0A%09%09out.collect(KV.of(k, (Iterable%3CV%3E) result)
+Iterator%3CKV%3CK, V%3E%3E it = values.iterator();%0A%09%09KV%3CK, V%3E first = it.next();%0A%09%09Iterable%3CV%3E passThrough = new PassThroughIterable%3C%3E(first, it);%0A%09%09out.collect(KV.of(first.getKey(), passThrough));%0A%09%7D%0A%0A%09private static class PassThroughIterable%3CK, V%3E implements Iterable%3CV%3E, Iterator%3CV%3E  %7B%0A%09%09private KV%3CK, V%3E first;%0A%09%09private Iterator%3CKV%3CK, V%3E%3E iterator;%0A%0A%09%09public PassThroughIterable(KV%3CK, V%3E first, Iterator%3CKV%3CK, V%3E%3E iterator) %7B%0A%09%09%09this.first = first;%0A%09%09%09this.iterator = iterator;%0A%09%09%7D%0A%0A%09%09@Override%0A%09%09public Iterator%3CV%3E iterator() %7B%0A%09%09%09return this;%0A%09%09%7D%0A%0A%09%09@Override%0A%09%09public boolean hasNext() %7B%0A%09%09%09return first != null %7C%7C iterator.hasNext();%0A%09%09%7D%0A%0A%09%09@Override%0A%09%09public V next() %7B%0A%09%09%09if (first != null) %7B%0A%09%09%09%09V result = first.getValue();%0A%09%09%09%09first = null;%0A%09%09%09%09return result;%0A%09%09%09%7D else %7B%0A%09%09%09%09return iterator.next().getValue();%0A%09%09%09%7D%0A%09%09%7D%0A%0A%09%09@Override%0A%09%09public void remove() %7B%0A%09%09%09throw new UnsupportedOperationException(%22Cannot remove elements from input.%22
 );%0A
+%09
 %09%7D%0A
+%09%7D
 %0A%7D%0A

Change Flink Grouping Reducer to not materialize values