You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by se...@apache.org on 2014/10/03 18:25:05 UTC
[09/23] git commit: [FLINK-1110] Add Collection-Based execution for
Reduce Operators
[FLINK-1110] Add Collection-Based execution for Reduce Operators
Also fix some bugs resulting from moving stuff between packages
Project: http://git-wip-us.apache.org/repos/asf/incubator-flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-flink/commit/fd3f5c27
Tree: http://git-wip-us.apache.org/repos/asf/incubator-flink/tree/fd3f5c27
Diff: http://git-wip-us.apache.org/repos/asf/incubator-flink/diff/fd3f5c27
Branch: refs/heads/master
Commit: fd3f5c279c2b91072d63c0fcf8a0ea935693b1a2
Parents: 114af5a
Author: Aljoscha Krettek <al...@gmail.com>
Authored: Mon Sep 22 17:47:19 2014 +0200
Committer: Stephan Ewen <se...@apache.org>
Committed: Fri Oct 3 16:22:33 2014 +0200
----------------------------------------------------------------------
.../operators/base/GroupReduceOperatorBase.java | 59 +++++-
.../operators/base/ReduceOperatorBase.java | 71 ++++++-
.../operators/util/ListKeyGroupedIterator.java | 190 +++++++++++++++++++
.../common/operators/util/TypeComparable.java | 49 +++++
.../operators/base/GroupReduceOperatorTest.java | 171 +++++++++++++++++
.../operators/base/ReduceOperatorTest.java | 148 +++++++++++++++
.../SemanticPropertiesTranslationTest.scala | 4 +-
.../translation/AggregateTranslationTest.scala | 3 +-
.../DeltaIterationTranslationTest.scala | 2 +-
.../translation/ReduceTranslationTest.scala | 3 +-
10 files changed, 692 insertions(+), 8 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/fd3f5c27/flink-core/src/main/java/org/apache/flink/api/common/operators/base/GroupReduceOperatorBase.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/operators/base/GroupReduceOperatorBase.java b/flink-core/src/main/java/org/apache/flink/api/common/operators/base/GroupReduceOperatorBase.java
index d039f18..b309706 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/operators/base/GroupReduceOperatorBase.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/operators/base/GroupReduceOperatorBase.java
@@ -19,14 +19,26 @@
package org.apache.flink.api.common.operators.base;
+import org.apache.flink.api.common.InvalidProgramException;
import org.apache.flink.api.common.functions.FlatCombineFunction;
import org.apache.flink.api.common.functions.GroupReduceFunction;
+import org.apache.flink.api.common.functions.RuntimeContext;
+import org.apache.flink.api.common.functions.util.FunctionUtils;
+import org.apache.flink.api.common.functions.util.ListCollector;
import org.apache.flink.api.common.operators.Ordering;
import org.apache.flink.api.common.operators.SingleInputOperator;
import org.apache.flink.api.common.operators.UnaryOperatorInformation;
+import org.apache.flink.api.common.operators.util.ListKeyGroupedIterator;
import org.apache.flink.api.common.operators.util.UserCodeClassWrapper;
import org.apache.flink.api.common.operators.util.UserCodeObjectWrapper;
import org.apache.flink.api.common.operators.util.UserCodeWrapper;
+import org.apache.flink.api.common.typeinfo.CompositeType;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeutils.TypeComparator;
+
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.List;
/**
* @see org.apache.flink.api.common.functions.GroupReduceFunction
@@ -56,7 +68,7 @@ public class GroupReduceOperatorBase<IN, OUT, FT extends GroupReduceFunction<IN,
public GroupReduceOperatorBase(UserCodeWrapper<FT> udf, UnaryOperatorInformation<IN, OUT> operatorInfo, String name) {
super(udf, operatorInfo, name);
}
-
+
public GroupReduceOperatorBase(FT udf, UnaryOperatorInformation<IN, OUT> operatorInfo, String name) {
super(new UserCodeObjectWrapper<FT>(udf), operatorInfo, name);
}
@@ -115,4 +127,49 @@ public class GroupReduceOperatorBase<IN, OUT, FT extends GroupReduceFunction<IN,
return this.combinable;
}
+ // --------------------------------------------------------------------------------------------
+
+ @Override
+ protected List<OUT> executeOnCollections(List<IN> inputData, RuntimeContext ctx)
+ throws Exception {
+ GroupReduceFunction<IN, OUT> function = this.userFunction.getUserCodeObject();
+
+ UnaryOperatorInformation<IN, OUT> operatorInfo = getOperatorInfo();
+ TypeInformation<IN> inputType = operatorInfo.getInputType();
+
+ if (!(inputType instanceof CompositeType)) {
+ throw new InvalidProgramException("Input type of groupReduce operation must be" +
+ " composite type.");
+ }
+
+ int[] inputColumns = getKeyColumns(0);
+ boolean[] inputOrderings = new boolean[inputColumns.length];
+ final TypeComparator<IN> inputComparator =
+ ((CompositeType<IN>) inputType).createComparator(inputColumns, inputOrderings);
+
+ FunctionUtils.setFunctionRuntimeContext(function, ctx);
+ FunctionUtils.openFunction(function, this.parameters);
+
+
+ ArrayList<OUT> result = new ArrayList<OUT>(inputData.size());
+ ListCollector<OUT> collector = new ListCollector<OUT>(result);
+
+ inputData.sort( new Comparator<IN>() {
+ @Override
+ public int compare(IN o1, IN o2) {
+ return - inputComparator.compare(o1, o2);
+ }
+ });
+ ListKeyGroupedIterator<IN> keyedIterator =
+ new ListKeyGroupedIterator<IN>(inputData, inputComparator);
+
+ while (keyedIterator.nextKey()) {
+ function.reduce(keyedIterator.getValues(), collector);
+ }
+
+ FunctionUtils.closeFunction(function);
+
+ return result;
+ }
+
}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/fd3f5c27/flink-core/src/main/java/org/apache/flink/api/common/operators/base/ReduceOperatorBase.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/operators/base/ReduceOperatorBase.java b/flink-core/src/main/java/org/apache/flink/api/common/operators/base/ReduceOperatorBase.java
index 7c07a64..f7975a1 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/operators/base/ReduceOperatorBase.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/operators/base/ReduceOperatorBase.java
@@ -19,12 +19,24 @@
package org.apache.flink.api.common.operators.base;
+import org.apache.flink.api.common.InvalidProgramException;
import org.apache.flink.api.common.functions.ReduceFunction;
+import org.apache.flink.api.common.functions.RuntimeContext;
+import org.apache.flink.api.common.functions.util.FunctionUtils;
import org.apache.flink.api.common.operators.SingleInputOperator;
import org.apache.flink.api.common.operators.UnaryOperatorInformation;
+import org.apache.flink.api.common.operators.util.TypeComparable;
import org.apache.flink.api.common.operators.util.UserCodeClassWrapper;
import org.apache.flink.api.common.operators.util.UserCodeObjectWrapper;
import org.apache.flink.api.common.operators.util.UserCodeWrapper;
+import org.apache.flink.api.common.typeinfo.CompositeType;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeutils.TypeComparator;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
/**
@@ -88,7 +100,7 @@ public class ReduceOperatorBase<T, FT extends ReduceFunction<T>> extends SingleI
public ReduceOperatorBase(UserCodeWrapper<FT> udf, UnaryOperatorInformation<T, T> operatorInfo, String name) {
super(udf, operatorInfo, name);
}
-
+
/**
* Creates a non-grouped reduce data flow operator (all-reduce).
*
@@ -110,4 +122,61 @@ public class ReduceOperatorBase<T, FT extends ReduceFunction<T>> extends SingleI
public ReduceOperatorBase(Class<? extends FT> udf, UnaryOperatorInformation<T, T> operatorInfo, String name) {
super(new UserCodeClassWrapper<FT>(udf), operatorInfo, name);
}
+
+// --------------------------------------------------------------------------------------------
+
+ @SuppressWarnings("unchecked")
+ @Override
+ protected List<T> executeOnCollections(List<T> inputData, RuntimeContext ctx)
+ throws Exception {
+ ReduceFunction<T> function = this.userFunction.getUserCodeObject();
+
+ UnaryOperatorInformation<T, T> operatorInfo = getOperatorInfo();
+ TypeInformation<T> inputType = operatorInfo.getInputType();
+
+ if (!(inputType instanceof CompositeType)) {
+ throw new InvalidProgramException("Input type of groupReduce operation must be" +
+ " composite type.");
+ }
+
+ FunctionUtils.setFunctionRuntimeContext(function, ctx);
+ FunctionUtils.openFunction(function, this.parameters);
+
+ int[] inputColumns = getKeyColumns(0);
+ if (inputColumns.length > 0) {
+ boolean[] inputOrderings = new boolean[inputColumns.length];
+ TypeComparator<T> inputComparator = ((CompositeType<T>) inputType).createComparator(inputColumns, inputOrderings);
+
+ Map<TypeComparable<T>, T> aggregateMap = new HashMap<TypeComparable<T>, T>(inputData.size() / 10);
+
+ for (T next : inputData) {
+ TypeComparable<T> wrapper = new TypeComparable<T>(next, inputComparator);
+ T existing = aggregateMap.get(wrapper);
+ T result;
+ if (existing != null) {
+ result = function.reduce(existing, next);
+ } else {
+ result = next;
+ }
+ aggregateMap.put(wrapper, result);
+ }
+
+ List<T> result = new ArrayList<T>(aggregateMap.values().size());
+ result.addAll(aggregateMap.values());
+
+ FunctionUtils.closeFunction(function);
+ return result;
+ } else {
+ T aggregate = inputData.get(0);
+ for (int i = 1; i < inputData.size(); i++) {
+ aggregate = function.reduce(aggregate, inputData.get(i));
+ }
+ List<T> result = new ArrayList<T>(1);
+ result.add(aggregate);
+
+ FunctionUtils.setFunctionRuntimeContext(function, ctx);
+ return result;
+ }
+
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/fd3f5c27/flink-core/src/main/java/org/apache/flink/api/common/operators/util/ListKeyGroupedIterator.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/operators/util/ListKeyGroupedIterator.java b/flink-core/src/main/java/org/apache/flink/api/common/operators/util/ListKeyGroupedIterator.java
new file mode 100644
index 0000000..8fcd712
--- /dev/null
+++ b/flink-core/src/main/java/org/apache/flink/api/common/operators/util/ListKeyGroupedIterator.java
@@ -0,0 +1,190 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.flink.api.common.operators.util;
+
+import org.apache.flink.api.common.typeutils.TypeComparator;
+
+import java.io.IOException;
+import java.util.Iterator;
+import java.util.List;
+import java.util.NoSuchElementException;
+
+/**
+ * The KeyValueIterator returns a key and all values that belong to the key (share the same key).
+ *
+ */
+public final class ListKeyGroupedIterator<E> {
+
+ private final List<E> input;
+
+// private final TypeSerializer<E> serializer;
+
+ private final TypeComparator<E> comparator;
+
+ private ValuesIterator valuesIterator;
+
+ private int currentPosition = 0;
+
+ private E lookahead = null;
+
+ private boolean done;
+
+ /**
+ * Initializes the ListKeyGroupedIterator..
+ *
+ * @param input The list with the input elements.
+ * @param comparator The comparator for the data type iterated over.
+ */
+ public ListKeyGroupedIterator(List<E> input, TypeComparator<E> comparator)
+ {
+ if (input == null || comparator == null) {
+ throw new NullPointerException();
+ }
+
+ this.input = input;
+// this.serializer = serializer;
+ this.comparator = comparator;
+ }
+
+ /**
+ * Moves the iterator to the next key. This method may skip any values that have not yet been returned by the
+ * iterator created by the {@link #getValues()} method. Hence, if called multiple times it "removes" key groups.
+ *
+ * @return true, if the input iterator has an other group of records with the same key.
+ */
+ public boolean nextKey() throws IOException {
+
+ if (lookahead != null) {
+ // common case: whole value-iterator was consumed and a new key group is available.
+ this.comparator.setReference(this.lookahead);
+ this.valuesIterator.next = this.lookahead;
+ this.lookahead = null;
+ return true;
+ }
+
+ // first element, empty/done, or the values iterator was not entirely consumed
+ if (this.done) {
+ return false;
+ }
+
+ if (this.valuesIterator != null) {
+ // values was not entirely consumed. move to the next key
+ // Required if user code / reduce() method did not read the whole value iterator.
+ E next;
+ while (true) {
+ if ((next = this.input.get(currentPosition++)) != null) {
+ if (!this.comparator.equalToReference(next)) {
+ // the keys do not match, so we have a new group. store the current key
+ this.comparator.setReference(next);
+ this.valuesIterator.next = next;
+ return true;
+ }
+ }
+ else {
+ // input exhausted
+ this.valuesIterator.next = null;
+ this.valuesIterator = null;
+ this.done = true;
+ return false;
+ }
+ }
+ }
+ else {
+ // first element
+ // get the next element
+ E first = input.get(currentPosition++);
+ if (first != null) {
+ this.comparator.setReference(first);
+ this.valuesIterator = new ValuesIterator(first);
+ return true;
+ }
+ else {
+ // empty input, set everything null
+ this.done = true;
+ return false;
+ }
+ }
+ }
+
+ private E advanceToNext() {
+ if (currentPosition < input.size()) {
+ E next = input.get(currentPosition++);
+ if (comparator.equalToReference(next)) {
+ // same key
+ return next;
+ } else {
+ // moved to the next key, no more values here
+ lookahead = next;
+ return null;
+ }
+ }
+ else {
+ // backing iterator is consumed
+ this.done = true;
+ return null;
+ }
+ }
+
+ /**
+ * Returns an iterator over all values that belong to the current key. The iterator is initially <code>null</code>
+ * (before the first call to {@link #nextKey()} and after all keys are consumed. In general, this method returns
+ * always a non-null value, if a previous call to {@link #nextKey()} return <code>true</code>.
+ *
+ * @return Iterator over all values that belong to the current key.
+ */
+ public ValuesIterator getValues() {
+ return this.valuesIterator;
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ public final class ValuesIterator implements Iterator<E>, Iterable<E> {
+
+ private E next;
+
+ private ValuesIterator(E first) {
+ this.next = first;
+ }
+
+ @Override
+ public boolean hasNext() {
+ return next != null;
+ }
+
+ @Override
+ public E next() {
+ if (this.next != null) {
+ E current = this.next;
+ this.next = ListKeyGroupedIterator.this.advanceToNext();
+ return current;
+ } else {
+ throw new NoSuchElementException();
+ }
+ }
+
+ @Override
+ public void remove() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Iterator<E> iterator() {
+ return this;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/fd3f5c27/flink-core/src/main/java/org/apache/flink/api/common/operators/util/TypeComparable.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/operators/util/TypeComparable.java b/flink-core/src/main/java/org/apache/flink/api/common/operators/util/TypeComparable.java
new file mode 100644
index 0000000..2b6e5d2
--- /dev/null
+++ b/flink-core/src/main/java/org/apache/flink/api/common/operators/util/TypeComparable.java
@@ -0,0 +1,49 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.flink.api.common.operators.util;
+
+import org.apache.flink.api.common.typeutils.TypeComparator;
+
+/**
+ * Wrapper that is used to store elements for which we have a TypeComparator in a Hash Map.
+ */
+public class TypeComparable<T> {
+ private final T elem;
+ private final TypeComparator<T> comparator;
+ private final int hashCode;
+
+ public TypeComparable(T elem, TypeComparator<T> comparator) {
+ this.elem = elem;
+ this.comparator = comparator;
+ this.hashCode = comparator.hash(elem);
+ }
+
+ @Override
+ public int hashCode() {
+ return hashCode;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (!(o instanceof TypeComparable)) {
+ return false;
+ }
+ TypeComparable<T> other = (TypeComparable<T>) o;
+ return comparator.compare(elem, other.elem) == 0;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/fd3f5c27/flink-java/src/test/java/org/apache/flink/api/common/operators/base/GroupReduceOperatorTest.java
----------------------------------------------------------------------
diff --git a/flink-java/src/test/java/org/apache/flink/api/common/operators/base/GroupReduceOperatorTest.java b/flink-java/src/test/java/org/apache/flink/api/common/operators/base/GroupReduceOperatorTest.java
new file mode 100644
index 0000000..6c655ce
--- /dev/null
+++ b/flink-java/src/test/java/org/apache/flink/api/common/operators/base/GroupReduceOperatorTest.java
@@ -0,0 +1,171 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.flink.api.common.operators.base;
+
+import org.apache.flink.api.common.functions.GroupReduceFunction;
+import org.apache.flink.api.common.functions.ReduceFunction;
+import org.apache.flink.api.common.functions.RichGroupReduceFunction;
+import org.apache.flink.api.common.functions.RichReduceFunction;
+import org.apache.flink.api.common.functions.RuntimeContext;
+import org.apache.flink.api.common.functions.util.RuntimeUDFContext;
+import org.apache.flink.api.common.operators.UnaryOperatorInformation;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.typeutils.TypeInfoParser;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.util.Collector;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import static java.util.Arrays.asList;
+import static org.junit.Assert.*;
+
+@SuppressWarnings({"serial", "unchecked"})
+public class GroupReduceOperatorTest implements java.io.Serializable {
+
+ @Test
+ public void testGroupReduceCollection() {
+ try {
+ final GroupReduceFunction<Tuple2<String, Integer>, Tuple2<String,
+ Integer>> reducer = new GroupReduceFunction<Tuple2<String, Integer>,
+ Tuple2<String, Integer>>() {
+
+ @Override
+ public void reduce(Iterable<Tuple2<String, Integer>> values,
+ Collector<Tuple2<String, Integer>> out) throws Exception {
+ Iterator<Tuple2<String, Integer>> input = values.iterator();
+
+ Tuple2<String, Integer> result = input.next();
+ int sum = result.f1;
+ while (input.hasNext()) {
+ Tuple2<String, Integer> next = input.next();
+ sum += next.f1;
+ }
+ result.f1 = sum;
+ out.collect(result);
+ }
+ };
+
+ GroupReduceOperatorBase<Tuple2<String, Integer>, Tuple2<String, Integer>,
+ GroupReduceFunction<Tuple2<String, Integer>, Tuple2<String,
+ Integer>>> op = new GroupReduceOperatorBase<Tuple2<String, Integer>,
+ Tuple2<String, Integer>, GroupReduceFunction<Tuple2<String, Integer>,
+ Tuple2<String, Integer>>>(reducer, new UnaryOperatorInformation<Tuple2<String,
+ Integer>, Tuple2<String, Integer>>(TypeInfoParser.<Tuple2<String,
+ Integer>>parse("Tuple2<String, Integer>"), TypeInfoParser.<Tuple2<String,
+ Integer>>parse("Tuple2<String, Integer>")), new int[]{0}, "TestReducer");
+
+ List<Tuple2<String, Integer>> input = new ArrayList<Tuple2<String,
+ Integer>>(asList(new Tuple2<String, Integer>("foo", 1), new Tuple2<String,
+ Integer>("foo", 3), new Tuple2<String, Integer>("bar", 2), new Tuple2<String,
+ Integer>("bar", 4)));
+
+ List<Tuple2<String, Integer>> result = op.executeOnCollections(input, null);
+ Set<Tuple2<String, Integer>> resultSet = new HashSet<Tuple2<String, Integer>>(result);
+
+ Set<Tuple2<String, Integer>> expectedResult = new HashSet<Tuple2<String,
+ Integer>>(asList(new Tuple2<String, Integer>("foo", 4), new Tuple2<String,
+ Integer>("bar", 6)));
+
+ assertEquals(expectedResult, resultSet);
+ } catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void testGroupReduceCollectionWithRuntimeContext() {
+ try {
+ final String taskName = "Test Task";
+ final AtomicBoolean opened = new AtomicBoolean();
+ final AtomicBoolean closed = new AtomicBoolean();
+
+ final RichGroupReduceFunction<Tuple2<String, Integer>, Tuple2<String,
+ Integer>> reducer = new RichGroupReduceFunction<Tuple2<String, Integer>,
+ Tuple2<String, Integer>>() {
+
+
+ @Override
+ public void reduce(Iterable<Tuple2<String, Integer>> values,
+ Collector<Tuple2<String, Integer>> out) throws Exception {
+ Iterator<Tuple2<String, Integer>> input = values.iterator();
+
+ Tuple2<String, Integer> result = input.next();
+ int sum = result.f1;
+ while (input.hasNext()) {
+ Tuple2<String, Integer> next = input.next();
+ sum += next.f1;
+ }
+ result.f1 = sum;
+ out.collect(result);
+ }
+
+ @Override
+ public void open(Configuration parameters) throws Exception {
+ opened.set(true);
+ RuntimeContext ctx = getRuntimeContext();
+ assertEquals(0, ctx.getIndexOfThisSubtask());
+ assertEquals(1, ctx.getNumberOfParallelSubtasks());
+ assertEquals(taskName, ctx.getTaskName());
+ }
+
+ @Override
+ public void close() throws Exception {
+ closed.set(true);
+ }
+ };
+
+ GroupReduceOperatorBase<Tuple2<String, Integer>, Tuple2<String, Integer>,
+ GroupReduceFunction<Tuple2<String, Integer>, Tuple2<String,
+ Integer>>> op = new GroupReduceOperatorBase<Tuple2<String, Integer>,
+ Tuple2<String, Integer>, GroupReduceFunction<Tuple2<String, Integer>,
+ Tuple2<String, Integer>>>(reducer, new UnaryOperatorInformation<Tuple2<String,
+ Integer>, Tuple2<String, Integer>>(TypeInfoParser.<Tuple2<String,
+ Integer>>parse("Tuple2<String, Integer>"), TypeInfoParser.<Tuple2<String,
+ Integer>>parse("Tuple2<String, Integer>")), new int[]{0}, "TestReducer");
+
+ List<Tuple2<String, Integer>> input = new ArrayList<Tuple2<String,
+ Integer>>(asList(new Tuple2<String, Integer>("foo", 1), new Tuple2<String,
+ Integer>("foo", 3), new Tuple2<String, Integer>("bar", 2), new Tuple2<String,
+ Integer>("bar", 4)));
+
+ List<Tuple2<String, Integer>> result = op.executeOnCollections(input,
+ new RuntimeUDFContext(taskName, 1, 0));
+ Set<Tuple2<String, Integer>> resultSet = new HashSet<Tuple2<String, Integer>>(result);
+
+ Set<Tuple2<String, Integer>> expectedResult = new HashSet<Tuple2<String,
+ Integer>>(asList(new Tuple2<String, Integer>("foo", 4), new Tuple2<String,
+ Integer>("bar", 6)));
+
+ assertEquals(expectedResult, resultSet);
+
+ assertTrue(opened.get());
+ assertTrue(closed.get());
+ } catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/fd3f5c27/flink-java/src/test/java/org/apache/flink/api/common/operators/base/ReduceOperatorTest.java
----------------------------------------------------------------------
diff --git a/flink-java/src/test/java/org/apache/flink/api/common/operators/base/ReduceOperatorTest.java b/flink-java/src/test/java/org/apache/flink/api/common/operators/base/ReduceOperatorTest.java
new file mode 100644
index 0000000..a4de40b
--- /dev/null
+++ b/flink-java/src/test/java/org/apache/flink/api/common/operators/base/ReduceOperatorTest.java
@@ -0,0 +1,148 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.flink.api.common.operators.base;
+
+import org.apache.flink.api.common.functions.ReduceFunction;
+import org.apache.flink.api.common.functions.RichReduceFunction;
+import org.apache.flink.api.common.functions.RuntimeContext;
+import org.apache.flink.api.common.functions.util.RuntimeUDFContext;
+import org.apache.flink.api.common.operators.UnaryOperatorInformation;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.typeutils.TypeInfoParser;
+import org.apache.flink.configuration.Configuration;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import static java.util.Arrays.asList;
+import static org.junit.Assert.*;
+
+@SuppressWarnings({"serial", "unchecked"})
+public class ReduceOperatorTest implements java.io.Serializable {
+
+ @Test
+ public void testReduceCollection() {
+ try {
+ final ReduceFunction<Tuple2<String, Integer>> reducer = new
+ ReduceFunction<Tuple2<String, Integer>>() {
+
+ @Override
+ public Tuple2<String, Integer> reduce(Tuple2<String, Integer> value1,
+ Tuple2<String, Integer> value2) throws
+ Exception {
+ return new Tuple2<String, Integer>(value1.f0, value1.f1 + value2.f1);
+ }
+ };
+
+ ReduceOperatorBase<Tuple2<String, Integer>, ReduceFunction<Tuple2<String,
+ Integer>>> op = new ReduceOperatorBase<Tuple2<String, Integer>,
+ ReduceFunction<Tuple2<String, Integer>>>(reducer,
+ new UnaryOperatorInformation<Tuple2<String, Integer>, Tuple2<String,
+ Integer>>(TypeInfoParser.<Tuple2<String,
+ Integer>>parse("Tuple2<String, Integer>"),
+ TypeInfoParser.<Tuple2<String, Integer>>parse("Tuple2<String, " +
+ "Integer>")), new int[]{0}, "TestReducer");
+
+ List<Tuple2<String, Integer>> input = new ArrayList<Tuple2<String,
+ Integer>>(asList(new Tuple2<String, Integer>("foo", 1), new Tuple2<String,
+ Integer>("foo", 3), new Tuple2<String, Integer>("bar", 2), new Tuple2<String,
+ Integer>("bar", 4)));
+
+ List<Tuple2<String, Integer>> result = op.executeOnCollections(input, null);
+ Set<Tuple2<String, Integer>> resultSet = new HashSet<Tuple2<String, Integer>>(result);
+
+ Set<Tuple2<String, Integer>> expectedResult = new HashSet<Tuple2<String,
+ Integer>>(asList(new Tuple2<String, Integer>("foo", 4), new Tuple2<String,
+ Integer>("bar", 6)));
+
+ assertEquals(expectedResult, resultSet);
+ } catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void testReduceCollectionWithRuntimeContext() {
+ try {
+ final String taskName = "Test Task";
+ final AtomicBoolean opened = new AtomicBoolean();
+ final AtomicBoolean closed = new AtomicBoolean();
+
+ final ReduceFunction<Tuple2<String, Integer>> reducer = new
+ RichReduceFunction<Tuple2<String, Integer>>() {
+ @Override
+ public Tuple2<String, Integer> reduce(Tuple2<String, Integer> value1,
+ Tuple2<String, Integer> value2) throws
+ Exception {
+ return new Tuple2<String, Integer>(value1.f0, value1.f1 + value2.f1);
+ }
+
+ @Override
+ public void open(Configuration parameters) throws Exception {
+ opened.set(true);
+ RuntimeContext ctx = getRuntimeContext();
+ assertEquals(0, ctx.getIndexOfThisSubtask());
+ assertEquals(1, ctx.getNumberOfParallelSubtasks());
+ assertEquals(taskName, ctx.getTaskName());
+ }
+
+ @Override
+ public void close() throws Exception {
+ closed.set(true);
+ }
+ };
+
+ ReduceOperatorBase<Tuple2<String, Integer>, ReduceFunction<Tuple2<String,
+ Integer>>> op = new ReduceOperatorBase<Tuple2<String, Integer>,
+ ReduceFunction<Tuple2<String, Integer>>>(reducer,
+ new UnaryOperatorInformation<Tuple2<String, Integer>, Tuple2<String,
+ Integer>>(TypeInfoParser.<Tuple2<String,
+ Integer>>parse("Tuple2<String, Integer>"),
+ TypeInfoParser.<Tuple2<String, Integer>>parse("Tuple2<String, " +
+ "Integer>")), new int[]{0}, "TestReducer");
+
+ List<Tuple2<String, Integer>> input = new ArrayList<Tuple2<String,
+ Integer>>(asList(new Tuple2<String, Integer>("foo", 1), new Tuple2<String,
+ Integer>("foo", 3), new Tuple2<String, Integer>("bar", 2), new Tuple2<String,
+ Integer>("bar", 4)));
+
+ List<Tuple2<String, Integer>> result = op.executeOnCollections(input,
+ new RuntimeUDFContext(taskName, 1, 0));
+
+ Set<Tuple2<String, Integer>> resultSet = new HashSet<Tuple2<String, Integer>>(result);
+
+ Set<Tuple2<String, Integer>> expectedResult = new HashSet<Tuple2<String,
+ Integer>>(asList(new Tuple2<String, Integer>("foo", 4), new Tuple2<String,
+ Integer>("bar", 6)));
+
+ assertEquals(expectedResult, resultSet);
+
+ assertTrue(opened.get());
+ assertTrue(closed.get());
+ } catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/fd3f5c27/flink-scala/src/test/scala/org/apache/flink/api/scala/functions/SemanticPropertiesTranslationTest.scala
----------------------------------------------------------------------
diff --git a/flink-scala/src/test/scala/org/apache/flink/api/scala/functions/SemanticPropertiesTranslationTest.scala b/flink-scala/src/test/scala/org/apache/flink/api/scala/functions/SemanticPropertiesTranslationTest.scala
index e6223b6..543c05a 100644
--- a/flink-scala/src/test/scala/org/apache/flink/api/scala/functions/SemanticPropertiesTranslationTest.scala
+++ b/flink-scala/src/test/scala/org/apache/flink/api/scala/functions/SemanticPropertiesTranslationTest.scala
@@ -20,8 +20,8 @@ package org.apache.flink.api.scala.functions
import org.junit.Assert._
import org.apache.flink.api.common.functions.RichJoinFunction
import org.apache.flink.api.common.functions.RichMapFunction
-import org.apache.flink.api.common.operators.SingleInputSemanticProperties
-import org.apache.flink.api.common.operators.base.{JoinOperatorBase, GenericDataSinkBase, MapOperatorBase}
+import org.apache.flink.api.common.operators.{GenericDataSinkBase, SingleInputSemanticProperties}
+import org.apache.flink.api.common.operators.base.{JoinOperatorBase, MapOperatorBase}
import org.apache.flink.api.common.operators.util.FieldSet
import org.apache.flink.api.java.functions.FunctionAnnotation.ConstantFields
import org.apache.flink.api.java.functions.FunctionAnnotation.ConstantFieldsFirst
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/fd3f5c27/flink-scala/src/test/scala/org/apache/flink/api/scala/operators/translation/AggregateTranslationTest.scala
----------------------------------------------------------------------
diff --git a/flink-scala/src/test/scala/org/apache/flink/api/scala/operators/translation/AggregateTranslationTest.scala b/flink-scala/src/test/scala/org/apache/flink/api/scala/operators/translation/AggregateTranslationTest.scala
index 22a137c..c4d7dc8 100644
--- a/flink-scala/src/test/scala/org/apache/flink/api/scala/operators/translation/AggregateTranslationTest.scala
+++ b/flink-scala/src/test/scala/org/apache/flink/api/scala/operators/translation/AggregateTranslationTest.scala
@@ -18,7 +18,8 @@
package org.apache.flink.api.scala.operators.translation
import org.apache.flink.api.common.Plan
-import org.apache.flink.api.common.operators.base.{GenericDataSourceBase, GroupReduceOperatorBase}
+import org.apache.flink.api.common.operators.GenericDataSourceBase
+import org.apache.flink.api.common.operators.base.GroupReduceOperatorBase
import org.apache.flink.api.java.aggregation.Aggregations
import org.apache.flink.api.scala._
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/fd3f5c27/flink-scala/src/test/scala/org/apache/flink/api/scala/operators/translation/DeltaIterationTranslationTest.scala
----------------------------------------------------------------------
diff --git a/flink-scala/src/test/scala/org/apache/flink/api/scala/operators/translation/DeltaIterationTranslationTest.scala b/flink-scala/src/test/scala/org/apache/flink/api/scala/operators/translation/DeltaIterationTranslationTest.scala
index 9117c03..4a60154 100644
--- a/flink-scala/src/test/scala/org/apache/flink/api/scala/operators/translation/DeltaIterationTranslationTest.scala
+++ b/flink-scala/src/test/scala/org/apache/flink/api/scala/operators/translation/DeltaIterationTranslationTest.scala
@@ -19,6 +19,7 @@ package org.apache.flink.api.scala.operators.translation
import org.apache.flink.api.common.functions.{RichCoGroupFunction, RichMapFunction,
RichJoinFunction}
+import org.apache.flink.api.common.operators.GenericDataSinkBase
import org.apache.flink.api.java.operators.translation.WrappingFunction
import org.junit.Assert.assertArrayEquals
import org.junit.Assert.assertEquals
@@ -26,7 +27,6 @@ import org.junit.Assert.fail
import org.apache.flink.api.common.{InvalidProgramException, Plan}
import org.apache.flink.api.common.aggregators.LongSumAggregator
import org.apache.flink.api.common.operators.base.DeltaIterationBase
-import org.apache.flink.api.common.operators.base.GenericDataSinkBase
import org.apache.flink.api.common.operators.base.JoinOperatorBase
import org.apache.flink.api.common.operators.base.MapOperatorBase
import org.junit.Test
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/fd3f5c27/flink-scala/src/test/scala/org/apache/flink/api/scala/operators/translation/ReduceTranslationTest.scala
----------------------------------------------------------------------
diff --git a/flink-scala/src/test/scala/org/apache/flink/api/scala/operators/translation/ReduceTranslationTest.scala b/flink-scala/src/test/scala/org/apache/flink/api/scala/operators/translation/ReduceTranslationTest.scala
index fc63431..22e4914 100644
--- a/flink-scala/src/test/scala/org/apache/flink/api/scala/operators/translation/ReduceTranslationTest.scala
+++ b/flink-scala/src/test/scala/org/apache/flink/api/scala/operators/translation/ReduceTranslationTest.scala
@@ -17,13 +17,12 @@
*/
package org.apache.flink.api.scala.operators.translation
+import org.apache.flink.api.common.operators.{GenericDataSourceBase, GenericDataSinkBase}
import org.apache.flink.api.java.operators.translation.{KeyExtractingMapper,
PlanUnwrappingReduceOperator}
import org.apache.flink.api.common.typeinfo.BasicTypeInfo
import org.apache.flink.api.java.typeutils.TupleTypeInfo
import org.junit.Assert._
-import org.apache.flink.api.common.operators.base.GenericDataSinkBase
-import org.apache.flink.api.common.operators.base.GenericDataSourceBase
import org.apache.flink.api.common.operators.base.MapOperatorBase
import org.apache.flink.api.common.operators.base.ReduceOperatorBase
import org.junit.Test