dhp-workflows/dhp-graph-mapper/src/main/java/eu/dnetlib/dhp/oa/graph/raw/PatchRelationsApplication.java

@@ -1,10 +1,12 @@
+
 package eu.dnetlib.dhp.oa.graph.raw;
 
-import com.fasterxml.jackson.databind.ObjectMapper;
+import static eu.dnetlib.dhp.common.SparkSessionSupport.runWithSparkSession;
-import eu.dnetlib.dhp.application.ArgumentApplicationParser;
+
-import eu.dnetlib.dhp.oa.graph.dump.Utils;
+import java.io.FileNotFoundException;
-import eu.dnetlib.dhp.oa.graph.raw.common.RelationIdMapping;
+import java.util.Objects;
-import eu.dnetlib.dhp.schema.oaf.Relation;
+import java.util.Optional;
+
 import org.apache.commons.io.IOUtils;
 import org.apache.spark.SparkConf;
 import org.apache.spark.api.java.function.FilterFunction;
@@ -15,14 +17,15 @@ import org.apache.spark.sql.SaveMode;
 import org.apache.spark.sql.SparkSession;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+import eu.dnetlib.dhp.application.ArgumentApplicationParser;
+import eu.dnetlib.dhp.oa.graph.dump.Utils;
+import eu.dnetlib.dhp.oa.graph.raw.common.RelationIdMapping;
+import eu.dnetlib.dhp.schema.oaf.Relation;
 import scala.Tuple2;
 
-import java.io.FileNotFoundException;
-import java.util.Objects;
-import java.util.Optional;
-
-import static eu.dnetlib.dhp.common.SparkSessionSupport.runWithSparkSession;
-
 public class PatchRelationsApplication {
 
 	private static final Logger log = LoggerFactory.getLogger(PatchRelationsApplication.class);
@@ -33,12 +36,12 @@ public class PatchRelationsApplication {
 		final ArgumentApplicationParser parser = new ArgumentApplicationParser(
 			IOUtils
 				.toString(
-                                Optional.ofNullable(
+					Optional
+						.ofNullable(
 							PatchRelationsApplication.class
 								.getResourceAsStream(
 									"/eu/dnetlib/dhp/oa/graph/patch_relations_parameters.json"))
-                                .orElseThrow(FileNotFoundException::new)
+						.orElseThrow(FileNotFoundException::new)));
-                        ));
 		parser.parseArgument(args);
 
 		final Boolean isSparkSessionManaged = Optional
@@ -72,7 +75,8 @@ public class PatchRelationsApplication {
 	 * @param workingDir intermediate storage location
 	 * @param idMappingPath dataset providing the old -> new identifier mapping
 	 */
-    private static void patchRelations(final SparkSession spark, final String graphBasePath, final String workingDir, final String idMappingPath) {
+	private static void patchRelations(final SparkSession spark, final String graphBasePath, final String workingDir,
+		final String idMappingPath) {
 
 		final String relationPath = graphBasePath + "/relation";
 
@@ -82,19 +86,23 @@ public class PatchRelationsApplication {
 		log.info("relations: {}", rels.count());
 		log.info("idMapping: {}", idMapping.count());
 
-        rels
+		final Dataset<Relation> bySource = rels
 			.joinWith(idMapping, rels.col("source").equalTo(idMapping.col("oldId")), "left")
 			.map((MapFunction<Tuple2<Relation, RelationIdMapping>, Relation>) t -> {
 				final Relation r = t._1();
-                    Optional.ofNullable(t._2())
+				Optional
+					.ofNullable(t._2())
 					.map(RelationIdMapping::getNewId)
 					.ifPresent(r::setSource);
 				return r;
-                }, Encoders.bean(Relation.class))
+			}, Encoders.bean(Relation.class));
-                .joinWith(idMapping, rels.col("target").equalTo(idMapping.col("oldId")), "left")
+
+		bySource
+			.joinWith(idMapping, bySource.col("target").equalTo(idMapping.col("oldId")), "left")
 			.map((MapFunction<Tuple2<Relation, RelationIdMapping>, Relation>) t -> {
 				final Relation r = t._1();
-                    Optional.ofNullable(t._2())
+				Optional
+					.ofNullable(t._2())
 					.map(RelationIdMapping::getNewId)
 					.ifPresent(r::setTarget);
 				return r;
@@ -107,12 +115,13 @@ public class PatchRelationsApplication {
 			.option("compression", "gzip")
 			.text(workingDir);
 
-        spark.read().textFile(workingDir)
+		spark
+			.read()
+			.textFile(workingDir)
 			.write()
 			.mode(SaveMode.Overwrite)
 			.option("compression", "gzip")
 			.text(relationPath);
 	}
 
-
 }

dhp-workflows/dhp-graph-mapper/src/main/java/eu/dnetlib/dhp/oa/graph/raw/common/RelationIdMapping.java

@@ -1,3 +1,4 @@
+
 package eu.dnetlib.dhp.oa.graph.raw.common;
 
 public class RelationIdMapping {

dhp-workflows/dhp-graph-mapper/src/test/java/eu/dnetlib/dhp/oa/graph/raw/PatchRelationApplicationTest.java

@@ -1,21 +1,27 @@
+
 package eu.dnetlib.dhp.oa.graph.raw;
 
-import com.fasterxml.jackson.databind.ObjectMapper;
+import static org.junit.jupiter.api.Assertions.assertEquals;
-import org.apache.commons.io.FileUtils;
-import org.apache.commons.io.IOUtils;
-import org.apache.spark.SparkConf;
-import org.apache.spark.sql.SparkSession;
-import org.junit.jupiter.api.AfterAll;
-import org.junit.jupiter.api.BeforeAll;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
 import java.nio.file.Files;
 import java.nio.file.Path;
-import java.util.Optional;
+import java.util.List;
+
+import org.apache.commons.io.FileUtils;
+import org.apache.spark.SparkConf;
+import org.apache.spark.api.java.function.MapFunction;
+import org.apache.spark.sql.Encoders;
+import org.apache.spark.sql.SparkSession;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+import eu.dnetlib.dhp.schema.oaf.Relation;
 
 public class PatchRelationApplicationTest {
 
@@ -50,21 +56,15 @@ public class PatchRelationApplicationTest {
 			.config(conf)
 			.getOrCreate();
 
-        FileUtils.copyInputStreamToFile(
+		FileUtils
+			.copyInputStreamToFile(
 				PatchRelationApplicationTest.class.getResourceAsStream("id_mapping.json"),
-                workingDir.resolve(ID_MAPPING_PATH).toFile()
+				workingDir.resolve(ID_MAPPING_PATH).toFile());
-        );
 
-        FileUtils.copyInputStreamToFile(
+		FileUtils
+			.copyInputStreamToFile(
 				PatchRelationApplicationTest.class.getResourceAsStream("relations_to_patch.json"),
-                workingDir.resolve("graphBasePath/relation/rels.json").toFile()
+				workingDir.resolve("graphBasePath/relation/rels.json").toFile());
-        );
-
-    }
-
-    @BeforeEach
-    public void setUp() throws IOException {
-
 
 	}
 
@@ -77,13 +77,39 @@ public class PatchRelationApplicationTest {
 	@Test
 	public void testPatchRelationApplication() throws Exception {
 
+		final String graphBasePath = workingDir.toString() + "/graphBasePath";
 		PatchRelationsApplication.main(new String[] {
 			"-isSparkSessionManaged", Boolean.FALSE.toString(),
-                "-graphBasePath", workingDir.toString() + "/graphBasePath",
+			"-graphBasePath", graphBasePath,
 			"-workingDir", workingDir.toString() + "/workingDir",
 			"-idMappingPath", workingDir.toString() + "/" + ID_MAPPING_PATH
 		});
 
+		final List<Relation> rels = spark
+			.read()
+			.textFile(graphBasePath + "/relation")
+			.map(
+				(MapFunction<String, Relation>) s -> OBJECT_MAPPER.readValue(s, Relation.class),
+				Encoders.bean(Relation.class))
+			.collectAsList();
+
+		assertEquals(6, rels.size());
+
+		assertEquals(0, getCount(rels, "1a"), "should be patched to 1b");
+		assertEquals(0, getCount(rels, "2a"), "should be patched to 2b");
+
+		assertEquals(2, getCount(rels, "10a"), "not included in patching");
+		assertEquals(2, getCount(rels, "20a"), "not included in patching");
+
+		assertEquals(2, getCount(rels, "15a"), "not included in patching");
+		assertEquals(2, getCount(rels, "25a"), "not included in patching");
+
+		assertEquals(2, getCount(rels, "1b"), "patched from 1a");
+		assertEquals(2, getCount(rels, "2b"), "patched from 2a");
+	}
+
+	private long getCount(List<Relation> rels, final String id) {
+		return rels.stream().filter(r -> r.getSource().equals(id) || r.getTarget().equals(id)).count();
 	}
 
 }