/*
 * Copyright 2017 Philip Fritzsche <p-f@users.noreply.github.com>.
 *
 * Licensed 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.gradoop.examples.io.mag.parse.flink;

import java.util.ArrayList;
import java.util.List;
import org.apache.flink.api.common.functions.GroupCombineFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;
import org.gradoop.common.model.impl.properties.Properties;
import org.gradoop.common.model.impl.properties.Property;
import org.gradoop.common.model.impl.properties.PropertyValue;

/**
 * A combiner that combines pairs of key and properties by key. This must only
 * be used on DataSets grouped by KEY.
 */
public class AttributeGroupCombiner implements
        GroupCombineFunction<Tuple2<String, Properties>, Tuple2<String, Properties>> {

    /**
     * Combine 2 {@link Properties} to 1. Elements will be combined using
     * {@link #combine(PropertyValue, PropertyValue)}.
     *
     * @param first The first object.
     * @param second The second object.
     * @return The first object (with all values from the second added).
     */
    public static Properties combine(Properties first, Properties second) {
        for (Property prop : second) {
            if (first.containsKey(prop.getKey())) {
                PropertyValue firstVal = first.remove(prop.getKey());
                PropertyValue secondVal = prop.getValue();
                first.set(prop.getKey(), combine(firstVal, secondVal));
            } else {
                first.set(prop);
            }
        }
        return first;
    }

    /**
     * Combine 2 {@link PropertyValue}s to 1. The result will be a {@link List}
     * that is a union of the 2 input values. ({@link PropertyValue}s that are
     * not {@link List}s will be converted to lists first.
     *
     * @param first The first value.
     * @param second The second value.
     * @return The union of both values.
     */
    public static PropertyValue combine(PropertyValue first,
            PropertyValue second) {
        List<PropertyValue> values = first.isList() ? first.getList()
                : new ArrayList<>();
        if (second.isList()) {
            values.addAll(second.getList());
        } else {
            values.add(second);
        }
        return PropertyValue.create(values);
    }

    @Override
    public void combine(Iterable<Tuple2<String, Properties>> values,
            Collector<Tuple2<String, Properties>> out)
            throws MagParserException {
        String key = null;
        Properties prop = null;
        for (Tuple2<String, Properties> value : values) {
            if (key == null) {
                key = value.f0;
            } else if (!key.equals(value.f0)) {
                throw new MagParserException("Key mismatch.");
            }
            prop = prop == null ? value.f1 : combine(prop, value.f1);
        }
        out.collect(new Tuple2<>(key, prop));
    }

}

@@ -794,16 +794,75 @@
 nction;%0A
+import org.apache.flink.api.common.functions.JoinFunction;%0A
 import o
@@ -1120,16 +1120,76 @@
 tyValue;
+%0Aimport org.gradoop.flink.io.impl.graph.tuples.ImportVertex;
 %0A%0A/**%0A *
@@ -1303,16 +1303,82 @@
  by KEY.
+ Also acts as a custom join to join%0A * multiattributes with nodes.
 %0A */%0Apub
@@ -1505,16 +1505,110 @@
 erties%3E%3E
+,%0A        JoinFunction%3CImportVertex%3CString%3E, Tuple2%3CString, Properties%3E, ImportVertex%3CString%3E%3E
  %7B%0A%0A    
@@ -3746,10 +3746,273 @@
 %0A    %7D%0A%0A
+    @Override%0A    public ImportVertex%3CString%3E join(ImportVertex%3CString%3E first,%0A            Tuple2%3CString, Properties%3E second) throws MagParserException %7B%0A        return new ImportVertex%3C%3E(second.f0, first.f1,%0A                combine(first.f2, second.f1));%0A    %7D%0A%0A
 %7D%0A

Implement custom join with AttributeGroupCombiner.