@Test public void testCloning() throws Exception { FlinkAsciiGraphLoader loader = getLoaderFromString( "org:Ga{k : 0}[(:Va{k : 0, l : 0})-[:ea{l : 1}]->(:Va{l : 1, m : 2})]" ); List<GradoopId> expectedGraphHeadIds = Lists.newArrayList(); List<GradoopId> expectedVertexIds = Lists.newArrayList(); List<GradoopId> expectedEdgeIds = Lists.newArrayList(); LogicalGraph original = loader.getLogicalGraphByVariable("org"); original.getGraphHead().map(new Id<>()).output( new LocalCollectionOutputFormat<>(expectedGraphHeadIds)); original.getVertices().map(new Id<>()).output( new LocalCollectionOutputFormat<>(expectedVertexIds)); original.getEdges().map(new Id<>()).output( new LocalCollectionOutputFormat<>(expectedEdgeIds)); LogicalGraph result = original.copy(); collectAndAssertTrue(result.equalsByElementData(original)); List<GradoopId> resultGraphHeadIds = Lists.newArrayList(); List<GradoopId> resultVertexIds = Lists.newArrayList(); List<GradoopId> resultEdgeIds = Lists.newArrayList(); result.getGraphHead() .map(new Id<>()) .output(new LocalCollectionOutputFormat<>(resultGraphHeadIds)); result.getVertices() .map(new Id<>()) .output(new LocalCollectionOutputFormat<>(resultVertexIds)); result.getEdges() .map(new Id<>()) .output(new LocalCollectionOutputFormat<>(resultEdgeIds)); List<GradoopIdSet> resultGraphIds = Lists.newArrayList(); result.getVertices() .map(new ExpandGraphsToIdSet<>()) .union(result.getEdges() .map(new ExpandGraphsToIdSet<>())) .union(result.getGraphHead() .map(new IdAsIdSet<>())) .reduce(new IdSetCombiner()) .output(new LocalCollectionOutputFormat<>(resultGraphIds)); getExecutionEnvironment().execute(); assertTrue("elements in multiple graphs", resultGraphIds.size() == 1); assertTrue("wrong number of graph heads", expectedGraphHeadIds.size() == resultGraphHeadIds.size()); assertTrue("wrong number of vertices", expectedVertexIds.size() == resultVertexIds.size()); assertTrue("wrong number of edges", expectedEdgeIds.size() == resultEdgeIds.size()); validateIdInequality(expectedGraphHeadIds, resultGraphHeadIds); validateIdInequality(expectedVertexIds, resultVertexIds); validateIdInequality(expectedEdgeIds, resultEdgeIds); }
@Override public LG execute(LG graph) { DataSet<G> graphHead = graph.getGraphHead() .map(new Clone<>()); DataSet<GradoopId> graphId = graphHead.map(new Id<>()); DataSet<Tuple2<V, GradoopId>> vertexTuple = graph.getVertices() .map(new PairElementWithNewId<>()); DataSet<Tuple2<GradoopId, GradoopId>> vertexIdTuple = vertexTuple .map(new Value0Of2ToId<>()); DataSet<V> vertices = vertexTuple .map(new ElementIdUpdater<>()) .map(new ElementGraphUpdater<>()) .withBroadcastSet(graphId, ElementGraphUpdater.GRAPHID); DataSet<E> edges = graph.getEdges() .map(new Clone<>()) .join(vertexIdTuple) .where(new SourceId<>()).equalTo(0) .with(new EdgeSourceUpdateJoin<>()) .join(vertexIdTuple) .where(new TargetId<>()).equalTo(0) .with(new EdgeTargetUpdateJoin<>()) .map(new ElementGraphUpdater<>()) .withBroadcastSet(graphId, ElementGraphUpdater.GRAPHID); return graph.getFactory().fromDataSets(graphHead, vertices, edges); }
Cloning implements UnaryBaseGraphToBaseGraphOperator<LG> { @Override public LG execute(LG graph) { DataSet<G> graphHead = graph.getGraphHead() .map(new Clone<>()); DataSet<GradoopId> graphId = graphHead.map(new Id<>()); DataSet<Tuple2<V, GradoopId>> vertexTuple = graph.getVertices() .map(new PairElementWithNewId<>()); DataSet<Tuple2<GradoopId, GradoopId>> vertexIdTuple = vertexTuple .map(new Value0Of2ToId<>()); DataSet<V> vertices = vertexTuple .map(new ElementIdUpdater<>()) .map(new ElementGraphUpdater<>()) .withBroadcastSet(graphId, ElementGraphUpdater.GRAPHID); DataSet<E> edges = graph.getEdges() .map(new Clone<>()) .join(vertexIdTuple) .where(new SourceId<>()).equalTo(0) .with(new EdgeSourceUpdateJoin<>()) .join(vertexIdTuple) .where(new TargetId<>()).equalTo(0) .with(new EdgeTargetUpdateJoin<>()) .map(new ElementGraphUpdater<>()) .withBroadcastSet(graphId, ElementGraphUpdater.GRAPHID); return graph.getFactory().fromDataSets(graphHead, vertices, edges); } }
Cloning implements UnaryBaseGraphToBaseGraphOperator<LG> { @Override public LG execute(LG graph) { DataSet<G> graphHead = graph.getGraphHead() .map(new Clone<>()); DataSet<GradoopId> graphId = graphHead.map(new Id<>()); DataSet<Tuple2<V, GradoopId>> vertexTuple = graph.getVertices() .map(new PairElementWithNewId<>()); DataSet<Tuple2<GradoopId, GradoopId>> vertexIdTuple = vertexTuple .map(new Value0Of2ToId<>()); DataSet<V> vertices = vertexTuple .map(new ElementIdUpdater<>()) .map(new ElementGraphUpdater<>()) .withBroadcastSet(graphId, ElementGraphUpdater.GRAPHID); DataSet<E> edges = graph.getEdges() .map(new Clone<>()) .join(vertexIdTuple) .where(new SourceId<>()).equalTo(0) .with(new EdgeSourceUpdateJoin<>()) .join(vertexIdTuple) .where(new TargetId<>()).equalTo(0) .with(new EdgeTargetUpdateJoin<>()) .map(new ElementGraphUpdater<>()) .withBroadcastSet(graphId, ElementGraphUpdater.GRAPHID); return graph.getFactory().fromDataSets(graphHead, vertices, edges); }  }
Cloning implements UnaryBaseGraphToBaseGraphOperator<LG> { @Override public LG execute(LG graph) { DataSet<G> graphHead = graph.getGraphHead() .map(new Clone<>()); DataSet<GradoopId> graphId = graphHead.map(new Id<>()); DataSet<Tuple2<V, GradoopId>> vertexTuple = graph.getVertices() .map(new PairElementWithNewId<>()); DataSet<Tuple2<GradoopId, GradoopId>> vertexIdTuple = vertexTuple .map(new Value0Of2ToId<>()); DataSet<V> vertices = vertexTuple .map(new ElementIdUpdater<>()) .map(new ElementGraphUpdater<>()) .withBroadcastSet(graphId, ElementGraphUpdater.GRAPHID); DataSet<E> edges = graph.getEdges() .map(new Clone<>()) .join(vertexIdTuple) .where(new SourceId<>()).equalTo(0) .with(new EdgeSourceUpdateJoin<>()) .join(vertexIdTuple) .where(new TargetId<>()).equalTo(0) .with(new EdgeTargetUpdateJoin<>()) .map(new ElementGraphUpdater<>()) .withBroadcastSet(graphId, ElementGraphUpdater.GRAPHID); return graph.getFactory().fromDataSets(graphHead, vertices, edges); }  @Override LG execute(LG graph); }
Cloning implements UnaryBaseGraphToBaseGraphOperator<LG> { @Override public LG execute(LG graph) { DataSet<G> graphHead = graph.getGraphHead() .map(new Clone<>()); DataSet<GradoopId> graphId = graphHead.map(new Id<>()); DataSet<Tuple2<V, GradoopId>> vertexTuple = graph.getVertices() .map(new PairElementWithNewId<>()); DataSet<Tuple2<GradoopId, GradoopId>> vertexIdTuple = vertexTuple .map(new Value0Of2ToId<>()); DataSet<V> vertices = vertexTuple .map(new ElementIdUpdater<>()) .map(new ElementGraphUpdater<>()) .withBroadcastSet(graphId, ElementGraphUpdater.GRAPHID); DataSet<E> edges = graph.getEdges() .map(new Clone<>()) .join(vertexIdTuple) .where(new SourceId<>()).equalTo(0) .with(new EdgeSourceUpdateJoin<>()) .join(vertexIdTuple) .where(new TargetId<>()).equalTo(0) .with(new EdgeTargetUpdateJoin<>()) .map(new ElementGraphUpdater<>()) .withBroadcastSet(graphId, ElementGraphUpdater.GRAPHID); return graph.getFactory().fromDataSets(graphHead, vertices, edges); }  @Override LG execute(LG graph);  }