/*
 * 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.test.operators;

import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.functions.GroupCombineFunction;
import org.apache.flink.api.common.functions.GroupReduceFunction;
import org.apache.flink.api.common.functions.RichGroupReduceFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.operators.Order;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.java.tuple.Tuple5;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.optimizer.Optimizer;
import org.apache.flink.test.operators.util.CollectionDataSets;
import org.apache.flink.test.operators.util.CollectionDataSets.CrazyNested;
import org.apache.flink.test.operators.util.CollectionDataSets.CustomType;
import org.apache.flink.test.operators.util.CollectionDataSets.FromTupleWithCTor;
import org.apache.flink.test.operators.util.CollectionDataSets.POJO;
import org.apache.flink.test.operators.util.CollectionDataSets.PojoContainingTupleAndWritable;
import org.apache.flink.test.util.MultipleProgramsTestBase;
import org.apache.flink.util.Collector;

import org.joda.time.DateTime;
import org.junit.Assert;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;

import java.util.Collection;
import java.util.Date;
import java.util.Iterator;
import java.util.List;

import scala.math.BigInt;

/**
 * Integration tests for {@link GroupReduceFunction}, {@link RichGroupReduceFunction}, and {@link
 * GroupCombineFunction}.
 */
@SuppressWarnings({"serial", "unchecked", "UnusedDeclaration"})
@RunWith(Parameterized.class)
public class GroupReduceITCase extends MultipleProgramsTestBase {

    public GroupReduceITCase(TestExecutionMode mode) {
        super(mode);
    }

    @Test
    public void
            testCorrectnessofGroupReduceOnTupleContainingPrimitiveByteArrayWithKeyFieldSelectors()
                    throws Exception {
        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        DataSet<Tuple2<byte[], Integer>> ds = CollectionDataSets.getTuple2WithByteArrayDataSet(env);
        DataSet<Integer> reduceDs = ds.groupBy(0).reduceGroup(new ByteArrayGroupReduce());

        List<Integer> result = reduceDs.collect();

        String expected = "0\n" + "1\n" + "2\n" + "3\n" + "4\n";

        compareResultAsText(result, expected);
    }

    private static class ByteArrayGroupReduce
            implements GroupReduceFunction<Tuple2<byte[], Integer>, Integer> {
        @Override
        public void reduce(Iterable<Tuple2<byte[], Integer>> values, Collector<Integer> out)
                throws Exception {
            int sum = 0;
            for (Tuple2<byte[], Integer> value : values) {
                sum += value.f1;
            }
            out.collect(sum);
        }
    }

    @Test
    public void testCorrectnessOfGroupReduceOnTuplesWithKeyFieldSelector() throws Exception {
        /*
         * check correctness of groupReduce on tuples with key field selector
         */

        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        DataSet<Tuple3<Integer, Long, String>> ds = CollectionDataSets.get3TupleDataSet(env);
        DataSet<Tuple2<Integer, Long>> reduceDs =
                ds.groupBy(1).reduceGroup(new Tuple3GroupReduce());

        List<Tuple2<Integer, Long>> result = reduceDs.collect();

        String expected = "1,1\n" + "5,2\n" + "15,3\n" + "34,4\n" + "65,5\n" + "111,6\n";

        compareResultAsTuples(result, expected);
    }

    @Test
    public void testCorrectnessOfGroupReduceOnTuplesWithMultipleKeyFieldSelectors()
            throws Exception {
        /*
         * check correctness of groupReduce on tuples with multiple key field selector
         */

        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        DataSet<Tuple5<Integer, Long, Integer, String, Long>> ds =
                CollectionDataSets.get5TupleDataSet(env);
        DataSet<Tuple5<Integer, Long, Integer, String, Long>> reduceDs =
                ds.groupBy(4, 0).reduceGroup(new Tuple5GroupReduce());

        List<Tuple5<Integer, Long, Integer, String, Long>> result = reduceDs.collect();

        String expected =
                "1,1,0,P-),1\n"
                        + "2,3,0,P-),1\n"
                        + "2,2,0,P-),2\n"
                        + "3,9,0,P-),2\n"
                        + "3,6,0,P-),3\n"
                        + "4,17,0,P-),1\n"
                        + "4,17,0,P-),2\n"
                        + "5,11,0,P-),1\n"
                        + "5,29,0,P-),2\n"
                        + "5,25,0,P-),3\n";

        compareResultAsTuples(result, expected);
    }

    @Test
    public void testCorrectnessOfGroupReduceOnTuplesWithKeyFieldSelectorAndGroupSorting()
            throws Exception {
        /*
         * check correctness of groupReduce on tuples with key field selector and group sorting
         */

        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataSet<Tuple3<Integer, Long, String>> ds = CollectionDataSets.get3TupleDataSet(env);
        DataSet<Tuple3<Integer, Long, String>> reduceDs =
                ds.groupBy(1)
                        .sortGroup(2, Order.ASCENDING)
                        .reduceGroup(new Tuple3SortedGroupReduce());

        List<Tuple3<Integer, Long, String>> result = reduceDs.collect();

        String expected =
                "1,1,Hi\n"
                        + "5,2,Hello-Hello world\n"
                        + "15,3,Hello world, how are you?-I am fine.-Luke Skywalker\n"
                        + "34,4,Comment#1-Comment#2-Comment#3-Comment#4\n"
                        + "65,5,Comment#5-Comment#6-Comment#7-Comment#8-Comment#9\n"
                        + "111,6,Comment#10-Comment#11-Comment#12-Comment#13-Comment#14-Comment#15\n";

        compareResultAsTuples(result, expected);
    }

    @Test
    public void testCorrectnessOfGroupReduceOnTuplesWithKeyExtractor() throws Exception {
        /*
         * check correctness of groupReduce on tuples with key extractor
         */

        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        DataSet<Tuple3<Integer, Long, String>> ds = CollectionDataSets.get3TupleDataSet(env);
        DataSet<Tuple2<Integer, Long>> reduceDs =
                ds.groupBy(new KeySelector1()).reduceGroup(new Tuple3GroupReduce());

        List<Tuple2<Integer, Long>> result = reduceDs.collect();

        String expected = "1,1\n" + "5,2\n" + "15,3\n" + "34,4\n" + "65,5\n" + "111,6\n";

        compareResultAsTuples(result, expected);
    }

    private static class KeySelector1 implements KeySelector<Tuple3<Integer, Long, String>, Long> {
        private static final long serialVersionUID = 1L;

        @Override
        public Long getKey(Tuple3<Integer, Long, String> in) {
            return in.f1;
        }
    }

    @Test
    public void testCorrectnessOfGroupReduceOnCustomTypeWithTypeExtractor() throws Exception {
        /*
         * check correctness of groupReduce on custom type with type extractor
         */

        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        DataSet<CustomType> ds = CollectionDataSets.getCustomTypeDataSet(env);
        DataSet<CustomType> reduceDs =
                ds.groupBy(new KeySelector2()).reduceGroup(new CustomTypeGroupReduce());

        List<CustomType> result = reduceDs.collect();

        String expected =
                "1,0,Hello!\n"
                        + "2,3,Hello!\n"
                        + "3,12,Hello!\n"
                        + "4,30,Hello!\n"
                        + "5,60,Hello!\n"
                        + "6,105,Hello!\n";

        compareResultAsText(result, expected);
    }

    private static class KeySelector2 implements KeySelector<CustomType, Integer> {
        private static final long serialVersionUID = 1L;

        @Override
        public Integer getKey(CustomType in) {
            return in.myInt;
        }
    }

    @Test
    public void testCorrectnessOfAllGroupReduceForTuples() throws Exception {
        /*
         * check correctness of all-groupreduce for tuples
         */

        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        DataSet<Tuple3<Integer, Long, String>> ds = CollectionDataSets.get3TupleDataSet(env);
        DataSet<Tuple3<Integer, Long, String>> reduceDs =
                ds.reduceGroup(new AllAddingTuple3GroupReduce());

        List<Tuple3<Integer, Long, String>> result = reduceDs.collect();

        String expected = "231,91,Hello World\n";

        compareResultAsTuples(result, expected);
    }

    @Test
    public void testCorrectnessOfAllGroupReduceForCustomTypes() throws Exception {
        /*
         * check correctness of all-groupreduce for custom types
         */

        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        DataSet<CustomType> ds = CollectionDataSets.getCustomTypeDataSet(env);
        DataSet<CustomType> reduceDs = ds.reduceGroup(new AllAddingCustomTypeGroupReduce());

        List<CustomType> result = reduceDs.collect();

        String expected = "91,210,Hello!";

        compareResultAsText(result, expected);
    }

    @Test
    public void testCorrectnessOfGroupReduceWithBroadcastSet() throws Exception {
        /*
         * check correctness of groupReduce with broadcast set
         */

        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        DataSet<Integer> intDs = CollectionDataSets.getIntegerDataSet(env);

        DataSet<Tuple3<Integer, Long, String>> ds = CollectionDataSets.get3TupleDataSet(env);
        DataSet<Tuple3<Integer, Long, String>> reduceDs =
                ds.groupBy(1)
                        .reduceGroup(new BCTuple3GroupReduce())
                        .withBroadcastSet(intDs, "ints");

        List<Tuple3<Integer, Long, String>> result = reduceDs.collect();

        String expected =
                "1,1,55\n" + "5,2,55\n" + "15,3,55\n" + "34,4,55\n" + "65,5,55\n" + "111,6,55\n";

        compareResultAsTuples(result, expected);
    }

    @Test
    public void testCorrectnessOfGroupReduceIfUDFReturnsInputObjectsMultipleTimesWhileChangingThem()
            throws Exception {
        /*
         * check correctness of groupReduce if UDF returns input objects multiple times and changes it in between
         */

        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        DataSet<Tuple3<Integer, Long, String>> ds = CollectionDataSets.get3TupleDataSet(env);
        DataSet<Tuple3<Integer, Long, String>> reduceDs =
                ds.groupBy(1).reduceGroup(new InputReturningTuple3GroupReduce());

        List<Tuple3<Integer, Long, String>> result = reduceDs.collect();

        String expected =
                "11,1,Hi!\n"
                        + "21,1,Hi again!\n"
                        + "12,2,Hi!\n"
                        + "22,2,Hi again!\n"
                        + "13,2,Hi!\n"
                        + "23,2,Hi again!\n";

        compareResultAsTuples(result, expected);
    }

    @Test
    public void testCorrectnessOfGroupReduceOnCustomTypeWithKeyExtractorAndCombine()
            throws Exception {
        /*
         * check correctness of groupReduce on custom type with key extractor and combine
         */
        org.junit.Assume.assumeTrue(mode != TestExecutionMode.COLLECTION);

        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        DataSet<CustomType> ds = CollectionDataSets.getCustomTypeDataSet(env);
        DataSet<CustomType> reduceDs =
                ds.groupBy(new KeySelector3()).reduceGroup(new CustomTypeGroupReduceWithCombine());

        List<CustomType> result = reduceDs.collect();

        String expected =
                "1,0,test1\n"
                        + "2,3,test2\n"
                        + "3,12,test3\n"
                        + "4,30,test4\n"
                        + "5,60,test5\n"
                        + "6,105,test6\n";

        compareResultAsText(result, expected);
    }

    private static class KeySelector3 implements KeySelector<CustomType, Integer> {
        private static final long serialVersionUID = 1L;

        @Override
        public Integer getKey(CustomType in) {
            return in.myInt;
        }
    }

    @Test
    public void testCorrectnessOfGroupReduceOnTuplesWithCombine() throws Exception {
        /*
         * check correctness of groupReduce on tuples with combine
         */
        org.junit.Assume.assumeTrue(mode != TestExecutionMode.COLLECTION);

        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2); // important because it determines how often the combiner is called

        DataSet<Tuple3<Integer, Long, String>> ds = CollectionDataSets.get3TupleDataSet(env);
        DataSet<Tuple2<Integer, String>> reduceDs =
                ds.groupBy(1).reduceGroup(new Tuple3GroupReduceWithCombine());

        List<Tuple2<Integer, String>> result = reduceDs.collect();

        String expected =
                "1,test1\n"
                        + "5,test2\n"
                        + "15,test3\n"
                        + "34,test4\n"
                        + "65,test5\n"
                        + "111,test6\n";

        compareResultAsTuples(result, expected);
    }

    @Test
    public void testCorrectnessOfAllGroupReduceForTuplesWithCombine() throws Exception {
        /*
         * check correctness of all-groupreduce for tuples with combine
         */
        org.junit.Assume.assumeTrue(mode != TestExecutionMode.COLLECTION);

        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        DataSet<Tuple3<Integer, Long, String>> ds =
                CollectionDataSets.get3TupleDataSet(env)
                        .map(new IdentityMapper<Tuple3<Integer, Long, String>>())
                        .setParallelism(4);

        Configuration cfg = new Configuration();
        cfg.setString(Optimizer.HINT_SHIP_STRATEGY, Optimizer.HINT_SHIP_STRATEGY_REPARTITION);
        DataSet<Tuple2<Integer, String>> reduceDs =
                ds.reduceGroup(new Tuple3AllGroupReduceWithCombine()).withParameters(cfg);

        List<Tuple2<Integer, String>> result = reduceDs.collect();

        String expected =
                "322,"
                        + "testtesttesttesttesttesttesttesttesttesttesttesttesttesttesttesttesttesttesttesttest\n";

        compareResultAsTuples(result, expected);
    }

    @Test
    public void testCorrectnessOfGroupreduceWithDescendingGroupSort() throws Exception {
        /*
         * check correctness of groupReduce with descending group sort
         */
        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataSet<Tuple3<Integer, Long, String>> ds = CollectionDataSets.get3TupleDataSet(env);
        DataSet<Tuple3<Integer, Long, String>> reduceDs =
                ds.groupBy(1)
                        .sortGroup(2, Order.DESCENDING)
                        .reduceGroup(new Tuple3SortedGroupReduce());

        List<Tuple3<Integer, Long, String>> result = reduceDs.collect();

        String expected =
                "1,1,Hi\n"
                        + "5,2,Hello world-Hello\n"
                        + "15,3,Luke Skywalker-I am fine.-Hello world, how are you?\n"
                        + "34,4,Comment#4-Comment#3-Comment#2-Comment#1\n"
                        + "65,5,Comment#9-Comment#8-Comment#7-Comment#6-Comment#5\n"
                        + "111,6,Comment#15-Comment#14-Comment#13-Comment#12-Comment#11-Comment#10\n";

        compareResultAsTuples(result, expected);
    }

    @Test
    public void testCorrectnessOfGroupReduceOnTuplesWithTupleReturningKeySelector()
            throws Exception {
        /*
         * check correctness of groupReduce on tuples with tuple-returning key selector
         */

        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        DataSet<Tuple5<Integer, Long, Integer, String, Long>> ds =
                CollectionDataSets.get5TupleDataSet(env);
        DataSet<Tuple5<Integer, Long, Integer, String, Long>> reduceDs =
                ds.groupBy(new KeySelector4()).reduceGroup(new Tuple5GroupReduce());

        List<Tuple5<Integer, Long, Integer, String, Long>> result = reduceDs.collect();

        String expected =
                "1,1,0,P-),1\n"
                        + "2,3,0,P-),1\n"
                        + "2,2,0,P-),2\n"
                        + "3,9,0,P-),2\n"
                        + "3,6,0,P-),3\n"
                        + "4,17,0,P-),1\n"
                        + "4,17,0,P-),2\n"
                        + "5,11,0,P-),1\n"
                        + "5,29,0,P-),2\n"
                        + "5,25,0,P-),3\n";

        compareResultAsTuples(result, expected);
    }

    private static class KeySelector4
            implements KeySelector<
                    Tuple5<Integer, Long, Integer, String, Long>, Tuple2<Integer, Long>> {
        private static final long serialVersionUID = 1L;

        @Override
        public Tuple2<Integer, Long> getKey(Tuple5<Integer, Long, Integer, String, Long> t) {
            return new Tuple2<>(t.f0, t.f4);
        }
    }

    @Test
    public void testInputOfCombinerIsSortedForCombinableGroupReduceWithGroupSorting()
            throws Exception {
        /*
         * check that input of combiner is also sorted for combinable groupReduce with group sorting
         */

        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataSet<Tuple3<Integer, Long, String>> ds = CollectionDataSets.get3TupleDataSet(env);
        DataSet<Tuple3<Integer, Long, String>> reduceDs =
                ds.groupBy(1)
                        .sortGroup(0, Order.ASCENDING)
                        .reduceGroup(new OrderCheckingCombinableReduce());

        List<Tuple3<Integer, Long, String>> result = reduceDs.collect();

        String expected =
                "1,1,Hi\n"
                        + "2,2,Hello\n"
                        + "4,3,Hello world, how are you?\n"
                        + "7,4,Comment#1\n"
                        + "11,5,Comment#5\n"
                        + "16,6,Comment#10\n";

        compareResultAsTuples(result, expected);
    }

    @Test
    public void testDeepNesting() throws Exception {
        /*
         * Deep nesting test
         * + null value in pojo
         */
        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        DataSet<CrazyNested> ds = CollectionDataSets.getCrazyNestedDataSet(env);
        DataSet<Tuple2<String, Integer>> reduceDs =
                ds.groupBy("nestLvl1.nestLvl2.nestLvl3.nestLvl4.f1nal")
                        .reduceGroup(new GroupReducer1());

        List<Tuple2<String, Integer>> result = reduceDs.collect();

        String expected = "aa,1\nbb,2\ncc,3\n";

        compareResultAsTuples(result, expected);
    }

    private static class GroupReducer1
            implements GroupReduceFunction<
                    CollectionDataSets.CrazyNested, Tuple2<String, Integer>> {
        private static final long serialVersionUID = 1L;

        @Override
        public void reduce(Iterable<CrazyNested> values, Collector<Tuple2<String, Integer>> out)
                throws Exception {
            int c = 0;
            String n = null;
            for (CrazyNested v : values) {
                c++; // haha
                n = v.nestLvl1.nestLvl2.nestLvl3.nestLvl4.f1nal;
            }
            out.collect(new Tuple2<>(n, c));
        }
    }

    @Test
    public void testPojoExtendingFromTupleWithCustomFields() throws Exception {
        /*
         * Test Pojo extending from tuple WITH custom fields
         */
        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        DataSet<FromTupleWithCTor> ds = CollectionDataSets.getPojoExtendingFromTuple(env);
        DataSet<Integer> reduceDs = ds.groupBy("special", "f2").reduceGroup(new GroupReducer2());

        List<Integer> result = reduceDs.collect();

        String expected = "3\n2\n";

        compareResultAsText(result, expected);
    }

    private static class GroupReducer2 implements GroupReduceFunction<FromTupleWithCTor, Integer> {

        @Override
        public void reduce(Iterable<FromTupleWithCTor> values, Collector<Integer> out) {
            out.collect(countElements(values));
        }
    }

    @Test
    public void testPojoContainigWritableAndTuples() throws Exception {
        /*
         * Test Pojo containing a Writable and Tuples
         */
        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        DataSet<PojoContainingTupleAndWritable> ds =
                CollectionDataSets.getPojoContainingTupleAndWritable(env);
        DataSet<Integer> reduceDs =
                ds.groupBy("hadoopFan", "theTuple.*") // full tuple selection
                        .reduceGroup(new GroupReducer3());

        List<Integer> result = reduceDs.collect();

        String expected = "1\n5\n";

        compareResultAsText(result, expected);
    }

    private static class GroupReducer3
            implements GroupReduceFunction<PojoContainingTupleAndWritable, Integer> {

        @Override
        public void reduce(
                Iterable<PojoContainingTupleAndWritable> values, Collector<Integer> out) {
            out.collect(countElements(values));
        }
    }

    @Test
    public void testTupleContainingPojosAndRegularFields() throws Exception {
        /*
         * Test Tuple containing pojos and regular fields
         */
        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        DataSet<Tuple3<Integer, CrazyNested, POJO>> ds =
                CollectionDataSets.getTupleContainingPojos(env);

        DataSet<Integer> reduceDs =
                ds.groupBy("f0", "f1.*") // nested full tuple selection
                        .reduceGroup(new GroupReducer4());

        List<Integer> result = reduceDs.collect();

        String expected = "3\n1\n";

        compareResultAsText(result, expected);
    }

    private static class GroupReducer4
            implements GroupReduceFunction<Tuple3<Integer, CrazyNested, POJO>, Integer> {
        private static final long serialVersionUID = 1L;

        @Override
        public void reduce(
                Iterable<Tuple3<Integer, CrazyNested, POJO>> values, Collector<Integer> out) {
            out.collect(countElements(values));
        }
    }

    @Test
    public void testStringBasedDefinitionOnGroupSort() throws Exception {
        /*
         * Test string-based definition on group sort, based on test:
         * check correctness of groupReduce with descending group sort
         */
        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataSet<Tuple3<Integer, Long, String>> ds = CollectionDataSets.get3TupleDataSet(env);
        DataSet<Tuple3<Integer, Long, String>> reduceDs =
                ds.groupBy(1)
                        .sortGroup("f2", Order.DESCENDING)
                        .reduceGroup(new Tuple3SortedGroupReduce());

        List<Tuple3<Integer, Long, String>> result = reduceDs.collect();

        String expected =
                "1,1,Hi\n"
                        + "5,2,Hello world-Hello\n"
                        + "15,3,Luke Skywalker-I am fine.-Hello world, how are you?\n"
                        + "34,4,Comment#4-Comment#3-Comment#2-Comment#1\n"
                        + "65,5,Comment#9-Comment#8-Comment#7-Comment#6-Comment#5\n"
                        + "111,6,Comment#15-Comment#14-Comment#13-Comment#12-Comment#11-Comment#10\n";

        compareResultAsTuples(result, expected);
    }

    @Test
    public void testIntBasedDefinitionOnGroupSortForFullNestedTuple() throws Exception {
        /*
         * Test int-based definition on group sort, for (full) nested Tuple
         */
        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataSet<Tuple2<Tuple2<Integer, Integer>, String>> ds =
                CollectionDataSets.getGroupSortedNestedTupleDataSet(env);
        DataSet<String> reduceDs =
                ds.groupBy("f1")
                        .sortGroup(0, Order.DESCENDING)
                        .reduceGroup(new NestedTupleReducer());
        List<String> result = reduceDs.collect();

        String expected = "a--(2,1)-(1,3)-(1,2)-\n" + "b--(2,2)-\n" + "c--(4,9)-(3,6)-(3,3)-\n";

        compareResultAsText(result, expected);
    }

    @Test
    public void testIntBasedDefinitionOnGroupSortForPartialNestedTuple() throws Exception {
        /*
         * Test int-based definition on group sort, for (partial) nested Tuple ASC
         */
        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataSet<Tuple2<Tuple2<Integer, Integer>, String>> ds =
                CollectionDataSets.getGroupSortedNestedTupleDataSet(env);
        // f0.f0 is first integer
        DataSet<String> reduceDs =
                ds.groupBy("f1")
                        .sortGroup("f0.f0", Order.ASCENDING)
                        .sortGroup("f0.f1", Order.ASCENDING)
                        .reduceGroup(new NestedTupleReducer());
        List<String> result = reduceDs.collect();

        String expected = "a--(1,2)-(1,3)-(2,1)-\n" + "b--(2,2)-\n" + "c--(3,3)-(3,6)-(4,9)-\n";

        compareResultAsText(result, expected);
    }

    @Test
    public void testStringBasedDefinitionOnGroupSortForPartialNestedTuple() throws Exception {
        /*
         * Test string-based definition on group sort, for (partial) nested Tuple DESC
         */
        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataSet<Tuple2<Tuple2<Integer, Integer>, String>> ds =
                CollectionDataSets.getGroupSortedNestedTupleDataSet(env);
        // f0.f0 is first integer
        DataSet<String> reduceDs =
                ds.groupBy("f1")
                        .sortGroup("f0.f0", Order.DESCENDING)
                        .reduceGroup(new NestedTupleReducer());
        List<String> result = reduceDs.collect();

        String expected = "a--(2,1)-(1,3)-(1,2)-\n" + "b--(2,2)-\n" + "c--(4,9)-(3,3)-(3,6)-\n";

        compareResultAsText(result, expected);
    }

    @Test
    public void testStringBasedDefinitionOnGroupSortForTwoGroupingKeys() throws Exception {
        /*
         * Test string-based definition on group sort, for two grouping keys
         */
        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataSet<Tuple2<Tuple2<Integer, Integer>, String>> ds =
                CollectionDataSets.getGroupSortedNestedTupleDataSet(env);
        // f0.f0 is first integer
        DataSet<String> reduceDs =
                ds.groupBy("f1")
                        .sortGroup("f0.f0", Order.DESCENDING)
                        .sortGroup("f0.f1", Order.DESCENDING)
                        .reduceGroup(new NestedTupleReducer());
        List<String> result = reduceDs.collect();

        String expected = "a--(2,1)-(1,3)-(1,2)-\n" + "b--(2,2)-\n" + "c--(4,9)-(3,6)-(3,3)-\n";

        compareResultAsText(result, expected);
    }

    @Test
    public void testStringBasedDefinitionOnGroupSortForTwoGroupingKeysWithPojos() throws Exception {
        /*
         * Test string-based definition on group sort, for two grouping keys with Pojos
         */
        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataSet<PojoContainingTupleAndWritable> ds =
                CollectionDataSets.getGroupSortedPojoContainingTupleAndWritable(env);
        // f0.f0 is first integer
        DataSet<String> reduceDs =
                ds.groupBy("hadoopFan")
                        .sortGroup("theTuple.f0", Order.DESCENDING)
                        .sortGroup("theTuple.f1", Order.DESCENDING)
                        .reduceGroup(new GroupReducer5());
        List<String> result = reduceDs.collect();

        String expected = "1---(10,100)-\n" + "2---(30,600)-(30,400)-(30,200)-(20,201)-(20,200)-\n";

        compareResultAsText(result, expected);
    }

    @Test
    public void testTupleKeySelectorGroupSort() throws Exception {
        /*
         * check correctness of sorted groupReduce on tuples with keyselector sorting
         */

        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataSet<Tuple3<Integer, Long, String>> ds = CollectionDataSets.get3TupleDataSet(env);
        DataSet<Tuple3<Integer, Long, String>> reduceDs =
                ds.groupBy(new LongFieldExtractor<Tuple3<Integer, Long, String>>(1))
                        .sortGroup(
                                new StringFieldExtractor<Tuple3<Integer, Long, String>>(2),
                                Order.DESCENDING)
                        .reduceGroup(new Tuple3SortedGroupReduce());

        List<Tuple3<Integer, Long, String>> result = reduceDs.collect();

        String expected =
                "1,1,Hi\n"
                        + "5,2,Hello world-Hello\n"
                        + "15,3,Luke Skywalker-I am fine.-Hello world, how are you?\n"
                        + "34,4,Comment#4-Comment#3-Comment#2-Comment#1\n"
                        + "65,5,Comment#9-Comment#8-Comment#7-Comment#6-Comment#5\n"
                        + "111,6,Comment#15-Comment#14-Comment#13-Comment#12-Comment#11-Comment#10\n";

        compareResultAsTuples(result, expected);
    }

    private static class TwoTuplePojoExtractor
            implements KeySelector<CustomType, Tuple2<Integer, Integer>> {
        private static final long serialVersionUID = 1L;

        @Override
        public Tuple2<Integer, Integer> getKey(CustomType value) throws Exception {
            return new Tuple2<>(value.myInt, value.myInt);
        }
    }

    private static class StringPojoExtractor implements KeySelector<CustomType, String> {
        private static final long serialVersionUID = 1L;

        @Override
        public String getKey(CustomType value) throws Exception {
            return value.myString;
        }
    }

    @Test
    public void testPojoKeySelectorGroupSort() throws Exception {
        /*
         * check correctness of sorted groupReduce on custom type with keyselector sorting
         */

        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        DataSet<CustomType> ds = CollectionDataSets.getCustomTypeDataSet(env);
        DataSet<CustomType> reduceDs =
                ds.groupBy(new TwoTuplePojoExtractor())
                        .sortGroup(new StringPojoExtractor(), Order.DESCENDING)
                        .reduceGroup(new CustomTypeSortedGroupReduce());

        List<CustomType> result = reduceDs.collect();

        String expected =
                "1,0,Hi\n"
                        + "2,3,Hello world-Hello\n"
                        + "3,12,Luke Skywalker-I am fine.-Hello world, how are you?\n"
                        + "4,30,Comment#4-Comment#3-Comment#2-Comment#1\n"
                        + "5,60,Comment#9-Comment#8-Comment#7-Comment#6-Comment#5\n"
                        + "6,105,Comment#15-Comment#14-Comment#13-Comment#12-Comment#11-Comment#10\n";

        compareResultAsText(result, expected);
    }

    private static class LongFieldExtractor<T extends Tuple> implements KeySelector<T, Long> {
        private static final long serialVersionUID = 1L;
        private int field;

        public LongFieldExtractor() {}

        public LongFieldExtractor(int field) {
            this.field = field;
        }

        @Override
        public Long getKey(T t) throws Exception {
            return ((Tuple) t).getField(field);
        }
    }

    private static class IntFieldExtractor<T extends Tuple> implements KeySelector<T, Integer> {
        private static final long serialVersionUID = 1L;
        private int field;

        public IntFieldExtractor() {}

        public IntFieldExtractor(int field) {
            this.field = field;
        }

        @Override
        public Integer getKey(T t) throws Exception {
            return ((Tuple) t).getField(field);
        }
    }

    private static class StringFieldExtractor<T extends Tuple> implements KeySelector<T, String> {
        private static final long serialVersionUID = 1L;
        private int field;

        public StringFieldExtractor() {}

        public StringFieldExtractor(int field) {
            this.field = field;
        }

        @Override
        public String getKey(T t) throws Exception {
            return t.getField(field);
        }
    }

    @Test
    public void testTupleKeySelectorSortWithCombine() throws Exception {
        /*
         * check correctness of sorted groupReduce with combine on tuples with keyselector sorting
         */

        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataSet<Tuple3<Integer, Long, String>> ds = CollectionDataSets.get3TupleDataSet(env);
        DataSet<Tuple2<Integer, String>> reduceDs =
                ds.groupBy(new LongFieldExtractor<Tuple3<Integer, Long, String>>(1))
                        .sortGroup(
                                new StringFieldExtractor<Tuple3<Integer, Long, String>>(2),
                                Order.DESCENDING)
                        .reduceGroup(new Tuple3SortedGroupReduceWithCombine());

        List<Tuple2<Integer, String>> result = reduceDs.collect();

        if (super.mode != TestExecutionMode.COLLECTION) {
            String expected =
                    "1,Hi\n"
                            + "5,Hello world-Hello\n"
                            + "15,Luke Skywalker-I am fine.-Hello world, how are you?\n"
                            + "34,Comment#4-Comment#3-Comment#2-Comment#1\n"
                            + "65,Comment#9-Comment#8-Comment#7-Comment#6-Comment#5\n"
                            + "111,Comment#15-Comment#14-Comment#13-Comment#12-Comment#11-Comment#10\n";

            compareResultAsTuples(result, expected);
        }
    }

    private static class FiveToTwoTupleExtractor
            implements KeySelector<
                    Tuple5<Integer, Long, Integer, String, Long>, Tuple2<Long, Integer>> {
        private static final long serialVersionUID = 1L;

        @Override
        public Tuple2<Long, Integer> getKey(Tuple5<Integer, Long, Integer, String, Long> in) {
            return new Tuple2<>(in.f4, in.f2);
        }
    }

    @Test
    public void testTupleKeySelectorSortCombineOnTuple() throws Exception {
        /*
         * check correctness of sorted groupReduceon with Tuple2 keyselector sorting
         */

        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataSet<Tuple5<Integer, Long, Integer, String, Long>> ds =
                CollectionDataSets.get5TupleDataSet(env);
        DataSet<Tuple5<Integer, Long, Integer, String, Long>> reduceDs =
                ds.groupBy(new IntFieldExtractor<Tuple5<Integer, Long, Integer, String, Long>>(0))
                        .sortGroup(new FiveToTwoTupleExtractor(), Order.DESCENDING)
                        .reduceGroup(new Tuple5SortedGroupReduce());

        List<Tuple5<Integer, Long, Integer, String, Long>> result = reduceDs.collect();

        String expected =
                "1,1,0,Hallo,1\n"
                        + "2,5,0,Hallo Welt-Hallo Welt wie,1\n"
                        + "3,15,0,BCD-ABC-Hallo Welt wie gehts?,2\n"
                        + "4,34,0,FGH-CDE-EFG-DEF,1\n"
                        + "5,65,0,IJK-HIJ-KLM-JKL-GHI,1\n";

        compareResultAsTuples(result, expected);
    }

    private static class GroupReducer5
            implements GroupReduceFunction<
                    CollectionDataSets.PojoContainingTupleAndWritable, String> {
        @Override
        public void reduce(Iterable<PojoContainingTupleAndWritable> values, Collector<String> out)
                throws Exception {
            boolean once = false;
            StringBuilder concat = new StringBuilder();
            for (PojoContainingTupleAndWritable value : values) {
                if (!once) {
                    concat.append(value.hadoopFan.get());
                    concat.append("---");
                    once = true;
                }
                concat.append(value.theTuple);
                concat.append("-");
            }
            out.collect(concat.toString());
        }
    }

    @Test
    public void testGroupingWithPojoContainingMultiplePojos() throws Exception {
        /*
         * Test grouping with pojo containing multiple pojos (was a bug)
         */
        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataSet<CollectionDataSets.PojoWithMultiplePojos> ds =
                CollectionDataSets.getPojoWithMultiplePojos(env);

        // f0.f0 is first integer
        DataSet<String> reduceDs = ds.groupBy("p2.a2").reduceGroup(new GroupReducer6());
        List<String> result = reduceDs.collect();

        String expected = "b\nccc\nee\n";

        compareResultAsText(result, expected);
    }

    private static class GroupReducer6
            implements GroupReduceFunction<CollectionDataSets.PojoWithMultiplePojos, String> {
        @Override
        public void reduce(
                Iterable<CollectionDataSets.PojoWithMultiplePojos> values, Collector<String> out)
                throws Exception {
            StringBuilder concat = new StringBuilder();
            for (CollectionDataSets.PojoWithMultiplePojos value : values) {
                concat.append(value.p2.a2);
            }
            out.collect(concat.toString());
        }
    }

    @Test
    public void testJavaCollectionsWithinPojos() throws Exception {
        /*
         * Test Java collections within pojos ( == test kryo)
         */
        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataSet<CollectionDataSets.PojoWithCollection> ds =
                CollectionDataSets.getPojoWithCollection(env);
        // f0.f0 is first integer
        DataSet<String> reduceDs = ds.groupBy("key").reduceGroup(new GroupReducer7());
        List<String> result = reduceDs.collect();

        String expected =
                "callFor key 0 we got: pojo.a=apojo.a=bFor key 0 we got: pojo.a=a2pojo.a=b2\n";

        compareResultAsText(result, expected);
    }

    private static class GroupReducer7
            implements GroupReduceFunction<CollectionDataSets.PojoWithCollection, String> {

        @Override
        public void reduce(
                Iterable<CollectionDataSets.PojoWithCollection> values, Collector<String> out) {
            StringBuilder concat = new StringBuilder();
            concat.append("call");
            for (CollectionDataSets.PojoWithCollection value : values) {
                concat.append("For key ").append(value.key).append(" we got: ");

                for (CollectionDataSets.Pojo1 p : value.pojos) {
                    concat.append("pojo.a=").append(p.a);
                }
            }
            out.collect(concat.toString());
        }
    }

    @Test
    public void testGroupByGenericType() throws Exception {
        /*
         * Group by generic type
         */
        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataSet<CollectionDataSets.PojoWithCollection> ds =
                CollectionDataSets.getPojoWithCollection(env);

        // f0.f0 is first integer
        DataSet<String> reduceDs = ds.groupBy("bigInt").reduceGroup(new GroupReducer8());
        List<String> result = reduceDs.collect();
        ExecutionConfig ec = env.getConfig();

        // check if automatic type registration with Kryo worked
        Assert.assertTrue(ec.getRegisteredKryoTypes().contains(BigInt.class));
        Assert.assertFalse(ec.getRegisteredKryoTypes().contains(java.sql.Date.class));

        String expected = null;

        String localExpected =
                "[call\n"
                        + "For key 92233720368547758070 we got:\n"
                        + "PojoWithCollection{pojos.size()=2, key=0, sqlDate=2033-05-18, bigInt=92233720368547758070, bigDecimalKeepItNull=null, scalaBigInt=10, mixed=[{someKey=1}, /this/is/wrong, uhlala]}\n"
                        + "For key 92233720368547758070 we got:\n"
                        + "PojoWithCollection{pojos.size()=2, key=0, sqlDate=1976-05-03, bigInt=92233720368547758070, bigDecimalKeepItNull=null, scalaBigInt=31104000, mixed=null}]";

        Assert.assertEquals(localExpected, result.toString());
    }

    @Test
    public void testGroupReduceSelectorKeysWithSemProps() throws Exception {

        /*
         * Test that semantic properties are correctly adapted when using Selector Keys
         */

        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);

        DataSet<Tuple5<Integer, Long, Integer, String, Long>> ds =
                CollectionDataSets.get5TupleDataSet(env);
        DataSet<Tuple2<Integer, Long>> reduceDs =
                ds
                        // group by selector key
                        .groupBy(
                                new KeySelector<
                                        Tuple5<Integer, Long, Integer, String, Long>, Long>() {
                                    @Override
                                    public Long getKey(
                                            Tuple5<Integer, Long, Integer, String, Long> v)
                                            throws Exception {
                                        return (v.f0 * v.f1) - (v.f2 * v.f4);
                                    }
                                })
                        .reduceGroup(
                                new GroupReduceFunction<
                                        Tuple5<Integer, Long, Integer, String, Long>,
                                        Tuple5<Integer, Long, Integer, String, Long>>() {
                                    @Override
                                    public void reduce(
                                            Iterable<Tuple5<Integer, Long, Integer, String, Long>>
                                                    values,
                                            Collector<Tuple5<Integer, Long, Integer, String, Long>>
                                                    out)
                                            throws Exception {
                                        for (Tuple5<Integer, Long, Integer, String, Long> v :
                                                values) {
                                            out.collect(v);
                                        }
                                    }
                                })
                        // add forward field information
                        .withForwardedFields("0")
                        // group again and reduce
                        .groupBy(0)
                        .reduceGroup(
                                new GroupReduceFunction<
                                        Tuple5<Integer, Long, Integer, String, Long>,
                                        Tuple2<Integer, Long>>() {
                                    @Override
                                    public void reduce(
                                            Iterable<Tuple5<Integer, Long, Integer, String, Long>>
                                                    values,
                                            Collector<Tuple2<Integer, Long>> out)
                                            throws Exception {
                                        int k = 0;
                                        long s = 0;
                                        for (Tuple5<Integer, Long, Integer, String, Long> v :
                                                values) {
                                            k = v.f0;
                                            s += v.f1;
                                        }
                                        out.collect(new Tuple2<>(k, s));
                                    }
                                });

        List<Tuple2<Integer, Long>> result = reduceDs.collect();

        String expected = "1,1\n" + "2,5\n" + "3,15\n" + "4,34\n" + "5,65\n";

        compareResultAsTuples(result, expected);
    }

    @Test
    public void testGroupReduceWithAtomicValue() throws Exception {
        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        DataSet<Integer> ds = env.fromElements(1, 1, 2, 3, 4);
        DataSet<Integer> reduceDs =
                ds.groupBy("*")
                        .reduceGroup(
                                new GroupReduceFunction<Integer, Integer>() {
                                    @Override
                                    public void reduce(
                                            Iterable<Integer> values, Collector<Integer> out)
                                            throws Exception {
                                        out.collect(values.iterator().next());
                                    }
                                });

        List<Integer> result = reduceDs.collect();

        String expected = "1\n" + "2\n" + "3\n" + "4";

        compareResultAsText(result, expected);
    }

    /**
     * Fix for FLINK-2019.
     *
     * @throws Exception
     */
    @Test
    public void testJodatimeDateTimeWithKryo() throws Exception {
        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        DataSet<Tuple2<Integer, DateTime>> ds = env.fromElements(new Tuple2<>(1, DateTime.now()));
        DataSet<Tuple2<Integer, DateTime>> reduceDs = ds.groupBy("f1").sum(0).project(0);

        List<Tuple2<Integer, DateTime>> result = reduceDs.collect();

        String expected = "1\n";

        compareResultAsTuples(result, expected);
    }

    /**
     * Fix for FLINK-2158.
     *
     * @throws Exception
     */
    @Test
    public void testDateNullException() throws Exception {
        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        DataSet<Tuple2<Integer, Date>> in =
                env.fromElements(
                        new Tuple2<>(0, new Date(1230000000)),
                        new Tuple2<Integer, Date>(1, null),
                        new Tuple2<>(2, new Date(1230000000)));

        DataSet<String> r =
                in.groupBy(0)
                        .reduceGroup(
                                new GroupReduceFunction<Tuple2<Integer, Date>, String>() {
                                    @Override
                                    public void reduce(
                                            Iterable<Tuple2<Integer, Date>> values,
                                            Collector<String> out)
                                            throws Exception {
                                        for (Tuple2<Integer, Date> e : values) {
                                            out.collect(Integer.toString(e.f0));
                                        }
                                    }
                                });

        List<String> result = r.collect();

        String expected = "0\n1\n2\n";
        compareResultAsText(result, expected);
    }

    private static class GroupReducer8
            implements GroupReduceFunction<CollectionDataSets.PojoWithCollection, String> {
        @Override
        public void reduce(
                Iterable<CollectionDataSets.PojoWithCollection> values, Collector<String> out) {
            StringBuilder concat = new StringBuilder();
            concat.append("call");
            for (CollectionDataSets.PojoWithCollection value : values) {
                concat.append("\nFor key ").append(value.bigInt).append(" we got:\n").append(value);
            }
            out.collect(concat.toString());
        }
    }

    private static class NestedTupleReducer
            implements GroupReduceFunction<Tuple2<Tuple2<Integer, Integer>, String>, String> {
        @Override
        public void reduce(
                Iterable<Tuple2<Tuple2<Integer, Integer>, String>> values, Collector<String> out) {
            boolean once = false;
            StringBuilder concat = new StringBuilder();
            for (Tuple2<Tuple2<Integer, Integer>, String> value : values) {
                if (!once) {
                    concat.append(value.f1).append("--");
                    once = true;
                }
                concat.append(value.f0); // the tuple with the sorted groups
                concat.append("-");
            }
            out.collect(concat.toString());
        }
    }

    private static class Tuple3GroupReduce
            implements GroupReduceFunction<Tuple3<Integer, Long, String>, Tuple2<Integer, Long>> {
        private static final long serialVersionUID = 1L;

        @Override
        public void reduce(
                Iterable<Tuple3<Integer, Long, String>> values,
                Collector<Tuple2<Integer, Long>> out) {
            int i = 0;
            long l = 0L;

            for (Tuple3<Integer, Long, String> t : values) {
                i += t.f0;
                l = t.f1;
            }

            out.collect(new Tuple2<>(i, l));
        }
    }

    private static class Tuple3SortedGroupReduce
            implements GroupReduceFunction<
                    Tuple3<Integer, Long, String>, Tuple3<Integer, Long, String>> {
        private static final long serialVersionUID = 1L;

        @Override
        public void reduce(
                Iterable<Tuple3<Integer, Long, String>> values,
                Collector<Tuple3<Integer, Long, String>> out) {
            int sum = 0;
            long key = 0;
            StringBuilder concat = new StringBuilder();

            for (Tuple3<Integer, Long, String> next : values) {
                sum += next.f0;
                key = next.f1;
                concat.append(next.f2).append("-");
            }

            if (concat.length() > 0) {
                concat.setLength(concat.length() - 1);
            }

            out.collect(new Tuple3<>(sum, key, concat.toString()));
        }
    }

    private static class Tuple5GroupReduce
            implements GroupReduceFunction<
                    Tuple5<Integer, Long, Integer, String, Long>,
                    Tuple5<Integer, Long, Integer, String, Long>> {
        private static final long serialVersionUID = 1L;

        @Override
        public void reduce(
                Iterable<Tuple5<Integer, Long, Integer, String, Long>> values,
                Collector<Tuple5<Integer, Long, Integer, String, Long>> out) {
            int i = 0;
            long l = 0L;
            long l2 = 0L;

            for (Tuple5<Integer, Long, Integer, String, Long> t : values) {
                i = t.f0;
                l += t.f1;
                l2 = t.f4;
            }

            out.collect(new Tuple5<>(i, l, 0, "P-)", l2));
        }
    }

    private static class Tuple5SortedGroupReduce
            implements GroupReduceFunction<
                    Tuple5<Integer, Long, Integer, String, Long>,
                    Tuple5<Integer, Long, Integer, String, Long>> {
        private static final long serialVersionUID = 1L;

        @Override
        public void reduce(
                Iterable<Tuple5<Integer, Long, Integer, String, Long>> values,
                Collector<Tuple5<Integer, Long, Integer, String, Long>> out) {
            int i = 0;
            long l = 0L;
            long l2 = 0L;
            StringBuilder concat = new StringBuilder();

            for (Tuple5<Integer, Long, Integer, String, Long> t : values) {
                i = t.f0;
                l += t.f1;
                concat.append(t.f3).append("-");
                l2 = t.f4;
            }
            if (concat.length() > 0) {
                concat.setLength(concat.length() - 1);
            }

            out.collect(new Tuple5<>(i, l, 0, concat.toString(), l2));
        }
    }

    private static class CustomTypeGroupReduce
            implements GroupReduceFunction<CustomType, CustomType> {
        private static final long serialVersionUID = 1L;

        @Override
        public void reduce(Iterable<CustomType> values, Collector<CustomType> out) {
            final Iterator<CustomType> iter = values.iterator();

            CustomType o = new CustomType();
            CustomType c = iter.next();

            o.myString = "Hello!";
            o.myInt = c.myInt;
            o.myLong = c.myLong;

            while (iter.hasNext()) {
                CustomType next = iter.next();
                o.myLong += next.myLong;
            }

            out.collect(o);
        }
    }

    private static class CustomTypeSortedGroupReduce
            implements GroupReduceFunction<CustomType, CustomType> {
        private static final long serialVersionUID = 1L;

        @Override
        public void reduce(Iterable<CustomType> values, Collector<CustomType> out) {
            final Iterator<CustomType> iter = values.iterator();

            CustomType o = new CustomType();
            CustomType c = iter.next();

            StringBuilder concat = new StringBuilder(c.myString);
            o.myInt = c.myInt;
            o.myLong = c.myLong;

            while (iter.hasNext()) {
                CustomType next = iter.next();
                concat.append("-").append(next.myString);
                o.myLong += next.myLong;
            }

            o.myString = concat.toString();
            out.collect(o);
        }
    }

    private static class InputReturningTuple3GroupReduce
            implements GroupReduceFunction<
                    Tuple3<Integer, Long, String>, Tuple3<Integer, Long, String>> {
        private static final long serialVersionUID = 1L;

        @Override
        public void reduce(
                Iterable<Tuple3<Integer, Long, String>> values,
                Collector<Tuple3<Integer, Long, String>> out) {

            for (Tuple3<Integer, Long, String> t : values) {

                if (t.f0 < 4) {
                    t.f2 = "Hi!";
                    t.f0 += 10;
                    out.collect(t);
                    t.f0 += 10;
                    t.f2 = "Hi again!";
                    out.collect(t);
                }
            }
        }
    }

    private static class AllAddingTuple3GroupReduce
            implements GroupReduceFunction<
                    Tuple3<Integer, Long, String>, Tuple3<Integer, Long, String>> {
        private static final long serialVersionUID = 1L;

        @Override
        public void reduce(
                Iterable<Tuple3<Integer, Long, String>> values,
                Collector<Tuple3<Integer, Long, String>> out) {

            int i = 0;
            long l = 0L;

            for (Tuple3<Integer, Long, String> t : values) {
                i += t.f0;
                l += t.f1;
            }

            out.collect(new Tuple3<>(i, l, "Hello World"));
        }
    }

    private static class AllAddingCustomTypeGroupReduce
            implements GroupReduceFunction<CustomType, CustomType> {
        private static final long serialVersionUID = 1L;

        @Override
        public void reduce(Iterable<CustomType> values, Collector<CustomType> out) {

            CustomType o = new CustomType(0, 0, "Hello!");

            for (CustomType next : values) {
                o.myInt += next.myInt;
                o.myLong += next.myLong;
            }

            out.collect(o);
        }
    }

    private static class BCTuple3GroupReduce
            extends RichGroupReduceFunction<
                    Tuple3<Integer, Long, String>, Tuple3<Integer, Long, String>> {
        private static final long serialVersionUID = 1L;
        private String f2Replace = "";

        @Override
        public void open(Configuration config) {

            Collection<Integer> ints = this.getRuntimeContext().getBroadcastVariable("ints");
            int sum = 0;
            for (Integer i : ints) {
                sum += i;
            }
            f2Replace = sum + "";
        }

        @Override
        public void reduce(
                Iterable<Tuple3<Integer, Long, String>> values,
                Collector<Tuple3<Integer, Long, String>> out) {

            int i = 0;
            long l = 0L;

            for (Tuple3<Integer, Long, String> t : values) {
                i += t.f0;
                l = t.f1;
            }

            out.collect(new Tuple3<>(i, l, this.f2Replace));
        }
    }

    private static class Tuple3GroupReduceWithCombine
            implements GroupReduceFunction<Tuple3<Integer, Long, String>, Tuple2<Integer, String>>,
                    GroupCombineFunction<
                            Tuple3<Integer, Long, String>, Tuple3<Integer, Long, String>> {
        private static final long serialVersionUID = 1L;

        @Override
        public void combine(
                Iterable<Tuple3<Integer, Long, String>> values,
                Collector<Tuple3<Integer, Long, String>> out) {

            Tuple3<Integer, Long, String> o = new Tuple3<>(0, 0L, "");

            for (Tuple3<Integer, Long, String> t : values) {
                o.f0 += t.f0;
                o.f1 = t.f1;
                o.f2 = "test" + o.f1;
            }

            out.collect(o);
        }

        @Override
        public void reduce(
                Iterable<Tuple3<Integer, Long, String>> values,
                Collector<Tuple2<Integer, String>> out) {

            int i = 0;
            String s = "";

            for (Tuple3<Integer, Long, String> t : values) {
                i += t.f0;
                s = t.f2;
            }

            out.collect(new Tuple2<>(i, s));
        }
    }

    private static class Tuple3SortedGroupReduceWithCombine
            implements GroupReduceFunction<Tuple3<Integer, Long, String>, Tuple2<Integer, String>>,
                    GroupCombineFunction<
                            Tuple3<Integer, Long, String>, Tuple3<Integer, Long, String>> {
        private static final long serialVersionUID = 1L;

        @Override
        public void combine(
                Iterable<Tuple3<Integer, Long, String>> values,
                Collector<Tuple3<Integer, Long, String>> out) {
            int sum = 0;
            long key = 0;
            StringBuilder concat = new StringBuilder();

            for (Tuple3<Integer, Long, String> next : values) {
                sum += next.f0;
                key = next.f1;
                concat.append(next.f2).append("-");
            }

            if (concat.length() > 0) {
                concat.setLength(concat.length() - 1);
            }

            out.collect(new Tuple3<>(sum, key, concat.toString()));
        }

        @Override
        public void reduce(
                Iterable<Tuple3<Integer, Long, String>> values,
                Collector<Tuple2<Integer, String>> out) {
            int i = 0;
            String s = "";

            for (Tuple3<Integer, Long, String> t : values) {
                i += t.f0;
                s = t.f2;
            }

            out.collect(new Tuple2<>(i, s));
        }
    }

    private static class Tuple3AllGroupReduceWithCombine
            implements GroupReduceFunction<Tuple3<Integer, Long, String>, Tuple2<Integer, String>>,
                    GroupCombineFunction<
                            Tuple3<Integer, Long, String>, Tuple3<Integer, Long, String>> {
        private static final long serialVersionUID = 1L;

        @Override
        public void combine(
                Iterable<Tuple3<Integer, Long, String>> values,
                Collector<Tuple3<Integer, Long, String>> out) {

            Tuple3<Integer, Long, String> o = new Tuple3<>(0, 0L, "");

            for (Tuple3<Integer, Long, String> t : values) {
                o.f0 += t.f0;
                o.f1 += t.f1;
                o.f2 += "test";
            }

            out.collect(o);
        }

        @Override
        public void reduce(
                Iterable<Tuple3<Integer, Long, String>> values,
                Collector<Tuple2<Integer, String>> out) {

            int i = 0;
            String s = "";

            for (Tuple3<Integer, Long, String> t : values) {
                i += t.f0 + t.f1;
                s += t.f2;
            }

            out.collect(new Tuple2<>(i, s));
        }
    }

    private static class CustomTypeGroupReduceWithCombine
            implements GroupReduceFunction<CustomType, CustomType>,
                    GroupCombineFunction<CustomType, CustomType> {
        private static final long serialVersionUID = 1L;

        @Override
        public void combine(Iterable<CustomType> values, Collector<CustomType> out)
                throws Exception {

            CustomType o = new CustomType();

            for (CustomType c : values) {
                o.myInt = c.myInt;
                o.myLong += c.myLong;
                o.myString = "test" + c.myInt;
            }

            out.collect(o);
        }

        @Override
        public void reduce(Iterable<CustomType> values, Collector<CustomType> out) {

            CustomType o = new CustomType(0, 0, "");

            for (CustomType c : values) {
                o.myInt = c.myInt;
                o.myLong += c.myLong;
                o.myString = c.myString;
            }

            out.collect(o);
        }
    }

    private static class OrderCheckingCombinableReduce
            implements GroupReduceFunction<
                            Tuple3<Integer, Long, String>, Tuple3<Integer, Long, String>>,
                    GroupCombineFunction<
                            Tuple3<Integer, Long, String>, Tuple3<Integer, Long, String>> {
        private static final long serialVersionUID = 1L;

        @Override
        public void reduce(
                Iterable<Tuple3<Integer, Long, String>> values,
                Collector<Tuple3<Integer, Long, String>> out)
                throws Exception {
            Iterator<Tuple3<Integer, Long, String>> it = values.iterator();
            Tuple3<Integer, Long, String> t = it.next();

            int i = t.f0;
            out.collect(t);

            while (it.hasNext()) {
                t = it.next();
                if (i > t.f0 || t.f2.equals("INVALID-ORDER!")) {
                    t.f2 = "INVALID-ORDER!";
                    out.collect(t);
                }
            }
        }

        @Override
        public void combine(
                Iterable<Tuple3<Integer, Long, String>> values,
                Collector<Tuple3<Integer, Long, String>> out) {

            Iterator<Tuple3<Integer, Long, String>> it = values.iterator();
            Tuple3<Integer, Long, String> t = it.next();

            int i = t.f0;
            out.collect(t);

            while (it.hasNext()) {
                t = it.next();
                if (i > t.f0) {
                    t.f2 = "INVALID-ORDER!";
                    out.collect(t);
                }
            }
        }
    }

    private static final class IdentityMapper<T> extends RichMapFunction<T, T> {
        @Override
        public T map(T value) {
            return value;
        }
    }

    private static int countElements(Iterable<?> iterable) {
        int c = 0;
        for (@SuppressWarnings("unused") Object o : iterable) {
            c++;
        }
        return c;
    }
}