fixed typo

dhp-schemas/src/main/java/eu/dnetlib/dhp/schema/common/ModelSupport.java

@@ -45,6 +45,30 @@ public class ModelSupport {
 		entityTypes.put(EntityType.publication, Publication.class);
 	}
 
+	/**
+	 * Enumerated the classes of the OafEntities that are also Results
+	 */
+	public static final Map<String, Class> resultTypes = Maps.newHashMap();
+
+	static {
+		resultTypes.put("publication", Publication.class);
+		resultTypes.put("dataset", Dataset.class);
+		resultTypes.put("software", Software.class);
+		resultTypes.put("otherresearchproduct", OtherResearchProduct.class);
+	}
+
+	/**
+	 *
+	 */
+
+	public static final Map<Class, String> resultRelationType = Maps.newHashMap();
+
+	static {
+		resultRelationType.put(Result.class, "result");
+		resultRelationType.put(Relation.class, "relation");
+
+	}
+
 	public static final Map<String, Class> oafTypes = Maps.newHashMap();
 
 	static {

dhp-workflows/dhp-actionset-remapping/pom.xml

@@ -0,0 +1,43 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <artifactId>dhp-workflows</artifactId>
+        <groupId>eu.dnetlib.dhp</groupId>
+        <version>1.2.4-SNAPSHOT</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>dhp-actionset-remapping</artifactId>
+
+    <dependencies>
+
+    <dependency>
+        <groupId>org.apache.spark</groupId>
+        <artifactId>spark-core_2.11</artifactId>
+    </dependency>
+    <dependency>
+        <groupId>org.apache.spark</groupId>
+        <artifactId>spark-sql_2.11</artifactId>
+    </dependency>
+
+    <dependency>
+        <groupId>eu.dnetlib.dhp</groupId>
+        <artifactId>dhp-common</artifactId>
+        <version>${project.version}</version>
+    </dependency>
+
+    <dependency>
+        <groupId>eu.dnetlib.dhp</groupId>
+        <artifactId>dhp-schemas</artifactId>
+        <version>${project.version}</version>
+    </dependency>
+        <dependency>
+            <groupId>eu.dnetlib.dhp</groupId>
+            <artifactId>dhp-aggregation</artifactId>
+            <version>1.2.4-SNAPSHOT</version>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+</project>

dhp-workflows/dhp-actionset-remapping/src/main/java/eu/dnetlib/dhp/actionset/common/ASResultInfo.java

@@ -0,0 +1,45 @@
+
+package eu.dnetlib.dhp.actionset.common;
+
+import java.io.Serializable;
+import java.util.List;
+import java.util.stream.Collectors;
+
+public class ASResultInfo implements Serializable {
+	private String id;
+	private String type; // result or relation
+	private List<InferenceInfo> value; // the community or the project
+
+	public String getId() {
+		return id;
+	}
+
+	public String getType() {
+		return type;
+	}
+
+	public void setType(String type) {
+		this.type = type;
+	}
+
+	public List<InferenceInfo> getValue() {
+		return value;
+	}
+
+	public void setValue(List<InferenceInfo> value) {
+		this.value = value;
+	}
+
+	public void setId(String id) {
+		this.id = id;
+	}
+
+	public static ASResultInfo copy(ASResultInfo as) {
+		ASResultInfo ar = new ASResultInfo();
+		ar.id = as.id;
+		ar.type = as.type;
+		ar.value = as.value.stream().map(InferenceInfo::copy).collect(Collectors.toList());
+		return ar;
+	}
+
+}

dhp-workflows/dhp-actionset-remapping/src/main/java/eu/dnetlib/dhp/actionset/common/ActionSet.java

@@ -0,0 +1,43 @@
+
+package eu.dnetlib.dhp.actionset.common;
+
+import java.io.Serializable;
+
+public class ActionSet implements Serializable {
+	private String name;
+	private String rawset;
+	private String directory;
+
+	public String getDirectory() {
+		return directory;
+	}
+
+	public void setDirectory(String directory) {
+		this.directory = directory;
+	}
+
+	public String getName() {
+		return name;
+	}
+
+	public void setName(String name) {
+		this.name = name;
+	}
+
+	public String getRawset() {
+		return rawset;
+	}
+
+	public void setRawset(String rawset) {
+		this.rawset = rawset;
+	}
+
+	public static ActionSet newInstance(String name, String rawset, String directory) {
+		ActionSet as = new ActionSet();
+		as.name = name;
+		as.rawset = rawset;
+		as.directory = directory;
+		return as;
+
+	}
+}

dhp-workflows/dhp-actionset-remapping/src/main/java/eu/dnetlib/dhp/actionset/common/Common.java

@@ -0,0 +1,30 @@
+
+package eu.dnetlib.dhp.actionset.common;
+
+import java.io.Serializable;
+
+import org.apache.spark.api.java.function.MapFunction;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Encoders;
+import org.apache.spark.sql.SparkSession;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+import eu.dnetlib.dhp.common.HdfsSupport;
+
+public class Common implements Serializable {
+
+	public static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+
+	public static <R> Dataset<R> readPath(
+		SparkSession spark, String inputPath, Class<R> clazz) {
+		return spark
+			.read()
+			.textFile(inputPath)
+			.map((MapFunction<String, R>) value -> OBJECT_MAPPER.readValue(value, clazz), Encoders.bean(clazz));
+	}
+
+	public static void removeOutputDir(SparkSession spark, String path) {
+		HdfsSupport.remove(path, spark.sparkContext().hadoopConfiguration());
+	}
+}

dhp-workflows/dhp-actionset-remapping/src/main/java/eu/dnetlib/dhp/actionset/common/InferenceInfo.java

@@ -0,0 +1,43 @@
+
+package eu.dnetlib.dhp.actionset.common;
+
+import java.io.Serializable;
+
+public class InferenceInfo implements Serializable {
+	private String value;
+	private String trust;
+	private String inference_provenance;
+
+	public String getValue() {
+		return value;
+	}
+
+	public void setValue(String value) {
+		this.value = value;
+	}
+
+	public String getTrust() {
+		return trust;
+	}
+
+	public void setTrust(String trust) {
+		this.trust = trust;
+	}
+
+	public String getInference_provenance() {
+		return inference_provenance;
+	}
+
+	public void setInference_provenance(String inference_provenance) {
+		this.inference_provenance = inference_provenance;
+	}
+
+	public static InferenceInfo copy(InferenceInfo ii) {
+		InferenceInfo iinfo = new InferenceInfo();
+		iinfo.value = ii.value;
+		iinfo.trust = ii.trust;
+		iinfo.inference_provenance = ii.inference_provenance;
+		return iinfo;
+	}
+
+}

dhp-workflows/dhp-actionset-remapping/src/main/java/eu/dnetlib/dhp/actionset/common/PrepareInfo.java

@@ -0,0 +1,158 @@
+
+package eu.dnetlib.dhp.actionset.common;
+
+import static eu.dnetlib.dhp.common.SparkSessionSupport.runWithSparkSession;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Objects;
+import java.util.stream.Collectors;
+
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.io.compress.GzipCodec;
+import org.apache.spark.SparkConf;
+import org.apache.spark.api.java.JavaRDD;
+import org.apache.spark.api.java.JavaSparkContext;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Encoders;
+import org.apache.spark.sql.SaveMode;
+import org.apache.spark.sql.SparkSession;
+
+import com.google.gson.Gson;
+
+import eu.dnetlib.dhp.schema.action.AtomicAction;
+import eu.dnetlib.dhp.schema.common.ModelSupport;
+import eu.dnetlib.dhp.schema.oaf.Relation;
+import eu.dnetlib.dhp.schema.oaf.Result;
+
+public class PrepareInfo implements Serializable {
+	private String outputPath;
+	private Boolean isSparkSessionManaged;
+	private String asInputPath;
+	private String inputGraphPath;
+	private List<String> actionSetActive;
+	private List<ActionSet> actionSetList;
+
+	public PrepareInfo(Boolean isSparkSessionManaged,
+		String outputPath,
+		String asInputPath,
+		String inputGraphPath,
+		List<String> actionSetActive,
+		List<ActionSet> actionSetList) {
+		this.isSparkSessionManaged = isSparkSessionManaged;
+		this.outputPath = outputPath;
+		this.inputGraphPath = inputGraphPath;
+		this.asInputPath = asInputPath;
+		this.actionSetActive = actionSetActive;
+		this.actionSetList = actionSetList;
+	}
+
+	public void run() {
+
+		SparkConf conf = new SparkConf();
+
+		runWithSparkSession(
+			conf,
+			isSparkSessionManaged,
+			spark -> {
+				Common.removeOutputDir(spark, outputPath);
+				exec(
+					spark, asInputPath, inputGraphPath, outputPath,
+					actionSetList.stream().map(as -> {
+						if (actionSetActive.contains(as.getName())) {
+							return as;
+						}
+						return null;
+					})
+						.filter(Objects::nonNull)
+						.collect(Collectors.toList()));
+			});
+	}
+
+	private static void exec(SparkSession spark, String asInputPath, String graphInputPath, String outputPath,
+		List<ActionSet> actionSets) {
+
+		Dataset<Relation> relation = Common.readPath(spark, graphInputPath + "/relation", Relation.class);
+
+		actionSets
+			.forEach(
+				as -> resolveActionSet(
+					spark, asInputPath + "/" + as.getDirectory() + "/" + as.getRawset(),
+					outputPath));
+
+		relation.createOrReplaceTempView("relation");
+
+		spark
+			.sql(
+				"SELECT source dedupId, collect_set(target) as merges " +
+					"FROM relation " +
+					"WHERE relclass = 'merges' " +
+					"GROUP BY source")
+			.write()
+			.mode(SaveMode.Overwrite)
+			.option("compression", "gzip")
+			.json(outputPath + "/relation/");
+
+	}
+
+	private static void resolveActionSet(SparkSession spark, String asInputPath, String outputPath) {
+		final JavaSparkContext sc = JavaSparkContext.fromSparkContext(spark.sparkContext());
+
+		spark
+			.createDataset(
+				sc
+					.sequenceFile(asInputPath, Text.class, Text.class)
+					.map(a -> Common.OBJECT_MAPPER.readValue(a._2().toString(), AtomicAction.class))
+					.map(aa -> getAsResultInfo(aa))
+					.filter(Objects::nonNull)
+					.rdd(),
+				Encoders.bean(ASResultInfo.class))
+			.write()
+			.option("compression", "gzip")
+			.mode(SaveMode.Append)
+			.json(outputPath + "/actionset/");
+
+	}
+
+	private static ASResultInfo getAsResultInfo(AtomicAction aa) {
+		ASResultInfo ri = new ASResultInfo();
+		if (ModelSupport.resultRelationType.get(aa.getClazz()).equals("result")) {
+
+			Result result = (Result) aa.getPayload();
+			ri.setId(result.getId());
+			ri.setType("result");
+			List<InferenceInfo> inferenceInfoList = new ArrayList<>();
+
+			result.getContext().forEach(c -> {
+				String id = c.getId();
+				c.getDataInfo().forEach(di -> {
+					InferenceInfo ii = new InferenceInfo();
+					ii.setValue(id);
+					ii.setTrust(di.getTrust());
+					ii.setInference_provenance(di.getInferenceprovenance());
+					inferenceInfoList.add(ii);
+				});
+			});
+
+			ri.setValue(inferenceInfoList);
+
+		} else {
+			Relation rel = (Relation) aa.getPayload();
+			if (rel.getSource().startsWith("50|")) {
+				ri.setId(rel.getSource());
+				ri.setType("relation");
+				InferenceInfo ii = new InferenceInfo();
+				ii.setInference_provenance(rel.getDataInfo().getInferenceprovenance());
+				ii.setTrust(rel.getDataInfo().getTrust());
+				ii.setValue(rel.getTarget());
+				ri.setValue(Arrays.asList(ii));
+			} else {
+				return null;
+			}
+		}
+		return ri;
+	}
+
+}

dhp-workflows/dhp-actionset-remapping/src/main/java/eu/dnetlib/dhp/actionset/common/QueryInformationSystem.java

@@ -0,0 +1,42 @@
+
+package eu.dnetlib.dhp.actionset.common;
+
+import java.io.Serializable;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import eu.dnetlib.dhp.utils.ISLookupClientFactory;
+import eu.dnetlib.enabling.is.lookup.rmi.ISLookUpException;
+import eu.dnetlib.enabling.is.lookup.rmi.ISLookUpService;
+
+public class QueryInformationSystem implements Serializable {
+
+	private static final String ACTION_MANAGER_PATH_QUERY = "for $x in " +
+		"collection(' /db/DRIVER/ServiceResources/ActionManagerServiceResourceType') " +
+		"return data($x//PROPERTY[./@key='basePath']/@value)";
+
+	private static final String ACTION_SET_QUERY = "for $x in " +
+		"collection('/db/DRIVER/ActionManagerSetDSResources/ActionManagerSetDSResourceType') " +
+		"return concat (data($x//SET/@id),'@@',data($x//LATEST/@id),'@@',data($x//SET/@directory))";
+
+	public static String getActionManagerPath(final String isLookupUrl)
+		throws ISLookUpException {
+		ISLookUpService isLookUp = ISLookupClientFactory.getLookUpService(isLookupUrl);
+		return isLookUp.quickSearchProfile(ACTION_MANAGER_PATH_QUERY).get(0);
+
+	}
+
+	public static List<ActionSet> getActionSet(final String isLookupUrl)
+		throws ISLookUpException {
+		ISLookUpService isLookUp = ISLookupClientFactory.getLookUpService(isLookupUrl);
+		return isLookUp
+			.quickSearchProfile(ACTION_SET_QUERY)
+			.stream()
+			.map(res -> {
+				String[] tmp = res.split("@@");
+				return ActionSet.newInstance(tmp[0], tmp[1], tmp[2]);
+			})
+			.collect(Collectors.toList());
+	}
+
+}

dhp-workflows/dhp-actionset-remapping/src/main/java/eu/dnetlib/dhp/actionset/common/RelationMerges.java

@@ -0,0 +1,26 @@
+
+package eu.dnetlib.dhp.actionset.common;
+
+import java.io.Serializable;
+import java.util.List;
+
+public class RelationMerges implements Serializable {
+	private String dedupId;
+	private List<String> merges;
+
+	public String getDedupId() {
+		return dedupId;
+	}
+
+	public void setDedupId(String dedupId) {
+		this.dedupId = dedupId;
+	}
+
+	public List<String> getMerges() {
+		return merges;
+	}
+
+	public void setMerges(List<String> merges) {
+		this.merges = merges;
+	}
+}

dhp-workflows/dhp-actionset-remapping/src/main/java/eu/dnetlib/dhp/actionset/common/SparkExpandResultInfo.java

@@ -0,0 +1,92 @@
+
+package eu.dnetlib.dhp.actionset.common;
+
+import static eu.dnetlib.dhp.common.SparkSessionSupport.runWithSparkSession;
+
+import java.io.Serializable;
+import java.util.*;
+import java.util.stream.Collectors;
+
+import org.apache.commons.io.IOUtils;
+import org.apache.spark.SparkConf;
+import org.apache.spark.api.java.function.FlatMapFunction;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Encoders;
+import org.apache.spark.sql.SaveMode;
+import org.apache.spark.sql.SparkSession;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import eu.dnetlib.dhp.application.ArgumentApplicationParser;
+import scala.Tuple2;
+
+public class SparkExpandResultInfo implements Serializable {
+	private static final Logger log = LoggerFactory.getLogger(SparkExpandResultInfo.class);
+
+	public static void main(String[] args) throws Exception {
+		String jsonConfiguration = IOUtils
+			.toString(
+				SparkExpandResultInfo.class
+					.getResourceAsStream(
+						"/eu/dnetlib/dhp/actionset/common/input_expand_parameters.json"));
+
+		final ArgumentApplicationParser parser = new ArgumentApplicationParser(jsonConfiguration);
+		parser.parseArgument(args);
+
+		Boolean isSparkSessionManaged = Optional
+			.ofNullable(parser.get("isSparkSessionManaged"))
+			.map(Boolean::valueOf)
+			.orElse(Boolean.TRUE);
+		log.info("isSparkSessionManaged: {}", isSparkSessionManaged);
+
+		final String outputPath = parser.get("outputPath");
+		log.info("outputPath: {}", outputPath);
+
+		final String inputPath = parser.get("inputPath");
+		log.info("inputPath: {}", inputPath);
+
+		SparkConf conf = new SparkConf();
+
+		runWithSparkSession(
+			conf,
+			isSparkSessionManaged,
+			spark -> {
+				Common.removeOutputDir(spark, outputPath);
+				exec(spark, inputPath, outputPath);
+			});
+	}
+
+	private static void exec(SparkSession spark, String inputPath, String outputPath) {
+		Dataset<RelationMerges> rel = Common.readPath(spark, inputPath + "/relation", RelationMerges.class);
+
+		Dataset<ASResultInfo> asResultInfo = Common.readPath(spark, inputPath + "/actionset", ASResultInfo.class);
+
+		asResultInfo
+			.joinWith(rel, asResultInfo.col("id").equalTo(rel.col("dedupId")), "left")
+			.flatMap((FlatMapFunction<Tuple2<ASResultInfo, RelationMerges>, ASResultInfo>) value -> {
+				ASResultInfo ri = value._1();
+				if (Objects.isNull(value._2())) {
+					return Arrays.asList(ri).iterator();
+				}
+
+				return value
+					._2()
+					.getMerges()
+					.stream()
+					.map(res -> {
+						ASResultInfo copy = ASResultInfo.copy(ri);
+						copy.setId(res);
+						return copy;
+					})
+					.collect(Collectors.toList())
+					.iterator();
+
+			}, Encoders.bean(ASResultInfo.class))
+			.write()
+			.mode(SaveMode.Overwrite)
+			.option("compression", "gzip")
+			.json(outputPath);
+
+	}
+
+}

dhp-workflows/dhp-actionset-remapping/src/main/java/eu/dnetlib/dhp/actionset/common/SparkPrepareInfo.java

@@ -0,0 +1,55 @@
+
+package eu.dnetlib.dhp.actionset.common;
+
+import java.io.Serializable;
+import java.util.*;
+
+import org.apache.commons.io.IOUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.gson.Gson;
+
+import eu.dnetlib.dhp.application.ArgumentApplicationParser;
+
+public class SparkPrepareInfo implements Serializable {
+
+	private static final Logger log = LoggerFactory.getLogger(SparkPrepareInfo.class);
+
+	public static void main(String[] args) throws Exception {
+		String jsonConfiguration = IOUtils
+			.toString(
+				SparkPrepareInfo.class
+					.getResourceAsStream(
+						"/eu/dnetlib/dhp/actionset/common/input_prepare_parameters.json"));
+
+		final ArgumentApplicationParser parser = new ArgumentApplicationParser(jsonConfiguration);
+		parser.parseArgument(args);
+
+		Boolean isSparkSessionManaged = Optional
+			.ofNullable(parser.get("isSparkSessionManaged"))
+			.map(Boolean::valueOf)
+			.orElse(Boolean.TRUE);
+		log.info("isSparkSessionManaged: {}", isSparkSessionManaged);
+
+		final String isLookUpUrl = parser.get("isLookUpUrl");
+		log.info("isLookUpUrl: {}", isLookUpUrl);
+
+		final String outputPath = parser.get("outputPath");
+		log.info("outputPath: {}", outputPath);
+
+		final String inputPath = parser.get("inputPath");
+
+		final List<String> actionSetActive = new Gson().fromJson(parser.get("actionSets"), List.class);
+
+		final String asInputPath = QueryInformationSystem.getActionManagerPath(isLookUpUrl);
+
+		final List<ActionSet> actionSetList = QueryInformationSystem.getActionSet(isLookUpUrl);
+
+		PrepareInfo prepareInfo = new PrepareInfo(isSparkSessionManaged, outputPath, asInputPath, inputPath,
+			actionSetActive, actionSetList);
+
+		prepareInfo.run();
+	}
+
+}

dhp-workflows/dhp-actionset-remapping/src/main/java/eu/dnetlib/dhp/actionset/stable/ResultPid.java

@@ -0,0 +1,25 @@
+
+package eu.dnetlib.dhp.actionset.stable;
+
+import java.io.Serializable;
+
+public class ResultPid implements Serializable {
+	private String resultId;
+	private String doi;
+
+	public String getResultId() {
+		return resultId;
+	}
+
+	public void setResultId(String resultId) {
+		this.resultId = resultId;
+	}
+
+	public String getDoi() {
+		return doi;
+	}
+
+	public void setDoi(String doi) {
+		this.doi = doi;
+	}
+}

dhp-workflows/dhp-actionset-remapping/src/main/java/eu/dnetlib/dhp/actionset/stable/SparkRedistributeIISRelations.java

@@ -0,0 +1,136 @@
+
+package eu.dnetlib.dhp.actionset.stable;
+
+import static eu.dnetlib.dhp.common.SparkSessionSupport.runWithSparkSession;
+
+import java.io.Serializable;
+import java.util.*;
+import java.util.stream.Collectors;
+
+import org.apache.commons.io.IOUtils;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.mapred.SequenceFileOutputFormat;
+import org.apache.spark.SparkConf;
+import org.apache.spark.api.java.JavaPairRDD;
+import org.apache.spark.api.java.JavaRDD;
+import org.apache.spark.api.java.function.FlatMapFunction;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Encoders;
+import org.apache.spark.sql.SparkSession;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+
+import eu.dnetlib.dhp.actionset.common.ASResultInfo;
+import eu.dnetlib.dhp.actionset.common.Common;
+import eu.dnetlib.dhp.application.ArgumentApplicationParser;
+import eu.dnetlib.dhp.schema.action.AtomicAction;
+import eu.dnetlib.dhp.schema.oaf.*;
+import scala.Tuple2;
+
+public class SparkRedistributeIISRelations implements Serializable {
+
+	private static final Logger log = LoggerFactory.getLogger(SparkRedistributeIISRelations.class);
+
+	public static void main(String[] args) throws Exception {
+		String jsonConfiguration = IOUtils
+			.toString(
+				SparkRedistributeIISRelations.class
+					.getResourceAsStream(
+						"/eu/dnetlib/dhp/actionset/stable/input_redistribute_parameters.json"));
+
+		final ArgumentApplicationParser parser = new ArgumentApplicationParser(jsonConfiguration);
+		parser.parseArgument(args);
+
+		Boolean isSparkSessionManaged = Optional
+			.ofNullable(parser.get("isSparkSessionManaged"))
+			.map(Boolean::valueOf)
+			.orElse(Boolean.TRUE);
+		log.info("isSparkSessionManaged: {}", isSparkSessionManaged);
+
+		final String outputPath = parser.get("outputPath");
+		log.info("outputPath: {}", outputPath);
+
+		final String inputPath = parser.get("inputPath");
+
+		final String asInputPath = parser.get("asInputPath");
+
+		SparkConf conf = new SparkConf();
+
+		runWithSparkSession(
+			conf,
+			isSparkSessionManaged,
+			spark -> {
+				Common.removeOutputDir(spark, outputPath);
+				Dataset<ResultPid> resultPidDataset = Common.readPath(spark, inputPath, ResultPid.class);
+				Dataset<ASResultInfo> asResultInfoDataset = Common.readPath(spark, asInputPath, ASResultInfo.class);
+				execRelation(spark, asResultInfoDataset.filter("type = 'relation'"), resultPidDataset, outputPath);
+			});
+	}
+
+	private static void execRelation(SparkSession spark, Dataset<ASResultInfo> asResultInfoDataset,
+		Dataset<ResultPid> resultPidDataset, String outputPathRelation) {
+		resultPidDataset
+			.joinWith(
+				asResultInfoDataset, resultPidDataset.col("resultId").equalTo(asResultInfoDataset.col("id")), "left")
+			.flatMap((FlatMapFunction<Tuple2<ResultPid, ASResultInfo>, Relation>) value -> {
+				List<Relation> relationList = new ArrayList<>();
+				if (Objects.nonNull(value._2())) {
+					relationList.add(getRelation(value._2(), "result"));
+					relationList.add(getRelation(value._2(), "project"));
+
+				}
+				return relationList.iterator();
+			}, Encoders.bean(Relation.class))
+			.filter(Objects::nonNull)
+			.toJavaRDD()
+			.map(p -> new AtomicAction(Relation.class, p))
+			.mapToPair(
+				aa -> getTextTextTuple2(aa))
+			.saveAsHadoopFile(outputPathRelation, Text.class, Text.class, SequenceFileOutputFormat.class);
+
+	}
+
+	private static Relation getRelation(ASResultInfo asResultInfo, String type) {
+		Relation r = new Relation();
+		if (type.equals("result")) {
+			r.setSource(asResultInfo.getId());
+			r.setRelClass("isProducedBy");
+			r.setTarget(asResultInfo.getValue().get(0).getValue());
+		} else {
+			r.setRelClass("produces");
+			r.setSource(asResultInfo.getValue().get(0).getValue());
+			r.setTarget(asResultInfo.getId());
+		}
+
+		r.setRelType("resultProject");
+		r.setSubRelType("outcome");
+
+		r.setDataInfo(getDataInfo(asResultInfo));
+		return r;
+	}
+
+	private static DataInfo getDataInfo(ASResultInfo asResultInfo) {
+		DataInfo di = new DataInfo();
+		di.setInvisible(false);
+		di.setInferred(true);
+		di.setDeletedbyinference(false);
+		di.setTrust(asResultInfo.getValue().get(0).getTrust());
+		di.setInferenceprovenance(asResultInfo.getValue().get(0).getInference_provenance());
+		Qualifier pAction = new Qualifier();
+		pAction.setClassid("iis");
+		pAction.setClassname("iss");
+		pAction.setSchemename("dnet:provenanceActions");
+		pAction.setSchemeid("dnet:provenanceActions");
+		di.setProvenanceaction(pAction);
+		return di;
+	}
+
+	private static Tuple2<Text, Text> getTextTextTuple2(AtomicAction aa) throws JsonProcessingException {
+		String st = "";
+		System.out.println(st);
+		return new Tuple2<>(new Text(aa.getClazz().getCanonicalName()),
+			new Text(Common.OBJECT_MAPPER.writeValueAsString(aa)));
+	}
+}

dhp-workflows/dhp-actionset-remapping/src/main/java/eu/dnetlib/dhp/actionset/stable/SparkRedistributeIISResults.java

@@ -0,0 +1,127 @@
+
+package eu.dnetlib.dhp.actionset.stable;
+
+import static eu.dnetlib.dhp.common.SparkSessionSupport.runWithSparkSession;
+
+import java.io.Serializable;
+import java.util.Arrays;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.stream.Collectors;
+
+import org.apache.commons.io.IOUtils;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.mapred.SequenceFileOutputFormat;
+import org.apache.spark.SparkConf;
+import org.apache.spark.api.java.function.MapFunction;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Encoders;
+import org.apache.spark.sql.SparkSession;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+
+import eu.dnetlib.dhp.actionset.common.ASResultInfo;
+import eu.dnetlib.dhp.actionset.common.Common;
+import eu.dnetlib.dhp.application.ArgumentApplicationParser;
+import eu.dnetlib.dhp.schema.action.AtomicAction;
+import eu.dnetlib.dhp.schema.oaf.Context;
+import eu.dnetlib.dhp.schema.oaf.DataInfo;
+import eu.dnetlib.dhp.schema.oaf.Qualifier;
+import eu.dnetlib.dhp.schema.oaf.Result;
+import scala.Tuple2;
+
+public class SparkRedistributeIISResults implements Serializable {
+
+	private static final Logger log = LoggerFactory.getLogger(SparkRedistributeIISResults.class);
+
+	public static void main(String[] args) throws Exception {
+		String jsonConfiguration = IOUtils
+			.toString(
+				SparkRedistributeIISResults.class
+					.getResourceAsStream(
+						"/eu/dnetlib/dhp/actionset/stable/input_redistribute_parameters.json"));
+
+		final ArgumentApplicationParser parser = new ArgumentApplicationParser(jsonConfiguration);
+		parser.parseArgument(args);
+
+		Boolean isSparkSessionManaged = Optional
+			.ofNullable(parser.get("isSparkSessionManaged"))
+			.map(Boolean::valueOf)
+			.orElse(Boolean.TRUE);
+		log.info("isSparkSessionManaged: {}", isSparkSessionManaged);
+
+		final String outputPath = parser.get("outputPath");
+		log.info("outputPath: {}", outputPath);
+
+		final String outputPathRelation = parser.get("outputPathRelation");
+		log.info("outputPathRelation: {}", outputPathRelation);
+
+		final String inputPath = parser.get("inputPath");
+
+		final String asInputPath = parser.get("asInputPath");
+
+		SparkConf conf = new SparkConf();
+
+		runWithSparkSession(
+			conf,
+			isSparkSessionManaged,
+			spark -> {
+				Common.removeOutputDir(spark, outputPath);
+				Dataset<ResultPid> resultPidDataset = Common.readPath(spark, inputPath, ResultPid.class);
+				Dataset<ASResultInfo> asResultInfoDataset = Common.readPath(spark, asInputPath, ASResultInfo.class);
+				execResult(spark, asResultInfoDataset.filter("type = 'result'"), resultPidDataset, outputPath);
+				// execRelation(spark, asResultInfoDataset.filter("type = 'relation'"), resultPidDataset,
+				// outputPathRelation);
+			});
+	}
+
+	private static void execResult(SparkSession spark, Dataset<ASResultInfo> info, Dataset<ResultPid> resultPidDataset,
+		String outputPathResult) {
+		info
+			.joinWith(resultPidDataset, info.col("id").equalTo(resultPidDataset.col("resultId")), "left")
+			.map((MapFunction<Tuple2<ASResultInfo, ResultPid>, Result>) value -> {
+				Result ri = null;
+				if (Objects.nonNull(value._2())) {
+					ri = new Result();
+					ASResultInfo asri = value._1();
+					ResultPid rp = value._2();
+					ri.setId(value._1().getId());
+					ri
+						.setContext(
+							asri
+								.getValue()
+								.stream()
+								.map(c -> {
+									Context context = new Context();
+									context.setId(c.getValue());
+									DataInfo di = new DataInfo();
+									di.setInferenceprovenance(c.getInference_provenance());
+									di.setTrust(c.getTrust());
+									di.setDeletedbyinference(false);
+									Qualifier pa = new Qualifier();
+									pa.setClassname("iis");
+									pa.setClassid("iis");
+									pa.setSchemeid("dnet:provenanceActions");
+									pa.setSchemename("dnet:provenanceActions");
+									di.setProvenanceaction(pa);
+									context.setDataInfo(Arrays.asList(di));
+									return context;
+								})
+								.collect(Collectors.toList()));
+				}
+				return ri;
+			}, Encoders.bean(Result.class))
+			.filter(Objects::nonNull)
+			.toJavaRDD()
+			.mapToPair(r -> getTextTextTuple2(r))
+			.saveAsHadoopFile(outputPathResult, Text.class, Text.class, SequenceFileOutputFormat.class);
+	}
+
+	private static Tuple2<Text, Text> getTextTextTuple2(Result r) throws JsonProcessingException {
+		AtomicAction aa = new AtomicAction(Result.class, r);
+		return new Tuple2<>(new Text(aa.getClazz().getCanonicalName()),
+			new Text(Common.OBJECT_MAPPER.writeValueAsString(aa)));
+	}
+}

dhp-workflows/dhp-actionset-remapping/src/main/java/eu/dnetlib/dhp/actionset/stable/SparkSelectResults.java

@@ -0,0 +1,89 @@
+
+package eu.dnetlib.dhp.actionset.stable;
+
+import static eu.dnetlib.dhp.common.SparkSessionSupport.runWithSparkSession;
+
+import java.io.Serializable;
+import java.util.Optional;
+
+import org.apache.commons.io.IOUtils;
+import org.apache.spark.SparkConf;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.SaveMode;
+import org.apache.spark.sql.SparkSession;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import eu.dnetlib.dhp.actionset.common.Common;
+import eu.dnetlib.dhp.application.ArgumentApplicationParser;
+import eu.dnetlib.dhp.schema.common.ModelSupport;
+import eu.dnetlib.dhp.schema.oaf.Result;
+
+public class SparkSelectResults implements Serializable {
+	private static final Logger log = LoggerFactory.getLogger(SparkSelectResults.class);
+
+	public static void main(String[] args) throws Exception {
+		String jsonConfiguration = IOUtils
+			.toString(
+				SparkSelectResults.class
+					.getResourceAsStream(
+						"/eu/dnetlib/dhp/actionset/stable/input_select_parameters.json"));
+
+		final ArgumentApplicationParser parser = new ArgumentApplicationParser(jsonConfiguration);
+		parser.parseArgument(args);
+
+		Boolean isSparkSessionManaged = Optional
+			.ofNullable(parser.get("isSparkSessionManaged"))
+			.map(Boolean::valueOf)
+			.orElse(Boolean.TRUE);
+		log.info("isSparkSessionManaged: {}", isSparkSessionManaged);
+
+		final String outputPath = parser.get("outputPath");
+		log.info("outputPath: {}", outputPath);
+
+		final String inputPath = parser.get("inputPath");
+
+//        final String resultClassName = parser.get("resultTableName");
+//        log.info("resultTableName: {}", resultClassName);
+//
+//        Class<? extends Result> resultClazz = (Class<? extends Result>) Class.forName(resultClassName);
+
+		SparkConf conf = new SparkConf();
+
+		runWithSparkSession(
+			conf,
+			isSparkSessionManaged,
+			spark -> {
+				Common.removeOutputDir(spark, outputPath);
+				run(spark, inputPath, outputPath);
+			});
+	}
+
+	private static void run(SparkSession spark, String inputPath, String outputPath) {
+		ModelSupport.resultTypes
+			.keySet()
+			.forEach(key -> exec(spark, inputPath + "/" + key, outputPath, ModelSupport.resultTypes.get(key)));
+	}
+
+	private static <R extends Result> void exec(SparkSession spark, String inputPath, String outputPath,
+		Class<R> resultClazz) {
+		Dataset<R> result = Common.readPath(spark, inputPath, resultClazz);
+
+		result.createOrReplaceTempView("result");
+
+		spark
+			.sql(
+				"SELECT id resultId, persId.value doi " +
+					"from result   " +
+					"lateral view explode(pid) p as persId  " +
+					"lateral view explode(collectedfrom) c as cf " +
+					"where persId.qualifier.classid = 'doi'  " +
+					"and (cf.key = '10|openaire____::9e3be59865b2c1c335d32dae2fe7b254' or  " +
+					"cf.key = '10|openaire____::081b82f96300b6a6e3d282bad31cb6e2') " +
+					"and result.id not like '50|dedup%' ")
+			.write()
+			.mode(SaveMode.Append)
+			.option("compression", "gzip")
+			.json(outputPath);
+	}
+}

dhp-workflows/dhp-actionset-remapping/src/main/resources/eu/dnetlib/dhp/actionset/common/input_expand_parameters.json

@@ -0,0 +1,21 @@
+[
+
+  {
+    "paramName": "issm",
+    "paramLongName": "isSparkSessionManaged",
+    "paramDescription": "when true will stop SparkSession after job execution",
+    "paramRequired": false
+  },
+  {
+    "paramName": "ip",
+    "paramLongName": "inputPath",
+    "paramDescription": "the input path of the prepared info",
+    "paramRequired": true
+  },
+  {
+    "paramName": "op",
+    "paramLongName": "outputPath",
+    "paramDescription": "the path of the new ActionSet",
+    "paramRequired": true
+  }
+]

dhp-workflows/dhp-actionset-remapping/src/main/resources/eu/dnetlib/dhp/actionset/common/input_prepare_parameters.json

@@ -0,0 +1,33 @@
+[
+{
+"paramName": "issm",
+"paramLongName": "isSparkSessionManaged",
+"paramDescription": "when true will stop SparkSession after job execution",
+"paramRequired": false
+},
+{
+"paramName": "as",
+"paramLongName": "actionSets",
+"paramDescription": "the allowed list of action sets",
+"paramRequired": true
+},
+{
+"paramName": "ip",
+"paramLongName": "inputPath",
+"paramDescription": "the input path",
+"paramRequired": true
+},
+{
+"paramName": "op",
+"paramLongName": "outputPath",
+"paramDescription": "the path of the new ActionSet",
+"paramRequired": true
+},
+
+{
+"paramName":"is",
+"paramLongName":"isLookUpUrl",
+"paramDescription": "URL of the isLookUp Service",
+"paramRequired": true
+}
+]

dhp-workflows/dhp-actionset-remapping/src/main/resources/eu/dnetlib/dhp/actionset/oozie_app/config-default.xml

@@ -0,0 +1,58 @@
+<configuration>
+    <property>
+        <name>jobTracker</name>
+        <value>yarnRM</value>
+    </property>
+    <property>
+        <name>nameNode</name>
+        <value>hdfs://nameservice1</value>
+    </property>
+    <property>
+        <name>oozie.use.system.libpath</name>
+        <value>true</value>
+    </property>
+    <property>
+        <name>oozie.action.sharelib.for.spark</name>
+        <value>spark2</value>
+    </property>
+    <property>
+        <name>hive_metastore_uris</name>
+        <value>thrift://iis-cdh5-test-m3.ocean.icm.edu.pl:9083</value>
+    </property>
+    <property>
+        <name>spark2YarnHistoryServerAddress</name>
+        <value>http://iis-cdh5-test-gw.ocean.icm.edu.pl:18089</value>
+    </property>
+    <property>
+        <name>spark2ExtraListeners</name>
+        <value>com.cloudera.spark.lineage.NavigatorAppListener</value>
+    </property>
+    <property>
+        <name>spark2SqlQueryExecutionListeners</name>
+        <value>com.cloudera.spark.lineage.NavigatorQueryListener</value>
+    </property>
+    <property>
+        <name>sparkExecutorNumber</name>
+        <value>4</value>
+    </property>
+    <property>
+        <name>spark2EventLogDir</name>
+        <value>/user/spark/spark2ApplicationHistory</value>
+    </property>
+    <property>
+        <name>sparkDriverMemory</name>
+        <value>15G</value>
+    </property>
+    <property>
+        <name>sparkExecutorMemory</name>
+        <value>6G</value>
+    </property>
+    <property>
+        <name>sparkExecutorCores</name>
+        <value>1</value>
+    </property>
+    <property>
+        <name>oozie.launcher.mapreduce.user.classpath.first</name>
+        <value>true</value>
+    </property>
+</configuration>

dhp-workflows/dhp-actionset-remapping/src/main/resources/eu/dnetlib/dhp/actionset/oozie_app/workflow.xml

@@ -0,0 +1,184 @@
+<workflow-app name="RemapIIS" xmlns="uri:oozie:workflow:0.5">
+    <parameters>
+        <property>
+            <name>isLookUpUrl</name>
+            <description>the IsLookUpUrl</description>
+        </property>
+
+        <property>
+            <name>inputPath</name>
+            <description>the graph input path</description>
+        </property>
+
+        <property>
+            <name>relationOutputPath</name>
+            <description>path where to store the stable action set for relations</description>
+        </property>
+        <property>
+            <name>resultOutputPath</name>
+            <description>path where to store the stable action set for results</description>
+        </property>
+        <property>
+            <name>actionSets</name>
+            <description>the allowed list of action sets</description>
+        </property>
+    </parameters>
+
+    <start to="deleteoutputpath"/>
+    <kill name="Kill">
+        <message>Action failed, error message[${wf:errorMessage(wf:lastErrorNode())}]</message>
+    </kill>
+    <action name="deleteoutputpath">
+        <fs>
+            <delete path='${relationOutputPath}'/>
+            <mkdir path='${relationOutputPath}'/>
+            <delete path='${resultOutputPath}'/>
+            <mkdir path='${resultOutputPath}'/>
+            <delete path='${workingDir}'/>
+            <mkdir path='${workingDir}'/>
+        </fs>
+        <ok to="prepare_info"/>
+        <error to="Kill"/>
+    </action>
+
+
+    <action name="prepare_info">
+        <spark xmlns="uri:oozie:spark-action:0.2">
+            <master>yarn</master>
+            <mode>cluster</mode>
+            <name>PrepareInfo</name>
+            <class>eu.dnetlib.dhp.actionset.common.SparkPrepareInfo</class>
+            <jar>dhp-actionset-remapping-${projectVersion}.jar</jar>
+            <spark-opts>
+                --executor-cores=${sparkExecutorCores}
+                --executor-memory=${sparkExecutorMemory}
+                --driver-memory=${sparkDriverMemory}
+                --conf spark.extraListeners=${spark2ExtraListeners}
+                --conf spark.sql.queryExecutionListeners=${spark2SqlQueryExecutionListeners}
+                --conf spark.yarn.historyServer.address=${spark2YarnHistoryServerAddress}
+                --conf spark.eventLog.dir=${nameNode}${spark2EventLogDir}
+                --conf spark.sql.shuffle.partitions=3840
+            </spark-opts>
+            <arg>--isLookUpUrl</arg><arg>${isLookUpUrl}</arg>
+            <arg>--actionSets</arg><arg>${actionSets}</arg>
+            <arg>--inputPath</arg><arg>${inputPath}</arg>
+            <arg>--outputPath</arg><arg>${workingDir}/preparedInfo</arg>
+        </spark>
+        <ok to="prepare_next"/>
+        <error to="Kill"/>
+    </action>
+
+    <fork name="prepare_next">
+        <path start="expand_relation"/>
+        <path start="select_results"/>
+    </fork>
+
+    <action name="expand_relation">
+        <spark xmlns="uri:oozie:spark-action:0.2">
+            <master>yarn</master>
+            <mode>cluster</mode>
+            <name>ExpandRelation</name>
+            <class>eu.dnetlib.dhp.actionset.common.SparkExpandResultInfo</class>
+            <jar>dhp-actionset-remapping-${projectVersion}.jar</jar>
+            <spark-opts>
+                --executor-cores=${sparkExecutorCores}
+                --executor-memory=${sparkExecutorMemory}
+                --driver-memory=${sparkDriverMemory}
+                --conf spark.extraListeners=${spark2ExtraListeners}
+                --conf spark.sql.queryExecutionListeners=${spark2SqlQueryExecutionListeners}
+                --conf spark.yarn.historyServer.address=${spark2YarnHistoryServerAddress}
+                --conf spark.eventLog.dir=${nameNode}${spark2EventLogDir}
+                --conf spark.sql.shuffle.partitions=3840
+            </spark-opts>
+            <arg>--inputPath</arg><arg>${workingDir}/preparedInfo</arg>
+            <arg>--outputPath</arg><arg>${workingDir}/expandedActionSet</arg>
+        </spark>
+        <ok to="wait1"/>
+        <error to="Kill"/>
+    </action>
+
+    <action name="select_results">
+        <spark xmlns="uri:oozie:spark-action:0.2">
+            <master>yarn</master>
+            <mode>cluster</mode>
+            <name>SelectResults</name>
+            <class>eu.dnetlib.dhp.actionset.stable.SparkSelectResults</class>
+            <jar>dhp-actionset-remapping-${projectVersion}.jar</jar>
+            <spark-opts>
+                --executor-cores=${sparkExecutorCores}
+                --executor-memory=${sparkExecutorMemory}
+                --driver-memory=${sparkDriverMemory}
+                --conf spark.extraListeners=${spark2ExtraListeners}
+                --conf spark.sql.queryExecutionListeners=${spark2SqlQueryExecutionListeners}
+                --conf spark.yarn.historyServer.address=${spark2YarnHistoryServerAddress}
+                --conf spark.eventLog.dir=${nameNode}${spark2EventLogDir}
+                --conf spark.sql.shuffle.partitions=3840
+            </spark-opts>
+            <arg>--inputPath</arg><arg>${inputPath}</arg>
+            <arg>--outputPath</arg><arg>${workingDir}/selectedResults</arg>
+        </spark>
+        <ok to="wait1"/>
+        <error to="Kill"/>
+    </action>
+
+    <join name="wait1" to="redistributeAA"/>
+
+    <fork name="redistributeAA">
+        <path start="redistribute_relation"/>
+        <path start="redistribute_result"/>
+    </fork>
+
+    <action name="redistribute_relation">
+        <spark xmlns="uri:oozie:spark-action:0.2">
+            <master>yarn</master>
+            <mode>cluster</mode>
+            <name>RedistributeRelation</name>
+            <class>eu.dnetlib.dhp.actionset.stable.SparkRedistributeIISRelations</class>
+            <jar>dhp-actionset-remapping-${projectVersion}.jar</jar>
+            <spark-opts>
+                --executor-cores=${sparkExecutorCores}
+                --executor-memory=${sparkExecutorMemory}
+                --driver-memory=${sparkDriverMemory}
+                --conf spark.extraListeners=${spark2ExtraListeners}
+                --conf spark.sql.queryExecutionListeners=${spark2SqlQueryExecutionListeners}
+                --conf spark.yarn.historyServer.address=${spark2YarnHistoryServerAddress}
+                --conf spark.eventLog.dir=${nameNode}${spark2EventLogDir}
+                --conf spark.sql.shuffle.partitions=3840
+            </spark-opts>
+            <arg>--inputPath</arg><arg>${workingDir}/selectedResults</arg>
+            <arg>--asInputPath</arg><arg>${workingDir}/expandedActionSet</arg>
+            <arg>--outputPath</arg><arg>${relationOutputPath}</arg>
+        </spark>
+        <ok to="wait2"/>
+        <error to="Kill"/>
+    </action>
+
+    <action name="redistribute_result">
+        <spark xmlns="uri:oozie:spark-action:0.2">
+            <master>yarn</master>
+            <mode>cluster</mode>
+            <name>SelectResults</name>
+            <class>eu.dnetlib.dhp.actionset.stable.SparkRedistributeIISResults</class>
+            <jar>dhp-actionset-remapping-${projectVersion}.jar</jar>
+            <spark-opts>
+                --executor-cores=${sparkExecutorCores}
+                --executor-memory=${sparkExecutorMemory}
+                --driver-memory=${sparkDriverMemory}
+                --conf spark.extraListeners=${spark2ExtraListeners}
+                --conf spark.sql.queryExecutionListeners=${spark2SqlQueryExecutionListeners}
+                --conf spark.yarn.historyServer.address=${spark2YarnHistoryServerAddress}
+                --conf spark.eventLog.dir=${nameNode}${spark2EventLogDir}
+                --conf spark.sql.shuffle.partitions=3840
+            </spark-opts>
+            <arg>--inputPath</arg><arg>${workingDir}/selectedResults</arg>
+            <arg>--asInputPath</arg><arg>${workingDir}/expandedActionSet</arg>
+            <arg>--outputPath</arg><arg>${resultOutputPath}</arg>
+        </spark>
+        <ok to="wait2"/>
+        <error to="Kill"/>
+    </action>
+
+    <join name="wait2" to="End"/>
+
+    <end name="End"/>
+</workflow-app>

dhp-workflows/dhp-actionset-remapping/src/main/resources/eu/dnetlib/dhp/actionset/stable/input_redistribute_parameters.json

@@ -0,0 +1,26 @@
+[
+  {
+    "paramName": "issm",
+    "paramLongName": "isSparkSessionManaged",
+    "paramDescription": "when true will stop SparkSession after job execution",
+    "paramRequired": false
+  },
+  {
+    "paramName": "op",
+    "paramLongName": "outputPath",
+    "paramDescription": "the output path for the stable output",
+    "paramRequired": true
+  },
+  {
+    "paramName": "ip",
+    "paramLongName": "inputPath",
+    "paramDescription": "the result with doi input path",
+    "paramRequired": true
+  },
+  {
+    "paramName": "aip",
+    "paramLongName": "asInputPath",
+    "paramDescription": "the input path for the expanded association",
+    "paramRequired": true
+  }
+]

dhp-workflows/dhp-actionset-remapping/src/main/resources/eu/dnetlib/dhp/actionset/stable/input_select_parameters.json

@@ -0,0 +1,20 @@
+[
+{
+"paramName": "issm",
+"paramLongName": "isSparkSessionManaged",
+"paramDescription": "when true will stop SparkSession after job execution",
+"paramRequired": false
+},
+{
+"paramName": "ip",
+"paramLongName": "inputPath",
+"paramDescription": "the input path",
+"paramRequired": true
+},
+{
+"paramName": "op",
+"paramLongName": "outputPath",
+"paramDescription": "the path of the new ActionSet",
+"paramRequired": true
+}
+]

dhp-workflows/dhp-actionset-remapping/src/test/java/eu/dnetlib/dhp/actionset/ExpandResultInfoTest.java

@@ -0,0 +1,173 @@
+
+package eu.dnetlib.dhp.actionset;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+
+import org.apache.commons.io.FileUtils;
+import org.apache.spark.SparkConf;
+import org.apache.spark.api.java.JavaRDD;
+import org.apache.spark.api.java.JavaSparkContext;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Encoders;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.SparkSession;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.Assertions;
+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.actionset.common.ASResultInfo;
+import eu.dnetlib.dhp.actionset.common.SparkExpandResultInfo;
+
+public class ExpandResultInfoTest {
+
+	private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+
+	private static SparkSession spark;
+	private static final String FAKE_ISLOOKUP = "http://beta.services.openaire.eu/";
+
+	private static Path workingDir;
+	private static final Logger log = LoggerFactory
+		.getLogger(ExpandResultInfoTest.class);
+
+	@BeforeAll
+	public static void beforeAll() throws IOException {
+		workingDir = Files
+			.createTempDirectory(ExpandResultInfoTest.class.getSimpleName());
+		log.info("using work dir {}", workingDir);
+
+		SparkConf conf = new SparkConf();
+		conf.setAppName(ExpandResultInfoTest.class.getSimpleName());
+
+		conf.setMaster("local[*]");
+		conf.set("spark.driver.host", "localhost");
+		conf.set("hive.metastore.local", "true");
+		conf.set("spark.ui.enabled", "false");
+		conf.set("spark.sql.warehouse.dir", workingDir.toString());
+		conf.set("hive.metastore.warehouse.dir", workingDir.resolve("warehouse").toString());
+
+		spark = SparkSession
+			.builder()
+			.appName(ExpandResultInfoTest.class.getSimpleName())
+			.config(conf)
+			.getOrCreate();
+	}
+
+	@AfterAll
+	public static void afterAll() throws IOException {
+		FileUtils.deleteDirectory(workingDir.toFile());
+		spark.stop();
+	}
+
+	@Test
+	public void expandAS() throws Exception {
+		SparkExpandResultInfo
+			.main(
+				new String[] {
+					"-isSparkSessionManaged",
+					Boolean.FALSE.toString(),
+					"-inputPath",
+					getClass().getResource("/eu/dnetlib/dhp/actionset/step2/preparedInfo").getPath(),
+					"-outputPath",
+					workingDir.toString() + "/expandedActionSet"
+				});
+
+		final JavaSparkContext sc = new JavaSparkContext(spark.sparkContext());
+
+		JavaRDD<ASResultInfo> tmp = sc
+			.textFile(workingDir.toString() + "/expandedActionSet")
+			.map(item -> OBJECT_MAPPER.readValue(item, ASResultInfo.class));
+
+		Dataset<ASResultInfo> verificationDataset = spark.createDataset(tmp.rdd(), Encoders.bean(ASResultInfo.class));
+
+		Assertions.assertEquals(25, verificationDataset.count());
+
+		Assertions.assertEquals(0, verificationDataset.filter("substr(id,1,8) = '50|dedup'").count());
+
+		Assertions
+			.assertEquals(
+				3, verificationDataset.filter("id = '50|doiboost____::0f10b8f21b7925a344f41edb774f0b0a'").count());
+		Assertions
+			.assertEquals(
+				3, verificationDataset.filter("id = '50|od_______166::779de9b3a2d224779be52fae43b5fc80'").count());
+		Assertions
+			.assertEquals(
+				3, verificationDataset.filter("id = '50|od_______165::779de9b3a2d224779be52fae43b5fc80'").count());
+		Assertions
+			.assertEquals(
+				3, verificationDataset.filter("id = '50|od______3515::779de9b3a2d224779be52fae43b5fc80'").count());
+
+		Assertions
+			.assertEquals(
+				2, verificationDataset.filter("id = '50|doiboost____::78329557c23bee513963ebf295d1434d'").count());
+		Assertions
+			.assertEquals(
+				2, verificationDataset.filter("id = '50|doiboost____::8978b9b797294da5306950a94a58d98c'").count());
+		Assertions
+			.assertEquals(
+				2, verificationDataset.filter("id = '50|doiboost____::fb2c70723d74f45329640255a959333d'").count());
+		Assertions
+			.assertEquals(
+				2, verificationDataset.filter("id = '50|base_oa_____::fb2c70723d74f45329640255a959333d'").count());
+
+		Assertions
+			.assertEquals(
+				5, verificationDataset
+					.filter(
+						"id = '50|_____OmicsDI::039dbb63f11b19dc15113b34ebceb0d2' " +
+							"or id = '50|_____OmicsDI::05f133acca27d72866c6720a95515f57' or " +
+							"id = '50|_____OmicsDI::2d508eba981699a30e969d1ab5a068b8' or " +
+							"id = '50|datacite____::00bddedc38dc045780dc84c27bc8fecd' or " +
+							"id = '50|datacite____::00f7f89392fa75e944dc8d329e9e8024'")
+					.count());
+
+		verificationDataset.createOrReplaceTempView("verificationDataset");
+
+		Dataset<Row> verify = spark
+			.sql(
+				("SELECT id, type, val.value value, val.trust trust, val.inference_provenance prov " +
+					"FROM verificationDataset " +
+					"LATERAL VIEW EXPLODE(value) v as val"));
+
+		Assertions.assertEquals(25, verify.count());
+
+		Assertions.assertEquals(20, verify.filter("type = 'relation'").count());
+		Assertions.assertEquals(5, verify.filter("type = 'result'").count());
+
+		Assertions
+			.assertEquals(
+				1, verify
+					.filter(
+						"id = '50|doiboost____::0f10b8f21b7925a344f41edb774f0b0a' " +
+							"and value = '40|rcuk________::8dec51859e6b66cd040670b432b9e59c' and " +
+							"prov = 'iis::document_referencedProjects' and " +
+							"trust = '0.897'")
+					.count());
+
+		Assertions
+			.assertEquals(
+				1, verify
+					.filter(
+						"id = '50|doiboost____::0f10b8f21b7925a344f41edb774f0b0a' " +
+							"and value = '40|rcuk________::5e312e08bd65f126d7d79b3d1d677eb3' and " +
+							"prov = 'iis::document_referencedProjects' and " +
+							"trust = '0.897'")
+					.count());
+
+		Assertions
+			.assertEquals(
+				1, verify
+					.filter(
+						"id = '50|doiboost____::0f10b8f21b7925a344f41edb774f0b0a' " +
+							"and value = '40|corda_______::6d500f8fceb2bb81b0750820469e1cd8' and " +
+							"prov = 'iis::document_referencedProjects' and " +
+							"trust = '0.7085'")
+					.count());
+	}
+}

dhp-workflows/dhp-actionset-remapping/src/test/java/eu/dnetlib/dhp/actionset/PrepareInfoTest.java

@@ -0,0 +1,567 @@
+
+package eu.dnetlib.dhp.actionset;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Arrays;
+import java.util.List;
+
+import org.apache.commons.io.FileUtils;
+import org.apache.spark.SparkConf;
+import org.apache.spark.api.java.JavaRDD;
+import org.apache.spark.api.java.JavaSparkContext;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Encoders;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.SparkSession;
+import org.junit.jupiter.api.*;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.gson.Gson;
+
+import eu.dnetlib.dhp.actionset.common.ASResultInfo;
+import eu.dnetlib.dhp.actionset.common.ActionSet;
+import eu.dnetlib.dhp.actionset.common.PrepareInfo;
+import eu.dnetlib.dhp.actionset.common.RelationMerges;
+
+@Disabled
+public class PrepareInfoTest {
+
+	private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+
+	private static final ClassLoader cl = PrepareInfoTest.class
+		.getClassLoader();
+
+	private static SparkSession spark;
+	private static final String FAKE_ISLOOKUP = "http://beta.services.openaire.eu/";
+
+	private static Path workingDir;
+	private static final Logger log = LoggerFactory
+		.getLogger(PrepareInfoTest.class);
+
+	private static final List<ActionSet> actionSetList = Arrays
+		.asList(
+			ActionSet
+				.newInstance(
+					"iis-dataset-entities-preprocessing", "rawset_74155091-ce3b-4951-849c-f41dd4186699_1555613756167",
+					"entities_dataset"),
+			ActionSet
+				.newInstance(
+					"iis-document-statistics", "rawset_d0a24381-1241-4d83-9669-22110ab72f63_1415965369474",
+					"document_statistics"),
+			ActionSet
+				.newInstance(
+					"iis-researchinitiative", "rawset_718b528a-0a10-4303-9290-4f61c04b7ace_1594030710764",
+					"document_research_initiative"),
+			ActionSet
+				.newInstance(
+					"iis-document-citations", "rawset_b22d36c2-36e5-4ff3-97ef-946fa84a57dd_1594030710774",
+					"document_referencedDocuments"),
+			ActionSet
+				.newInstance(
+					"iis-dataset-entities-main", "rawset_fffa6131-3e7d-4c2c-82d8-844517e721c0_1594030710760",
+					"entities_dataset"),
+			ActionSet
+				.newInstance(
+					"iis-document-affiliation", "rawset_d62066d3-b6d9-424a-bea0-bab884a55292_1594030710732",
+					"matched_doc_organizations"),
+			ActionSet
+				.newInstance(
+					"iis-document-classes", "rawset_35b252fb-2180-4115-818d-a8110616b892_1594030710771",
+					"document_classes"),
+			ActionSet
+				.newInstance(
+					"iis-document-similarities", "rawset_cc4706b4-ed1d-4862-a13b-b0afdd7016a3_1594030710768",
+					"document_similarities_standard"),
+			ActionSet
+				.newInstance(
+					"iis-referenced-datasets-main", "rawset_c2ea95d3-a2c0-48f4-9184-5f3478399cc6_1594030710757",
+					"document_referencedDatasets"),
+			ActionSet
+				.newInstance(
+					"iis-referenced-datasets-preprocessing",
+					"rawset_91543cfa-b543-46c1-a87f-b1a550bc6937_1555613756158", "document_referencedDatasets"),
+			ActionSet
+				.newInstance(
+					"iis-referenced-projects-main", "rawset_ccf0d39d-0077-4e61-af6c-dc191e2fca68_1594030710754",
+					"document_referencedProjects"),
+			ActionSet
+				.newInstance(
+					"iis-referenced-projects-preprocessing",
+					"rawset_564ae405-a221-472f-8bd0-ee8bfbbd9164_1555613756135", "document_referencedProjects"),
+			ActionSet
+				.newInstance(
+					"iis-referenceextraction-pdb", "rawset_07dc3b63-e5a4-4a54-90ba-5226fd55f1c9_1594030710776",
+					"document_pdb"),
+			ActionSet
+				.newInstance(
+					"document_software_url", "rawset_75e2b097-2a10-41a7-97eb-70737b678793_1594030710779",
+					"document_software_url"),
+			ActionSet
+				.newInstance(
+					"iis-wos-entities", "rawset_handled6-0ab3-4fd0-a33b-refereed2fc0_1555613756163",
+					"entities_document"),
+			ActionSet
+				.newInstance(
+					"iis-extracted-metadata", "rawset_f859722c-bfec-4711-9132-8b24766c208d_1415965369473",
+					"document_metadata"),
+			ActionSet
+				.newInstance(
+					"dedup-similarity-organization-simple", "rawset_a62362a8-9800-4ba3-a060-1fbc0e3ea1a5_1587998691703",
+					"dedup-similarity-organization-simple"),
+			ActionSet
+				.newInstance(
+					"dedup-similarity-organization", "rawset_9da5e0f1-a49a-40fc-aaac-17de80fa5ceb_1436537292583",
+					"dedup-similarity-organization"),
+			ActionSet
+				.newInstance(
+					"dedup-similarity-result-levenstein", "rawset_4921d674-aea3-4115-ad33-fe6833569176_1587984647217",
+					"dedup-similarity-result-levenstein"),
+			ActionSet.newInstance("dedup-similarity-person", "", "dedup-similarity-person"),
+			ActionSet
+				.newInstance(
+					"iis-entities-software", "rawset_c4b060b5-d620-45dd-9a0a-25befb23ef7c_1594030710782",
+					"entities_software"),
+			ActionSet
+				.newInstance(
+					"iis-communities", "rawset_4c632429-6f12-4f18-b54f-e60b346859d7_1594030710791",
+					"document_community"),
+			ActionSet
+				.newInstance(
+					"scholexplorer-dump", "rawset_d349ffdd-384a-47f6-986f-6c04edee3294_1592572750",
+					"scholexplorer-dump"),
+			ActionSet
+				.newInstance("gridac-dump", "rawset_a2854367-3586-4945-a124-1328e91568bd_1571646606840", "gridac-dump"),
+			ActionSet
+				.newInstance(
+					"doiboost-organizations", "rawset_7626c52f-7f17-47f0-9094-da2c6b883d41_1574951682027",
+					"doiboost-organizations"),
+			ActionSet.newInstance("doiboost", "rawset_handledb-a3ae-4d6e-8187-refereed6e18_15912730340000", "doiboost"),
+			ActionSet
+				.newInstance(
+					"orcidworks-no-doi", "rawset_handledf-ef8d-4e1d-89b6-refereed6ce6_1574862348031",
+					"orcidworks-no-doi"),
+			ActionSet
+				.newInstance(
+					"iis-entities-patent", "rawset_4dfd999e-7e3c-48eb-be92-92393da81e19_1594030710806",
+					"entities_patent"),
+			ActionSet
+				.newInstance(
+					"iis-referenced-patents", "rawset_b4387e83-ed2c-4c4b-9d1d-4da35a6ad752_1594030710803",
+					"document_patent"),
+			ActionSet
+				.newInstance(
+					"iis-covid-19", "rawset_437833b5-b2ef-4e3f-8642-e845086ccc2c_1594030710809", "document_covid19"),
+			ActionSet
+				.newInstance(
+					"h2020programme", "rawset_bcca8d44-6139-4aec-b579-761552440162_1590697435148", "h2020programme"));
+
+	@BeforeAll
+	public static void beforeAll() throws IOException {
+		workingDir = Files
+			.createTempDirectory(PrepareInfoTest.class.getSimpleName());
+		log.info("using work dir {}", workingDir);
+
+		SparkConf conf = new SparkConf();
+		conf.setAppName(PrepareInfoTest.class.getSimpleName());
+
+		conf.setMaster("local[*]");
+		conf.set("spark.driver.host", "localhost");
+		conf.set("hive.metastore.local", "true");
+		conf.set("spark.ui.enabled", "false");
+		conf.set("spark.sql.warehouse.dir", workingDir.toString());
+		conf.set("hive.metastore.warehouse.dir", workingDir.resolve("warehouse").toString());
+
+		spark = SparkSession
+			.builder()
+			.appName(PrepareInfoTest.class.getSimpleName())
+			.config(conf)
+			.getOrCreate();
+	}
+
+	@AfterAll
+	public static void afterAll() throws IOException {
+		FileUtils.deleteDirectory(workingDir.toFile());
+		spark.stop();
+	}
+
+	@Test
+	public void testRelationsAS() {
+		PrepareInfo pi = new PrepareInfo(false,
+			workingDir.toString() + "/preparedInfo",
+			getClass().getResource("/eu/dnetlib/dhp/actionset/step1/asInputPath").getPath(),
+			getClass().getResource("/eu/dnetlib/dhp/actionset/step1").getPath(),
+			new Gson().fromJson("[\"iis-referenced-projects-main\"]", List.class),
+			actionSetList);
+
+		pi.run();
+
+		final JavaSparkContext sc = new JavaSparkContext(spark.sparkContext());
+
+		JavaRDD<ASResultInfo> tmp = sc
+			.textFile(workingDir.toString() + "/preparedInfo/actionset")
+			.map(item -> OBJECT_MAPPER.readValue(item, ASResultInfo.class));
+
+		Dataset<ASResultInfo> verificationDataset = spark.createDataset(tmp.rdd(), Encoders.bean(ASResultInfo.class));
+
+		verificationDataset.show(false);
+		System.out.println(verificationDataset.count()); // 1813019 unidirectional relations
+		verificationDataset.createOrReplaceTempView("verificationDataset");
+
+		Dataset<Row> verify = spark
+			.sql(
+				("SELECT id, type, val.value value, val.trust trust, val.inference_provenance prov " +
+					"FROM verificationDataset " +
+					"LATERAL VIEW EXPLODE(value) v as val"));
+
+		Assertions.assertEquals(1813019, verificationDataset.count());
+
+		Assertions.assertEquals(44804, verificationDataset.filter("type = 'result'").count());
+		Assertions.assertEquals(1768215, verificationDataset.filter("type = 'relation'").count());
+
+		Assertions.assertEquals(0, verify.filter("substr(id,1,3) = '40|'").count());
+		Assertions.assertEquals(1, verify.filter("id = '50|od________18::c8e57f11074407d59f7114f047afd54e'").count());
+		Assertions
+			.assertEquals(
+				1,
+				verify
+					.filter("id = '50|od________18::c8e57f11074407d59f7114f047afd54e' and type = 'relation'")
+					.count());
+		Assertions
+			.assertEquals(
+				1,
+				verify
+					.filter(
+						"id = '50|od________18::c8e57f11074407d59f7114f047afd54e' and value = '40|nsf_________::2bedb915e92b7dd25b082c6c2f241085'")
+					.count());
+
+		Assertions
+			.assertEquals(
+				2, verificationDataset.filter("id = '50|dedup_wf_001::1cba00616e303863c34fadaf797d0f8f'").count());
+
+		Assertions
+			.assertEquals(
+				1, verify
+					.filter(
+						"id = '50|dedup_wf_001::1cba00616e303863c34fadaf797d0f8f' " +
+							"and type = 'relation' and value = '40|anr_________::5437f242b18aa615acf57dced27975c6'")
+					.count());
+
+		Assertions
+			.assertEquals(
+				1, verify
+					.filter(
+						"id = '50|dedup_wf_001::1cba00616e303863c34fadaf797d0f8f' " +
+							"and type = 'relation' and value = '40|dfgf________::7aa0b0185d7db055823cd0734ddd6521'")
+					.count());
+
+		Assertions.assertEquals(1828481, verify.filter("prov = 'iis::document_referencedProjects'").count());
+
+		Assertions
+			.assertEquals(
+				1,
+				verificationDataset
+					.filter("type = 'result' and id = '50|coactionpubl::127fa9acff1ee8e86da354d1da2378da'")
+					.count());
+		Assertions
+			.assertEquals(
+				2,
+				verify.filter("type = 'result' and id = '50|coactionpubl::127fa9acff1ee8e86da354d1da2378da'").count());
+		Assertions
+			.assertEquals(
+				1,
+				verify
+					.filter(
+						"type = 'result' and id = '50|coactionpubl::127fa9acff1ee8e86da354d1da2378da' and value = 'mes::projects::307'")
+					.count());
+		Assertions
+			.assertEquals(
+				1,
+				verify
+					.filter(
+						"type = 'result' and id = '50|coactionpubl::127fa9acff1ee8e86da354d1da2378da' and value = 'mes::projects::421'")
+					.count());
+
+	}
+
+	@Test
+	public void testInitiative() {
+		PrepareInfo pi = new PrepareInfo(false,
+			workingDir.toString() + "/preparedInfo",
+			getClass().getResource("/eu/dnetlib/dhp/actionset/step1/asInputPath").getPath(),
+			getClass().getResource("/eu/dnetlib/dhp/actionset/step1").getPath(),
+			new Gson().fromJson("[\"iis-researchinitiative\"]", List.class),
+			actionSetList);
+
+		pi.run();
+
+		final JavaSparkContext sc = new JavaSparkContext(spark.sparkContext());
+
+		JavaRDD<ASResultInfo> tmp = sc
+			.textFile(workingDir.toString() + "/preparedInfo/actionset")
+			.map(item -> OBJECT_MAPPER.readValue(item, ASResultInfo.class));
+
+		Dataset<ASResultInfo> verificationDataset = spark.createDataset(tmp.rdd(), Encoders.bean(ASResultInfo.class));
+
+		verificationDataset.createOrReplaceTempView("verificationDataset");
+
+		Dataset<Row> verify = spark
+			.sql(
+				("SELECT id, type, val.value value, val.trust trust, val.inference_provenance prov " +
+					"FROM verificationDataset " +
+					"LATERAL VIEW EXPLODE(value) v as val"));
+
+		Assertions.assertEquals(14505, verificationDataset.filter("type = 'result'").count());
+		Assertions.assertEquals(0, verificationDataset.filter("type = 'relation'").count());
+
+		Assertions.assertEquals(0, verify.filter("prov != 'iis::document_research_initiative'").count());
+
+		Assertions.assertEquals(14639, verify.count());
+
+		Assertions.assertEquals(1, verify.filter("id = '50|_____OmicsDI::278d318ee9f051971236234b181d79ce'").count());
+		Assertions
+			.assertEquals(
+				1,
+				verify
+					.filter(
+						"id = '50|_____OmicsDI::278d318ee9f051971236234b181d79ce' and value = 'egi::virtual::10256'")
+					.count());
+		Assertions
+			.assertEquals(
+				1,
+				verify
+					.filter(
+						"id = '50|_____OmicsDI::278d318ee9f051971236234b181d79ce' and value = 'egi::virtual::10256' " +
+							"and trust = '0.9'")
+					.count());
+
+		Assertions.assertEquals(2, verify.filter("id = '50|dedup_wf_001::be0f7ddf838f07be2ab62e343244a255'").count());
+		Assertions
+			.assertEquals(
+				1,
+				verify
+					.filter("id = '50|dedup_wf_001::be0f7ddf838f07be2ab62e343244a255' and value = 'egi::virtual::150'")
+					.count());
+		Assertions
+			.assertEquals(
+				1,
+				verify
+					.filter(
+						"id = '50|dedup_wf_001::be0f7ddf838f07be2ab62e343244a255' and value = 'egi::virtual::71' " +
+							"and trust = '0.9'")
+					.count());
+
+	}
+
+	@Test
+	public void testCommunities() {
+		PrepareInfo pi = new PrepareInfo(false,
+			workingDir.toString() + "/preparedInfo",
+			getClass().getResource("/eu/dnetlib/dhp/actionset/step1/asInputPath").getPath(),
+			getClass().getResource("/eu/dnetlib/dhp/actionset/step1").getPath(),
+			new Gson().fromJson("[\"iis-communities\"]", List.class),
+			actionSetList);
+
+		pi.run();
+
+		final JavaSparkContext sc = new JavaSparkContext(spark.sparkContext());
+
+		JavaRDD<ASResultInfo> tmp = sc
+			.textFile(workingDir.toString() + "/preparedInfo/actionset")
+			.map(item -> OBJECT_MAPPER.readValue(item, ASResultInfo.class));
+
+		Dataset<ASResultInfo> verificationDataset = spark.createDataset(tmp.rdd(), Encoders.bean(ASResultInfo.class));
+
+		verificationDataset.createOrReplaceTempView("verificationDataset");
+
+		Dataset<Row> verify = spark
+			.sql(
+				("SELECT id, type, val.value value, val.trust trust, val.inference_provenance prov " +
+					"FROM verificationDataset " +
+					"LATERAL VIEW EXPLODE(value) v as val"));
+
+		Assertions.assertEquals(0, verificationDataset.filter("type = 'relation'").count());
+
+		Assertions.assertEquals(0, verify.filter("prov != 'iis::document_community'").count());
+
+		Assertions.assertEquals(1129, verificationDataset.count());
+
+		Assertions.assertEquals(1395, verify.count());
+
+		Assertions.assertEquals(1, verify.filter("id = '50|core_ac_uk__::2aab0d504dae88edc6d7214f4ab62e4f'").count());
+		Assertions
+			.assertEquals(
+				1, verify
+					.filter(
+						"id = '50|core_ac_uk__::2aab0d504dae88edc6d7214f4ab62e4f' and value = 'dariah' " +
+							"and trust = '0.9'")
+					.count());
+
+		Assertions.assertEquals(2, verify.filter("id = '50|core_ac_uk__::2e09ba90d7f712f24a4f48b39571f15f'").count());
+		Assertions
+			.assertEquals(
+				1,
+				verify.filter("id = '50|core_ac_uk__::2e09ba90d7f712f24a4f48b39571f15f' and value = 'clarin'").count());
+		Assertions
+			.assertEquals(
+				1,
+				verify
+					.filter(
+						"id = '50|core_ac_uk__::2e09ba90d7f712f24a4f48b39571f15f' and value = 'dh-ch::subcommunity::2' "
+							+
+							"and trust = '0.9'")
+					.count());
+
+	}
+
+	@Test
+	public void testCovid19() {
+		PrepareInfo pi = new PrepareInfo(false,
+			workingDir.toString() + "/preparedInfo",
+			getClass().getResource("/eu/dnetlib/dhp/actionset/step1/asInputPath").getPath(),
+			getClass().getResource("/eu/dnetlib/dhp/actionset/step1").getPath(),
+			new Gson().fromJson("[\"iis-covid-19\"]", List.class),
+			actionSetList);
+
+		pi.run();
+
+		final JavaSparkContext sc = new JavaSparkContext(spark.sparkContext());
+
+		JavaRDD<ASResultInfo> tmp = sc
+			.textFile(workingDir.toString() + "/preparedInfo/actionset")
+			.map(item -> OBJECT_MAPPER.readValue(item, ASResultInfo.class));
+
+		Dataset<ASResultInfo> verificationDataset = spark.createDataset(tmp.rdd(), Encoders.bean(ASResultInfo.class));
+
+		verificationDataset.createOrReplaceTempView("verificationDataset");
+
+		Dataset<Row> verify = spark
+			.sql(
+				("SELECT id, type, val.value value, val.trust trust, val.inference_provenance prov " +
+					"FROM verificationDataset " +
+					"LATERAL VIEW EXPLODE(value) v as val"));
+
+		Assertions.assertEquals(0, verificationDataset.filter("type = 'relation'").count());
+
+		Assertions.assertEquals(0, verify.filter("prov != 'iis::document_covid19'").count());
+
+		Assertions.assertEquals(45093, verify.filter("value = 'covid-19'").count());
+
+		Assertions.assertEquals(45093, verificationDataset.count());
+
+		Assertions
+			.assertEquals(
+				3, verify
+					.filter(
+						"id = '50|_____OmicsDI::039dbb63f11b19dc15113b34ebceb0d2' or " +
+							"id = '50|_____OmicsDI::05f133acca27d72866c6720a95515f57' or " +
+							"id = '50|_____OmicsDI::19c2cff8e86d7ae39f7e34f43ee06735'")
+					.count());
+
+	}
+
+	@Test
+	public void testAll() {
+		PrepareInfo pi = new PrepareInfo(false,
+			workingDir.toString() + "/preparedInfo",
+			getClass().getResource("/eu/dnetlib/dhp/actionset/step1/asInputPath").getPath(),
+			getClass().getResource("/eu/dnetlib/dhp/actionset/step1").getPath(),
+			new Gson()
+				.fromJson(
+					"[\"iis-researchinitiative\",\"iis-referenced-projects-main\",\"iis-communities\",\"iis-covid-19\"]",
+					List.class),
+			actionSetList);
+
+		pi.run();
+
+		final JavaSparkContext sc = new JavaSparkContext(spark.sparkContext());
+
+		JavaRDD<ASResultInfo> tmp = sc
+			.textFile(workingDir.toString() + "/preparedInfo/actionset")
+			.map(item -> OBJECT_MAPPER.readValue(item, ASResultInfo.class));
+
+		Dataset<ASResultInfo> verificationDataset = spark.createDataset(tmp.rdd(), Encoders.bean(ASResultInfo.class));
+
+		verificationDataset.createOrReplaceTempView("verificationDataset");
+
+		Dataset<Row> verify = spark
+			.sql(
+				("SELECT id, type, val.value value, val.trust trust, val.inference_provenance prov " +
+					"FROM verificationDataset " +
+					"LATERAL VIEW EXPLODE(value) v as val"));
+
+		Assertions.assertEquals(1768215, verificationDataset.filter("type = 'relation'").count());
+		Assertions.assertEquals((45093 + 1129 + 14505 + 44804), verificationDataset.filter("type = 'result'").count());
+
+	}
+
+	@Test
+	public void testRelationMerged() {
+		PrepareInfo pi = new PrepareInfo(false,
+			workingDir.toString() + "/preparedInfo",
+			getClass().getResource("/eu/dnetlib/dhp/actionset/step1/asInputPath").getPath(),
+			getClass().getResource("/eu/dnetlib/dhp/actionset/step1").getPath(),
+			new Gson().fromJson("[]", List.class),
+			actionSetList);
+
+		pi.run();
+
+		final JavaSparkContext sc = new JavaSparkContext(spark.sparkContext());
+
+		JavaRDD<RelationMerges> tmp = sc
+			.textFile(workingDir.toString() + "/preparedInfo/relation")
+			.map(item -> OBJECT_MAPPER.readValue(item, RelationMerges.class));
+
+		Dataset<RelationMerges> verificationDataset = spark
+			.createDataset(tmp.rdd(), Encoders.bean(RelationMerges.class));
+
+		verificationDataset.show(false);
+		Assertions.assertEquals(3, verificationDataset.count());
+
+		verificationDataset.createOrReplaceTempView("verificationDataset");
+
+		Dataset<Row> verify = spark
+			.sql(
+				"Select dedupId, me merged " +
+					"from verificationDataset " +
+					"lateral view explode(merges) m as me");
+
+		Assertions.assertEquals(8, verify.count());
+		Assertions
+			.assertEquals(2, verify.filter("dedupId = '50|dedup_wf_001::1cba00616e303863c34fadaf797d0f8f'").count());
+		Assertions
+			.assertEquals(2, verify.filter("dedupId = '50|dedup_wf_001::7df4b3b26df271628a837c209516902a'").count());
+		Assertions
+			.assertEquals(4, verify.filter("dedupId = '50|dedup_wf_001::b04d742132c133177e996add1325ec04'").count());
+
+		Assertions
+			.assertEquals(
+				2, verify
+					.filter(
+						"dedupId = '50|dedup_wf_001::1cba00616e303863c34fadaf797d0f8f' " +
+							"and (me ='50|base_oa_____::fb2c70723d74f45329640255a959333d' or me = '50|doiboost____::fb2c70723d74f45329640255a959333d')")
+					.count());
+
+		Assertions
+			.assertEquals(
+				2, verify
+					.filter(
+						"dedupId = '50|dedup_wf_001::7df4b3b26df271628a837c209516902a' " +
+							"and (me ='50|doiboost____::78329557c23bee513963ebf295d1434d' or me = '50|doiboost____::8978b9b797294da5306950a94a58d98c')")
+					.count());
+
+		Assertions
+			.assertEquals(
+				4, verify
+					.filter(
+						"dedupId = '50|dedup_wf_001::b04d742132c133177e996add1325ec04' " +
+							"and (me = '50|od______3515::779de9b3a2d224779be52fae43b5fc80' or me = '50|doiboost____::0f10b8f21b7925a344f41edb774f0b0a' "
+							+
+							"or me = '50|od_______166::779de9b3a2d224779be52fae43b5fc80' or me = '50|od_______165::779de9b3a2d224779be52fae43b5fc80')")
+					.count());
+
+	}
+}

dhp-workflows/dhp-actionset-remapping/src/test/java/eu/dnetlib/dhp/actionset/RedistributeIISLinksTest.java

@@ -0,0 +1,183 @@
+
+package eu.dnetlib.dhp.actionset;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+
+import org.apache.commons.io.FileUtils;
+import org.apache.hadoop.io.Text;
+import org.apache.spark.SparkConf;
+import org.apache.spark.api.java.JavaRDD;
+import org.apache.spark.api.java.JavaSparkContext;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Encoders;
+import org.apache.spark.sql.SparkSession;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.Assertions;
+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.actionset.stable.SparkRedistributeIISRelations;
+import eu.dnetlib.dhp.actionset.stable.SparkRedistributeIISResults;
+import eu.dnetlib.dhp.schema.action.AtomicAction;
+import eu.dnetlib.dhp.schema.oaf.Relation;
+import eu.dnetlib.dhp.schema.oaf.Result;
+
+public class RedistributeIISLinksTest {
+	private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+
+	private static SparkSession spark;
+
+	private static Path workingDir;
+	private static final Logger log = LoggerFactory
+		.getLogger(RedistributeIISLinksTest.class);
+
+	@BeforeAll
+	public static void beforeAll() throws IOException {
+		workingDir = Files
+			.createTempDirectory(RedistributeIISLinksTest.class.getSimpleName());
+		log.info("using work dir {}", workingDir);
+
+		SparkConf conf = new SparkConf();
+		conf.setAppName(RedistributeIISLinksTest.class.getSimpleName());
+
+		conf.setMaster("local[*]");
+		conf.set("spark.driver.host", "localhost");
+		conf.set("hive.metastore.local", "true");
+		conf.set("spark.ui.enabled", "false");
+		conf.set("spark.sql.warehouse.dir", workingDir.toString());
+		conf.set("hive.metastore.warehouse.dir", workingDir.resolve("warehouse").toString());
+
+		spark = SparkSession
+			.builder()
+			.appName(RedistributeIISLinksTest.class.getSimpleName())
+			.config(conf)
+			.getOrCreate();
+	}
+
+	@AfterAll
+	public static void afterAll() throws IOException {
+		FileUtils.deleteDirectory(workingDir.toFile());
+		spark.stop();
+	}
+
+	@Test
+	public void redistributeRelationTest() throws Exception {
+		SparkRedistributeIISRelations
+			.main(
+				new String[] {
+					"-isSparkSessionManaged",
+					Boolean.FALSE.toString(),
+					"-asInputPath",
+					getClass()
+						.getResource("/eu/dnetlib/dhp/actionset/step4/actionset")
+						.getPath(),
+					"-outputPath",
+					workingDir.toString() + "/relationActionSet",
+					"-inputPath",
+					getClass()
+						.getResource("/eu/dnetlib/dhp/actionset/step4/result")
+						.getPath()
+				});
+
+		final JavaSparkContext sc = new JavaSparkContext(spark.sparkContext());
+
+		JavaRDD<Relation> tmp = sc
+			.sequenceFile(workingDir.toString() + "/relationActionSet", Text.class, Text.class)
+			.map(item -> OBJECT_MAPPER.readValue(item._2().toString(), AtomicAction.class))
+			.map(aa -> ((Relation) aa.getPayload()));
+
+		Dataset<Relation> verificationDataset = spark.createDataset(tmp.rdd(), Encoders.bean(Relation.class));
+
+		Assertions.assertEquals(14, verificationDataset.count());
+
+		Assertions
+			.assertEquals(
+				3, verificationDataset
+					.filter(
+						"source = '50|doiboost____::0f10b8f21b7925a344f41edb774f0b0a' and " +
+							"(target = '40|rcuk________::8dec51859e6b66cd040670b432b9e59c' or " +
+							"target = '40|rcuk________::5e312e08bd65f126d7d79b3d1d677eb3' or " +
+							"target = '40|corda_______::6d500f8fceb2bb81b0750820469e1cd8')")
+					.count());
+
+		Assertions
+			.assertEquals(
+				2, verificationDataset
+					.filter(
+						"source = '50|doiboost____::fb2c70723d74f45329640255a959333d' and " +
+							"(target = '40|dfgf________::7aa0b0185d7db055823cd0734ddd6521' or target = '40|anr_________::5437f242b18aa615acf57dced27975c6') ")
+					.count());
+
+		Assertions
+			.assertEquals(
+				2, verificationDataset
+					.filter(
+						"source = '50|doiboost____::8978b9b797294da5306950a94a58d98c' and " +
+							"(target = '40|anr_________::55e85886263bf5abe9e28ba4fda9f4ce' or target = '40|anr_________::5eafc553789cd97a12cab7ed1742e2ca') ")
+					.count());
+
+		Assertions
+			.assertEquals(
+				1, verificationDataset.filter("target = '40|dfgf________::7aa0b0185d7db055823cd0734ddd6521'").count());
+		Assertions
+			.assertEquals(
+				1, verificationDataset.filter("target = '40|anr_________::5437f242b18aa615acf57dced27975c6'").count());
+		Assertions
+			.assertEquals(
+				1, verificationDataset.filter("target = '40|anr_________::5eafc553789cd97a12cab7ed1742e2ca'").count());
+		Assertions
+			.assertEquals(
+				1, verificationDataset.filter("target = '40|anr_________::55e85886263bf5abe9e28ba4fda9f4ce'").count());
+		Assertions
+			.assertEquals(
+				1, verificationDataset.filter("target = '40|corda_______::6d500f8fceb2bb81b0750820469e1cd8'").count());
+		Assertions
+			.assertEquals(
+				1, verificationDataset.filter("target = '40|rcuk________::5e312e08bd65f126d7d79b3d1d677eb3'").count());
+		Assertions
+			.assertEquals(
+				1, verificationDataset.filter("target = '40|rcuk________::8dec51859e6b66cd040670b432b9e59c'").count());
+
+	}
+
+	@Test
+	public void redistributeTestResult() throws Exception {
+		SparkRedistributeIISResults
+			.main(
+				new String[] {
+					"-isSparkSessionManaged",
+					Boolean.FALSE.toString(),
+					"-asInputPath",
+					getClass()
+						.getResource("/eu/dnetlib/dhp/actionset/step4/actionset")
+						.getPath(),
+					"-outputPath",
+					workingDir.toString() + "/resultActionSet",
+					"-inputPath",
+					getClass()
+						.getResource("/eu/dnetlib/dhp/actionset/step4/result")
+						.getPath()
+				});
+
+		final JavaSparkContext sc = new JavaSparkContext(spark.sparkContext());
+
+		JavaRDD<Result> tmp = sc
+			.sequenceFile(workingDir.toString() + "/resultActionSet", Text.class, Text.class)
+			.map(item -> OBJECT_MAPPER.readValue(item._2().toString(), AtomicAction.class))
+			.map(aa -> ((Result) aa.getPayload()));
+
+		Dataset<Result> verificationDataset = spark.createDataset(tmp.rdd(), Encoders.bean(Result.class));
+
+		Assertions.assertEquals(2, verificationDataset.count());
+
+		verificationDataset.createOrReplaceTempView("verificationDataset");
+
+	}
+
+}

dhp-workflows/dhp-actionset-remapping/src/test/java/eu/dnetlib/dhp/actionset/SelectResultTest.java

@@ -0,0 +1,96 @@
+
+package eu.dnetlib.dhp.actionset;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+
+import org.apache.commons.io.FileUtils;
+import org.apache.spark.SparkConf;
+import org.apache.spark.api.java.JavaRDD;
+import org.apache.spark.api.java.JavaSparkContext;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Encoders;
+import org.apache.spark.sql.SparkSession;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.Assertions;
+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.actionset.stable.ResultPid;
+import eu.dnetlib.dhp.actionset.stable.SparkSelectResults;
+
+public class SelectResultTest {
+	private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+
+	private static SparkSession spark;
+
+	private static Path workingDir;
+	private static final Logger log = LoggerFactory
+		.getLogger(SelectResultTest.class);
+
+	@BeforeAll
+	public static void beforeAll() throws IOException {
+		workingDir = Files
+			.createTempDirectory(SelectResultTest.class.getSimpleName());
+		log.info("using work dir {}", workingDir);
+
+		SparkConf conf = new SparkConf();
+		conf.setAppName(SelectResultTest.class.getSimpleName());
+
+		conf.setMaster("local[*]");
+		conf.set("spark.driver.host", "localhost");
+		conf.set("hive.metastore.local", "true");
+		conf.set("spark.ui.enabled", "false");
+		conf.set("spark.sql.warehouse.dir", workingDir.toString());
+		conf.set("hive.metastore.warehouse.dir", workingDir.resolve("warehouse").toString());
+
+		spark = SparkSession
+			.builder()
+			.appName(SelectResultTest.class.getSimpleName())
+			.config(conf)
+			.getOrCreate();
+	}
+
+	@AfterAll
+	public static void afterAll() throws IOException {
+		FileUtils.deleteDirectory(workingDir.toFile());
+		spark.stop();
+	}
+
+	@Test
+	public void testSelectResult() throws Exception {
+		SparkSelectResults
+			.main(
+				new String[] {
+					"-isSparkSessionManaged",
+					Boolean.FALSE.toString(),
+					"-inputPath",
+					getClass()
+						.getResource("/eu/dnetlib/dhp/actionset/step3")
+						.getPath(),
+					"-outputPath",
+					workingDir.toString() + "/selectedResults"
+				});
+
+		final JavaSparkContext sc = new JavaSparkContext(spark.sparkContext());
+
+		JavaRDD<ResultPid> tmp = sc
+			.textFile(workingDir.toString() + "/selectedResults")
+			.map(item -> OBJECT_MAPPER.readValue(item, ResultPid.class));
+
+		Dataset<ResultPid> verificationDataset = spark.createDataset(tmp.rdd(), Encoders.bean(ResultPid.class));
+
+		verificationDataset.show(false);
+
+		Assertions.assertEquals(5, verificationDataset.count());
+		Assertions.assertEquals(3, verificationDataset.filter("substr(resultId,1,11) = '50|doiboost'").count());
+		Assertions.assertEquals(2, verificationDataset.filter("substr(resultId,1,11) = '50|datacite'").count());
+
+	}
+
+}

dhp-workflows/dhp-actionset-remapping/src/test/resources/eu/dnetlib/dhp/actionset/step1/relation/relation

@@ -0,0 +1,12 @@
+{"collectedfrom":null,"dataInfo":{"invisible":false,"inferred":true,"deletedbyinference":false,"trust":"0.7348","inferenceprovenance":"iis::document_referencedProjects","provenanceaction":{"classid":"iis","classname":"iis","schemeid":"dnet:provenanceActions","schemename":"dnet:provenanceActions"}},"lastupdatetimestamp":1594214129915,"relType":"resultResult","subRelType":"dedup","relClass":"merges","source":"50|dedup_wf_001::1cba00616e303863c34fadaf797d0f8f","target":"50|base_oa_____::fb2c70723d74f45329640255a959333d","validated":null,"validationDate":null,"properties":[]}
+{"collectedfrom":null,"dataInfo":{"invisible":false,"inferred":true,"deletedbyinference":false,"trust":"0.7348","inferenceprovenance":"iis::document_referencedProjects","provenanceaction":{"classid":"iis","classname":"iis","schemeid":"dnet:provenanceActions","schemename":"dnet:provenanceActions"}},"lastupdatetimestamp":1594214129915,"relType":"resultProject","subRelType":"outcome","relClass":"merges","source":"50|dedup_wf_001::1cba00616e303863c34fadaf797d0f8f","target":"50|doiboost____::fb2c70723d74f45329640255a959333d","validated":null,"validationDate":null,"properties":[]}
+{"collectedfrom":null,"dataInfo":{"invisible":false,"inferred":true,"deletedbyinference":false,"trust":"0.7348","inferenceprovenance":"iis::document_referencedProjects","provenanceaction":{"classid":"iis","classname":"iis","schemeid":"dnet:provenanceActions","schemename":"dnet:provenanceActions"}},"lastupdatetimestamp":1594214129915,"relType":"resultProject","subRelType":"outcome","relClass":"merges","source":"50|dedup_wf_001::7df4b3b26df271628a837c209516902a","target":"50|doiboost____::8978b9b797294da5306950a94a58d98c","validated":null,"validationDate":null,"properties":[]}
+{"collectedfrom":null,"dataInfo":{"invisible":false,"inferred":true,"deletedbyinference":false,"trust":"0.7348","inferenceprovenance":"iis::document_referencedProjects","provenanceaction":{"classid":"iis","classname":"iis","schemeid":"dnet:provenanceActions","schemename":"dnet:provenanceActions"}},"lastupdatetimestamp":1594214129915,"relType":"resultProject","subRelType":"outcome","relClass":"merges","source":"50|dedup_wf_001::7df4b3b26df271628a837c209516902a","target":"50|doiboost____::78329557c23bee513963ebf295d1434d","validated":null,"validationDate":null,"properties":[]}
+{"collectedfrom":null,"dataInfo":{"invisible":false,"inferred":true,"deletedbyinference":false,"trust":"0.7348","inferenceprovenance":"iis::document_referencedProjects","provenanceaction":{"classid":"iis","classname":"iis","schemeid":"dnet:provenanceActions","schemename":"dnet:provenanceActions"}},"lastupdatetimestamp":1594214129915,"relType":"resultProject","subRelType":"outcome","relClass":"merges","source":"50|dedup_wf_001::b04d742132c133177e996add1325ec04","target":"50|od______3515::779de9b3a2d224779be52fae43b5fc80","validated":null,"validationDate":null,"properties":[]}
+{"collectedfrom":null,"dataInfo":{"invisible":false,"inferred":true,"deletedbyinference":false,"trust":"0.7348","inferenceprovenance":"iis::document_referencedProjects","provenanceaction":{"classid":"iis","classname":"iis","schemeid":"dnet:provenanceActions","schemename":"dnet:provenanceActions"}},"lastupdatetimestamp":1594214129915,"relType":"resultProject","subRelType":"outcome","relClass":"merges","source":"50|dedup_wf_001::b04d742132c133177e996add1325ec04","target":"50|doiboost____::0f10b8f21b7925a344f41edb774f0b0a","validated":null,"validationDate":null,"properties":[]}
+{"collectedfrom":null,"dataInfo":{"invisible":false,"inferred":true,"deletedbyinference":false,"trust":"0.7348","inferenceprovenance":"iis::document_referencedProjects","provenanceaction":{"classid":"iis","classname":"iis","schemeid":"dnet:provenanceActions","schemename":"dnet:provenanceActions"}},"lastupdatetimestamp":1594214129915,"relType":"resultProject","subRelType":"outcome","relClass":"merges","source":"50|dedup_wf_001::b04d742132c133177e996add1325ec04","target":"50|od_______166::779de9b3a2d224779be52fae43b5fc80","validated":null,"validationDate":null,"properties":[]}
+{"collectedfrom":null,"dataInfo":{"invisible":false,"inferred":true,"deletedbyinference":false,"trust":"0.7348","inferenceprovenance":"iis::document_referencedProjects","provenanceaction":{"classid":"iis","classname":"iis","schemeid":"dnet:provenanceActions","schemename":"dnet:provenanceActions"}},"lastupdatetimestamp":1594214129915,"relType":"resultProject","subRelType":"outcome","relClass":"merges","source":"50|dedup_wf_001::b04d742132c133177e996add1325ec04","target":"50|od_______165::779de9b3a2d224779be52fae43b5fc80","validated":null,"validationDate":null,"properties":[]}
+{"collectedfrom":null,"dataInfo":{"invisible":false,"inferred":true,"deletedbyinference":false,"trust":"0.7348","inferenceprovenance":"iis::document_referencedProjects","provenanceaction":{"classid":"iis","classname":"iis","schemeid":"dnet:provenanceActions","schemename":"dnet:provenanceActions"}},"lastupdatetimestamp":1594214129915,"relType":"resultProject","subRelType":"outcome","relClass":"produces","source":"40|nsf_________::2bedb915e92b7dd25b082c6c2f241085","target":"50|od________18::c8e57f11074407d59f7114f047afd54e","validated":null,"validationDate":null,"properties":[]}
+{"collectedfrom":null,"dataInfo":{"invisible":false,"inferred":true,"deletedbyinference":false,"trust":"0.7348","inferenceprovenance":"iis::document_referencedProjects","provenanceaction":{"classid":"iis","classname":"iis","schemeid":"dnet:provenanceActions","schemename":"dnet:provenanceActions"}},"lastupdatetimestamp":1594214129915,"relType":"resultProject","subRelType":"outcome","relClass":"produces","source":"40|nsf_________::2bedb915e92b7dd25b082c6c2f241085","target":"50|od________18::c8e57f11074407d59f7114f047afd54e","validated":null,"validationDate":null,"properties":[]}
+{"collectedfrom":null,"dataInfo":{"invisible":false,"inferred":true,"deletedbyinference":false,"trust":"0.7348","inferenceprovenance":"iis::document_referencedProjects","provenanceaction":{"classid":"iis","classname":"iis","schemeid":"dnet:provenanceActions","schemename":"dnet:provenanceActions"}},"lastupdatetimestamp":1594214129915,"relType":"resultProject","subRelType":"outcome","relClass":"produces","source":"40|nsf_________::2bedb915e92b7dd25b082c6c2f241085","target":"50|od________18::c8e57f11074407d59f7114f047afd54e","validated":null,"validationDate":null,"properties":[]}
+{"collectedfrom":null,"dataInfo":{"invisible":false,"inferred":true,"deletedbyinference":false,"trust":"0.7348","inferenceprovenance":"iis::document_referencedProjects","provenanceaction":{"classid":"iis","classname":"iis","schemeid":"dnet:provenanceActions","schemename":"dnet:provenanceActions"}},"lastupdatetimestamp":1594214129915,"relType":"resultProject","subRelType":"outcome","relClass":"produces","source":"40|nsf_________::2bedb915e92b7dd25b082c6c2f241085","target":"50|od________18::c8e57f11074407d59f7114f047afd54e","validated":null,"validationDate":null,"properties":[]}

dhp-workflows/dhp-actionset-remapping/src/test/resources/eu/dnetlib/dhp/actionset/step2/preparedInfo/actionset/as

@@ -0,0 +1,12 @@
+{"id":"50|dedup_wf_001::1cba00616e303863c34fadaf797d0f8f","type":"relation","value":[{"inference_provenance":"iis::document_referencedProjects","trust":"0.72","value":"40|dfgf________::7aa0b0185d7db055823cd0734ddd6521"}]}
+{"id":"50|dedup_wf_001::1cba00616e303863c34fadaf797d0f8f","type":"relation","value":[{"inference_provenance":"iis::document_referencedProjects","trust":"0.72","value":"40|anr_________::5437f242b18aa615acf57dced27975c6"}]}
+{"id":"50|dedup_wf_001::7df4b3b26df271628a837c209516902a","type":"relation","value":[{"inference_provenance":"iis::document_referencedProjects","trust":"0.72","value":"40|anr_________::5eafc553789cd97a12cab7ed1742e2ca"}]}
+{"id":"50|dedup_wf_001::7df4b3b26df271628a837c209516902a","type":"relation","value":[{"inference_provenance":"iis::document_referencedProjects","trust":"0.72","value":"40|anr_________::55e85886263bf5abe9e28ba4fda9f4ce"}]}
+{"id":"50|dedup_wf_001::b04d742132c133177e996add1325ec04","type":"relation","value":[{"inference_provenance":"iis::document_referencedProjects","trust":"0.897","value":"40|rcuk________::8dec51859e6b66cd040670b432b9e59c"}]}
+{"id":"50|dedup_wf_001::b04d742132c133177e996add1325ec04","type":"relation","value":[{"inference_provenance":"iis::document_referencedProjects","trust":"0.897","value":"40|rcuk________::5e312e08bd65f126d7d79b3d1d677eb3"}]}
+{"id":"50|dedup_wf_001::b04d742132c133177e996add1325ec04","type":"relation","value":[{"inference_provenance":"iis::document_referencedProjects","trust":"0.7085","value":"40|corda_______::6d500f8fceb2bb81b0750820469e1cd8"}]}
+{"id":"50|_____OmicsDI::039dbb63f11b19dc15113b34ebceb0d2","type":"result","value":[{"inference_provenance":"iis::document_covid19","trust":"0.9","value":"covid-19"}]}
+{"id":"50|_____OmicsDI::05f133acca27d72866c6720a95515f57","type":"result","value":[{"inference_provenance":"iis::document_covid19","trust":"0.9","value":"covid-19"}]}
+{"id":"50|_____OmicsDI::2d508eba981699a30e969d1ab5a068b8","type":"result","value":[{"inference_provenance":"iis::document_covid19","trust":"0.9","value":"covid-19"}]}
+{"id":"50|datacite____::00bddedc38dc045780dc84c27bc8fecd","type":"result","value":[{"inference_provenance":"iis::document_covid19","trust":"0.9","value":"covid-19"}]}
+{"id":"50|datacite____::00f7f89392fa75e944dc8d329e9e8024","type":"result","value":[{"inference_provenance":"iis::document_covid19","trust":"0.9","value":"covid-19"}]}

dhp-workflows/dhp-actionset-remapping/src/test/resources/eu/dnetlib/dhp/actionset/step2/preparedInfo/relation/relations

@@ -0,0 +1,3 @@
+{"dedupId":"50|dedup_wf_001::b04d742132c133177e996add1325ec04","merges":["50|doiboost____::0f10b8f21b7925a344f41edb774f0b0a","50|od_______166::779de9b3a2d224779be52fae43b5fc80","50|od_______165::779de9b3a2d224779be52fae43b5fc80","50|od______3515::779de9b3a2d224779be52fae43b5fc80"]}
+{"dedupId":"50|dedup_wf_001::7df4b3b26df271628a837c209516902a","merges":["50|doiboost____::78329557c23bee513963ebf295d1434d","50|doiboost____::8978b9b797294da5306950a94a58d98c"]}
+{"dedupId":"50|dedup_wf_001::1cba00616e303863c34fadaf797d0f8f","merges":["50|doiboost____::fb2c70723d74f45329640255a959333d","50|base_oa_____::fb2c70723d74f45329640255a959333d"]}

dhp-workflows/dhp-actionset-remapping/src/test/resources/eu/dnetlib/dhp/actionset/step4/actionset/expandedAs

@@ -0,0 +1,25 @@
+{"id":"50|doiboost____::fb2c70723d74f45329640255a959333d","type":"relation","value":[{"inference_provenance":"iis::document_referencedProjects","trust":"0.72","value":"40|dfgf________::7aa0b0185d7db055823cd0734ddd6521"}]}
+{"id":"50|base_oa_____::fb2c70723d74f45329640255a959333d","type":"relation","value":[{"inference_provenance":"iis::document_referencedProjects","trust":"0.72","value":"40|dfgf________::7aa0b0185d7db055823cd0734ddd6521"}]}
+{"id":"50|doiboost____::fb2c70723d74f45329640255a959333d","type":"relation","value":[{"inference_provenance":"iis::document_referencedProjects","trust":"0.72","value":"40|anr_________::5437f242b18aa615acf57dced27975c6"}]}
+{"id":"50|base_oa_____::fb2c70723d74f45329640255a959333d","type":"relation","value":[{"inference_provenance":"iis::document_referencedProjects","trust":"0.72","value":"40|anr_________::5437f242b18aa615acf57dced27975c6"}]}
+{"id":"50|doiboost____::78329557c23bee513963ebf295d1434d","type":"relation","value":[{"inference_provenance":"iis::document_referencedProjects","trust":"0.72","value":"40|anr_________::5eafc553789cd97a12cab7ed1742e2ca"}]}
+{"id":"50|doiboost____::8978b9b797294da5306950a94a58d98c","type":"relation","value":[{"inference_provenance":"iis::document_referencedProjects","trust":"0.72","value":"40|anr_________::5eafc553789cd97a12cab7ed1742e2ca"}]}
+{"id":"50|doiboost____::78329557c23bee513963ebf295d1434d","type":"relation","value":[{"inference_provenance":"iis::document_referencedProjects","trust":"0.72","value":"40|anr_________::55e85886263bf5abe9e28ba4fda9f4ce"}]}
+{"id":"50|doiboost____::8978b9b797294da5306950a94a58d98c","type":"relation","value":[{"inference_provenance":"iis::document_referencedProjects","trust":"0.72","value":"40|anr_________::55e85886263bf5abe9e28ba4fda9f4ce"}]}
+{"id":"50|doiboost____::0f10b8f21b7925a344f41edb774f0b0a","type":"relation","value":[{"inference_provenance":"iis::document_referencedProjects","trust":"0.897","value":"40|rcuk________::8dec51859e6b66cd040670b432b9e59c"}]}
+{"id":"50|od_______166::779de9b3a2d224779be52fae43b5fc80","type":"relation","value":[{"inference_provenance":"iis::document_referencedProjects","trust":"0.897","value":"40|rcuk________::8dec51859e6b66cd040670b432b9e59c"}]}
+{"id":"50|od_______165::779de9b3a2d224779be52fae43b5fc80","type":"relation","value":[{"inference_provenance":"iis::document_referencedProjects","trust":"0.897","value":"40|rcuk________::8dec51859e6b66cd040670b432b9e59c"}]}
+{"id":"50|od______3515::779de9b3a2d224779be52fae43b5fc80","type":"relation","value":[{"inference_provenance":"iis::document_referencedProjects","trust":"0.897","value":"40|rcuk________::8dec51859e6b66cd040670b432b9e59c"}]}
+{"id":"50|doiboost____::0f10b8f21b7925a344f41edb774f0b0a","type":"relation","value":[{"inference_provenance":"iis::document_referencedProjects","trust":"0.897","value":"40|rcuk________::5e312e08bd65f126d7d79b3d1d677eb3"}]}
+{"id":"50|od_______166::779de9b3a2d224779be52fae43b5fc80","type":"relation","value":[{"inference_provenance":"iis::document_referencedProjects","trust":"0.897","value":"40|rcuk________::5e312e08bd65f126d7d79b3d1d677eb3"}]}
+{"id":"50|od_______165::779de9b3a2d224779be52fae43b5fc80","type":"relation","value":[{"inference_provenance":"iis::document_referencedProjects","trust":"0.897","value":"40|rcuk________::5e312e08bd65f126d7d79b3d1d677eb3"}]}
+{"id":"50|od______3515::779de9b3a2d224779be52fae43b5fc80","type":"relation","value":[{"inference_provenance":"iis::document_referencedProjects","trust":"0.897","value":"40|rcuk________::5e312e08bd65f126d7d79b3d1d677eb3"}]}
+{"id":"50|doiboost____::0f10b8f21b7925a344f41edb774f0b0a","type":"relation","value":[{"inference_provenance":"iis::document_referencedProjects","trust":"0.7085","value":"40|corda_______::6d500f8fceb2bb81b0750820469e1cd8"}]}
+{"id":"50|od_______166::779de9b3a2d224779be52fae43b5fc80","type":"relation","value":[{"inference_provenance":"iis::document_referencedProjects","trust":"0.7085","value":"40|corda_______::6d500f8fceb2bb81b0750820469e1cd8"}]}
+{"id":"50|od_______165::779de9b3a2d224779be52fae43b5fc80","type":"relation","value":[{"inference_provenance":"iis::document_referencedProjects","trust":"0.7085","value":"40|corda_______::6d500f8fceb2bb81b0750820469e1cd8"}]}
+{"id":"50|od______3515::779de9b3a2d224779be52fae43b5fc80","type":"relation","value":[{"inference_provenance":"iis::document_referencedProjects","trust":"0.7085","value":"40|corda_______::6d500f8fceb2bb81b0750820469e1cd8"}]}
+{"id":"50|_____OmicsDI::039dbb63f11b19dc15113b34ebceb0d2","type":"result","value":[{"inference_provenance":"iis::document_covid19","trust":"0.9","value":"covid-19"}]}
+{"id":"50|_____OmicsDI::05f133acca27d72866c6720a95515f57","type":"result","value":[{"inference_provenance":"iis::document_covid19","trust":"0.9","value":"covid-19"}]}
+{"id":"50|_____OmicsDI::2d508eba981699a30e969d1ab5a068b8","type":"result","value":[{"inference_provenance":"iis::document_covid19","trust":"0.9","value":"covid-19"}]}
+{"id":"50|datacite____::00bddedc38dc045780dc84c27bc8fecd","type":"result","value":[{"inference_provenance":"iis::document_covid19","trust":"0.9","value":"covid-19"}]}
+{"id":"50|datacite____::00f7f89392fa75e944dc8d329e9e8024","type":"result","value":[{"inference_provenance":"iis::document_covid19","trust":"0.9","value":"covid-19"}]}

dhp-workflows/dhp-actionset-remapping/src/test/resources/eu/dnetlib/dhp/actionset/step4/result/reswithdoi

@@ -0,0 +1,5 @@
+{"resultId":"50|datacite____::00f7f89392fa75e944dc8d329e9e8024","doi":"10.13140/rg.2.2.12240.46089"}
+{"resultId":"50|doiboost____::0f10b8f21b7925a344f41edb774f0b0a","doi":"10.1111/jace.12635"}
+{"resultId":"50|doiboost____::8978b9b797294da5306950a94a58d98c","doi":"10.1055/s-0038-1671974"}
+{"resultId":"50|doiboost____::fb2c70723d74f45329640255a959333d","doi":"10.1016/j.apcbee.2014.01.002"}
+{"resultId":"50|datacite____::00bddedc38dc045780dc84c27bc8fecd","doi":"10.13140/rg.2.2.28413.92647"}

dhp-workflows/dhp-enrichment/src/main/java/eu/dnetlib/dhp/bulktag/community/ResultTagger.java

@@ -72,8 +72,9 @@ public class ResultTagger implements Serializable {
 		// tagging for Subject
 		final Set<String> subjects = new HashSet<>();
 
-		if (Objects.nonNull(result.getSubject())){
+		if (Objects.nonNull(result.getSubject())) {
-			result.getSubject()
+			result
+				.getSubject()
 				.stream()
 				.map(subject -> subject.getValue())
 				.filter(StringUtils::isNotBlank)
@@ -91,13 +92,13 @@ public class ResultTagger implements Serializable {
 
 		if (Objects.nonNull(result.getInstance())) {
 			for (Instance i : result.getInstance()) {
-				if(Objects.nonNull(i.getCollectedfrom())){
+				if (Objects.nonNull(i.getCollectedfrom())) {
-					if(Objects.nonNull(i.getCollectedfrom().getKey())){
+					if (Objects.nonNull(i.getCollectedfrom().getKey())) {
 						tmp.add(StringUtils.substringAfter(i.getCollectedfrom().getKey(), "|"));
 					}
 				}
-				if(Objects.nonNull(i.getHostedby())){
+				if (Objects.nonNull(i.getHostedby())) {
-					if(Objects.nonNull(i.getHostedby().getKey())){
+					if (Objects.nonNull(i.getHostedby().getKey())) {
 						tmp.add(StringUtils.substringAfter(i.getHostedby().getKey(), "|"));
 					}
 				}

dhp-workflows/pom.xml

@@ -30,6 +30,7 @@
         <module>dhp-stats-update</module>
         <module>dhp-broker-events</module>
         <module>dhp-doiboost</module>
+        <module>dhp-actionset-remapping</module>
     </modules>
 
     <pluginRepositories>