gt:     int size = parents.size();
context: /*
   * This file is part of Gradoop.
   *
   * Gradoop is free software: you can redistribute it and/or modify
   * it under the terms of the GNU General Public License as published by
   * the Free Software Foundation, either version 3 of the License, or
   * (at your option) any later version.
   *
   * Gradoop is distributed in the hope that it will be useful,
   * but WITHOUT ANY WARRANTY; without even the implied warranty of
   * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
   * GNU General Public License for more details.
   *
   * You should have received a copy of the GNU General Public License
   * along with Gradoop. If not, see <http://www.gnu.org/licenses/>.
   */
  
  package org.gradoop.flink.algorithms.fsm.transactional.tle.functions;
  
  import com.google.common.collect.Lists;
  import com.google.common.collect.Maps;
  import com.google.common.collect.Sets;
  import org.apache.flink.api.common.functions.FlatJoinFunction;
  import org.apache.flink.api.common.functions.FlatMapFunction;
  import org.apache.flink.util.Collector;
  import org.gradoop.flink.algorithms.fsm.transactional.common.FSMConfig;
  import org.gradoop.flink.algorithms.fsm.transactional.tle.canonicalization.CanonicalLabeler;
  import org.gradoop.flink.algorithms.fsm.transactional.tle.pojos.Embedding;
  import org.gradoop.flink.algorithms.fsm.transactional.tle.pojos.FSMEdge;
  import org.gradoop.flink.algorithms.fsm.transactional.tle.pojos.FSMGraph;
  import org.gradoop.flink.algorithms.fsm.transactional.tle.tuples.SubgraphEmbeddings;
  
  import java.util.Collection;
  import java.util.List;
  import java.util.Map;
  import java.util.Set;
  import java.util.TreeSet;
  
  /**
   * Superclass of classes joining subgraph embeddings.
   *
   * @param <G> graph type
   * @param <SE> subgraph type
   */
  public class JoinEmbeddings<G extends FSMGraph, SE extends SubgraphEmbeddings>
    implements FlatJoinFunction<SE, G, SE>, FlatMapFunction<SE, SE> {
  
    /**
     * Labeler used to generate canonical labels.
     */
    private final CanonicalLabeler canonicalLabeler;
  
    /**
     * Constructor.
     *
     * @param fsmConfig FSM configuration.
     */
    public JoinEmbeddings(FSMConfig fsmConfig) {
      this.canonicalLabeler = new CanonicalLabeler(fsmConfig.isDirected());
    }
  
    @Override
    public void join(SE embeddings, G graph, Collector<SE> out) throws
      Exception {
  
      Set<TreeSet<Integer>> edgeSets = Sets.newHashSet();
      Map<String, List<Embedding>> subgraphEmbeddings = Maps.newHashMap();
  
      for (Embedding parent : embeddings.getEmbeddings()) {
        for (Map.Entry<Integer, FSMEdge> entry : graph.getEdges().entrySet()) {
  
          int edgeId = entry.getKey();
  
          if (! parent.getEdges().containsKey(edgeId)) {
            FSMEdge edge = entry.getValue();
  
            int sourceId = edge.getSourceId();
            boolean containsSourceId =
              parent.getVertices().containsKey(sourceId);
  
            int targetId = edge.getTargetId();
            boolean containsTargetId =
              parent.getVertices().containsKey(targetId);
  
            if (containsSourceId ||  containsTargetId) {
              TreeSet<Integer> edgeSet = Sets.newTreeSet(parent.getEdgeIds());
              edgeSet.add(edgeId);
  
              if (! edgeSets.contains(edgeSet)) {
                edgeSets.add(edgeSet);
  
                Embedding child = parent.deepCopy();
                child.getEdges().put(edgeId, edge);
  
                if (! containsSourceId) {
                  child.getVertices()
                    .put(sourceId, graph.getVertices().get(sourceId));
                }
                if (! containsTargetId) {
                  child.getVertices()
                    .put(targetId, graph.getVertices().get(targetId));
                }
  
                String canonicalLabel = canonicalLabeler.label(child);
  
                List<Embedding> siblings =
                  subgraphEmbeddings.get(canonicalLabel);
  
                if (siblings == null) {
                  siblings = Lists.newArrayList(child);
                  subgraphEmbeddings.put(canonicalLabel, siblings);
                } else {
                  siblings.add(child);
                }
              }
            }
          }
        }
      }
  
      collect(embeddings, out, subgraphEmbeddings);
    }
  
    @Override
    public void flatMap(SE embeddings, Collector<SE> collector) throws Exception {
  
      Map<String, List<Embedding>> subgraphEmbeddings;
  
      if (embeddings.getSize() == 1) {
        subgraphEmbeddings = joinOnVertex(embeddings);
      } else {
        subgraphEmbeddings = joinOnEdgeOverlap(embeddings);
      }
  
      collect(embeddings, collector, subgraphEmbeddings);
    }
  
    /**
     * Cross-joins k-edge (k>1) embeddings on k-1-edge overlap.
     *
     * @param embeddings k-edge embeddings
     * @return Map of k+1-edge subgraphs and embeddings
     */
    private Map<String, List<Embedding>> joinOnEdgeOverlap(SE embeddings) {
      Map<String, List<Embedding>> subgraphEmbeddings;
      subgraphEmbeddings = Maps.newHashMap();
      List<Embedding> parents = embeddings.getEmbeddings();
  
      Set<TreeSet<Integer>> edgeSets = Sets.newHashSet();
  
      // index parents
  
      Map<TreeSet<Integer>, List<Integer>> subsetParents =  Maps.newHashMap();
  
      int parentIndex = 0;
      for (Embedding parent : parents) {
  
        for (Integer edgeId : parent.getEdgeIds()) {
          TreeSet<Integer> subset = Sets.newTreeSet(parent.getEdgeIds());
          subset.remove(edgeId);
  
          Collection<Integer> siblings = subsetParents.get(subset);
  
          if (siblings == null) {
            subsetParents.put(subset, Lists.newArrayList(parentIndex));
          } else {
            siblings.add(parentIndex);
          }
        }
  
        parentIndex++;
      }
  
      // join overlapping parents
  
      for (List<Integer> overlappingParents : subsetParents.values()) {
        int size = overlappingParents.size();
  
        if (size > 1) {
  
          for (int i = 0; i < size - 1; i++) {
            for (int j = i + 1; j < size; j++) {
  
              Embedding left = parents.get(overlappingParents.get(i));
              Embedding right = parents.get(overlappingParents.get(j));
  
              TreeSet<Integer> edgeSet = Sets.newTreeSet(left.getEdgeIds());
              edgeSet.addAll(right.getEdgeIds());
  
              if (!edgeSets.contains(edgeSet)) {
                edgeSets.add(edgeSet);
  
                Embedding child = left.deepCopy();
                child.getVertices().putAll(right.getVertices());
                child.getEdges().putAll(right.getEdges());
  
                String canonicalLabel = canonicalLabeler.label(child);
  
                List<Embedding> siblings =
                  subgraphEmbeddings.get(canonicalLabel);
  
                if (siblings == null) {
                  siblings = Lists.newArrayList(child);
                  subgraphEmbeddings.put(canonicalLabel, siblings);
                } else {
                  siblings.add(child);
                }
              }
            }
          }
        }
      }
      return subgraphEmbeddings;
    }
  
    /**
     * Cross-joins 1-edge embeddings on source and target vertices.
     *
     * @param embeddings 1-edge embeddings
     * @return Map of 2-edge subgraphs and embeddings
     */
    private Map<String, List<Embedding>> joinOnVertex(SE embeddings) {
      Map<String, List<Embedding>> subgraphEmbeddings;
      subgraphEmbeddings = Maps.newHashMap();
      List<Embedding> parents = embeddings.getEmbeddings();
  
  