@Test public void testFunction() throws Exception { CreateEdgesFromTriple<EPGMVertex, EPGMEdge> function = new CreateEdgesFromTriple<>( getConfig().getLogicalGraphFactory().getEdgeFactory(), "source", "target"); VertexFactory<EPGMVertex> vertexFactory = getConfig().getLogicalGraphFactory().getVertexFactory(); EPGMVertex testVertex1 = vertexFactory.createVertex(); EPGMVertex testVertex2 = vertexFactory.createVertex(); GradoopId source1 = GradoopId.get(); GradoopId source2 = GradoopId.get(); GradoopId target1 = GradoopId.get(); GradoopId target2 = GradoopId.get(); Tuple3<EPGMVertex, GradoopId, GradoopId> tuple1 = new Tuple3<>(testVertex1, source1, target1); Tuple3<EPGMVertex, GradoopId, GradoopId> tuple2 = new Tuple3<>(testVertex2, source2, target2); List<EPGMEdge> result = getExecutionEnvironment().fromElements(tuple1, tuple2).flatMap(function) .collect(); assertEquals(4, result.size()); assertEquals(4, result.stream().map(EPGMElement::getId).count()); assertEquals(4, result.stream().map(e -> Tuple2.of(e.getSourceId(), e.getTargetId())).distinct().count()); for (EPGMEdge resultEdge : result) { if (resultEdge.getLabel().equals("source")) { if (resultEdge.getSourceId().equals(source1)) { assertEquals(testVertex1.getId(), resultEdge.getTargetId()); } else if (resultEdge.getSourceId().equals(source2)) { assertEquals(testVertex2.getId(), resultEdge.getTargetId()); } else { fail("EPGMEdge with invalid source ID created."); } } else if (resultEdge.getLabel().equals("target")) { if (resultEdge.getSourceId().equals(testVertex1.getId())) { assertEquals(target1, resultEdge.getTargetId()); } else if (resultEdge.getSourceId().equals(testVertex2.getId())) { assertEquals(target2, resultEdge.getTargetId()); } else { fail("EPGMEdge with invalid source ID created."); } } else { fail("EPGMEdge with invalid label created."); } } }
@Override public void flatMap(Tuple3<V, GradoopId, GradoopId> triple, Collector<E> out) { reuse.setId(GradoopId.get()); reuse.setLabel(edgeLabelSourceToNew); reuse.setSourceId(triple.f1); reuse.setTargetId(triple.f0.getId()); out.collect(reuse); reuse.setId(GradoopId.get()); reuse.setLabel(edgeLabelNewToTarget); reuse.setSourceId(triple.f0.getId()); reuse.setTargetId(triple.f2); out.collect(reuse); }
CreateEdgesFromTriple implements FlatMapFunction<Tuple3<V, GradoopId, GradoopId>, E>, ResultTypeQueryable<E> { @Override public void flatMap(Tuple3<V, GradoopId, GradoopId> triple, Collector<E> out) { reuse.setId(GradoopId.get()); reuse.setLabel(edgeLabelSourceToNew); reuse.setSourceId(triple.f1); reuse.setTargetId(triple.f0.getId()); out.collect(reuse); reuse.setId(GradoopId.get()); reuse.setLabel(edgeLabelNewToTarget); reuse.setSourceId(triple.f0.getId()); reuse.setTargetId(triple.f2); out.collect(reuse); } }
CreateEdgesFromTriple implements FlatMapFunction<Tuple3<V, GradoopId, GradoopId>, E>, ResultTypeQueryable<E> { @Override public void flatMap(Tuple3<V, GradoopId, GradoopId> triple, Collector<E> out) { reuse.setId(GradoopId.get()); reuse.setLabel(edgeLabelSourceToNew); reuse.setSourceId(triple.f1); reuse.setTargetId(triple.f0.getId()); out.collect(reuse); reuse.setId(GradoopId.get()); reuse.setLabel(edgeLabelNewToTarget); reuse.setSourceId(triple.f0.getId()); reuse.setTargetId(triple.f2); out.collect(reuse); } CreateEdgesFromTriple(EdgeFactory<E> factory, String edgeLabelSourceToNew,
    String edgeLabelNewToTarget); }
CreateEdgesFromTriple implements FlatMapFunction<Tuple3<V, GradoopId, GradoopId>, E>, ResultTypeQueryable<E> { @Override public void flatMap(Tuple3<V, GradoopId, GradoopId> triple, Collector<E> out) { reuse.setId(GradoopId.get()); reuse.setLabel(edgeLabelSourceToNew); reuse.setSourceId(triple.f1); reuse.setTargetId(triple.f0.getId()); out.collect(reuse); reuse.setId(GradoopId.get()); reuse.setLabel(edgeLabelNewToTarget); reuse.setSourceId(triple.f0.getId()); reuse.setTargetId(triple.f2); out.collect(reuse); } CreateEdgesFromTriple(EdgeFactory<E> factory, String edgeLabelSourceToNew,
    String edgeLabelNewToTarget); @Override void flatMap(Tuple3<V, GradoopId, GradoopId> triple, Collector<E> out); @Override TypeInformation<E> getProducedType(); }
CreateEdgesFromTriple implements FlatMapFunction<Tuple3<V, GradoopId, GradoopId>, E>, ResultTypeQueryable<E> { @Override public void flatMap(Tuple3<V, GradoopId, GradoopId> triple, Collector<E> out) { reuse.setId(GradoopId.get()); reuse.setLabel(edgeLabelSourceToNew); reuse.setSourceId(triple.f1); reuse.setTargetId(triple.f0.getId()); out.collect(reuse); reuse.setId(GradoopId.get()); reuse.setLabel(edgeLabelNewToTarget); reuse.setSourceId(triple.f0.getId()); reuse.setTargetId(triple.f2); out.collect(reuse); } CreateEdgesFromTriple(EdgeFactory<E> factory, String edgeLabelSourceToNew,
    String edgeLabelNewToTarget); @Override void flatMap(Tuple3<V, GradoopId, GradoopId> triple, Collector<E> out); @Override TypeInformation<E> getProducedType();  }