[#1354] Assign graph id to newly created elements

gradoop-data-integration/src/main/java/org/gradoop/dataintegration/transformation/impl/ExtractPropertyFromVertex.java

@@ -30,6 +30,8 @@
 import org.gradoop.flink.model.api.operators.UnaryGraphToGraphOperator;
 import org.gradoop.flink.model.impl.epgm.LogicalGraph;
 import org.gradoop.flink.model.impl.functions.epgm.ByLabel;
+import org.gradoop.flink.model.impl.functions.epgm.Id;
+import org.gradoop.flink.model.impl.functions.graphcontainment.AddToGraphBroadcast;
 import org.gradoop.flink.model.impl.functions.tuple.Value0Of2;
 
 import java.util.List;
@@ -152,8 +154,11 @@ public LogicalGraph execute(LogicalGraph logicalGraph) {
           newPropertyName));
     }
 
+
     DataSet<EPGMVertex> vertices = newVerticesAndOriginIds
       .map(new Value0Of2<>())
+      .map(new AddToGraphBroadcast<>())
+      .withBroadcastSet(logicalGraph.getGraphHead().map(new Id<>()), AddToGraphBroadcast.GRAPH_ID)
       .union(logicalGraph.getVertices());
 
     // the newly created vertices should be linked to the original vertices
@@ -162,6 +167,8 @@ public LogicalGraph execute(LogicalGraph logicalGraph) {
       edges = newVerticesAndOriginIds
         .flatMap(new CreateNewEdges(logicalGraph.getFactory().getEdgeFactory(), edgeDirection,
           edgeLabel))
+        .map(new AddToGraphBroadcast<>())
+        .withBroadcastSet(logicalGraph.getGraphHead().map(new Id<>()), AddToGraphBroadcast.GRAPH_ID)
         .union(edges);
     }
 

gradoop-data-integration/src/test/java/org/gradoop/dataintegration/transformation/impl/ExtractPropertyFromVertexTest.java

@@ -31,13 +31,13 @@
 import org.junit.Test;
 
 import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
-import java.util.Set;
-import java.util.Arrays;
 import java.util.Map;
-import java.util.HashMap;
-import java.util.Collections;
+import java.util.Set;
 
 /**
  * Tests for the {@link ExtractPropertyFromVertex} operator.
@@ -110,7 +110,7 @@ public void edgeCreationBidirectionalTest() throws Exception {
   /**
    * A private convenience method for easier testing of different setups in the edge creation process.
    *
-   * @param graph The input graph for the tests.
+   * @param graph     The input graph for the tests.
    * @param direction The edge direction the graph is tested for.
    * @throws Exception Is thrown if the process cant be executed properly.
    */
@@ -174,7 +174,8 @@ public void nonDeduplicationTest() throws Exception {
 
     // this operation should create 6 vertices with the property values:
     // 'Berlin', 3 times 'Dresden', 2 times 'Leipzig'
-    ExtractPropertyFromVertex extract = new ExtractPropertyFromVertex("Person", "city", "City", "name");
+    ExtractPropertyFromVertex extract = new ExtractPropertyFromVertex("Person", "city",
+      "City", "name");
     extract.setCondensation(false);
 
     LogicalGraph extractedGraph = social.callForGraph(extract);
@@ -229,4 +230,44 @@ public void listPropertyTest() throws Exception {
     Assert.assertTrue(properties.containsAll(Arrays.asList("m", "n", "x", "y", "z")));
   }
 
+  /**
+   * Tests graph head ids get assigned to newly created elements.
+   *
+   * @throws Exception If collect doesn't work as expected.
+   */
+  @Test
+  public void containGraphHeadIdTest() throws Exception {
+    LogicalGraph socialGraph = getSocialNetworkLoader().getLogicalGraph();
+
+    // this operation should create 6 vertices with the property values:
+    // 'Berlin', 3 times 'Dresden', 2 times 'Leipzig'
+    ExtractPropertyFromVertex extract = new ExtractPropertyFromVertex("Person", "city",
+      "City", "name", EdgeDirection.ORIGIN_TO_NEWVERTEX, "livesIn");
+
+    LogicalGraph extractedGraph = socialGraph.callForGraph(extract);
+
+    List<EPGMVertex> createdVertices = new ArrayList<>();
+    extractedGraph
+      .getVerticesByLabel("City")
+      .output(new LocalCollectionOutputFormat<>(createdVertices));
+
+    List<EPGMEdge> createdEdges = new ArrayList<>();
+    extractedGraph
+      .getEdgesByLabel("livesIn")
+      .output(new LocalCollectionOutputFormat<>(createdEdges));
+
+    getConfig().getExecutionEnvironment().execute();
+
+    // get only graph id from graph head
+    GradoopId socialGraphId = socialGraph.getGraphHead().collect().get(0).getId();
+
+    // check if the id of the original graph is assigned to the newly created elements
+    for (EPGMVertex vertex: createdVertices) {
+      Assert.assertTrue(vertex.getGraphIds().contains(socialGraphId));
+    }
+    for (EPGMEdge edge: createdEdges) {
+      Assert.assertTrue(edge.getGraphIds().contains(socialGraphId));
+    }
+  }
+
 }