D-Net

package eu.dnetlib.data.mapreduce.hbase.broker.enrich;

import java.io.IOException;

import java.util.Map;

import com.google.common.collect.Iterables;

import eu.dnetlib.data.mapreduce.util.DedupUtils;

import eu.dnetlib.data.mapreduce.util.OafRowKeyDecoder;

import eu.dnetlib.data.mapreduce.util.UpdateMerger;

import eu.dnetlib.data.proto.DedupProtos.Dedup.RelName;

import eu.dnetlib.data.proto.OafProtos.Oaf;

import eu.dnetlib.data.proto.TypeProtos.Type;

import org.apache.commons.collections.MapUtils;

import org.apache.hadoop.hbase.client.Result;

import org.apache.hadoop.hbase.io.ImmutableBytesWritable;

import org.apache.hadoop.hbase.util.Bytes;

/**

 * Created by claudio on 08/07/16.

 */

public class EnrichmentMapper extends AbstractEnrichmentMapper {

	@Override

	protected String counterGroup() {

		return "Broker Enrichment";

	}

	@Override

	protected void map(final ImmutableBytesWritable key, final Result value, final Context context) throws IOException, InterruptedException {

		final String type = OafRowKeyDecoder.decode(key.copyBytes()).getType().toString();

		final Map<byte[], byte[]> result = value.getFamilyMap(Bytes.toBytes(Type.result.name()));

		if (MapUtils.isEmpty(result)) {

			context.getCounter(counterGroup(), type + ": empty family map").increment(1);

			return;

		}

		final Oaf body = UpdateMerger.mergeBodyUpdates(context, result);

		if (body == null) {

			context.getCounter(counterGroup(), type + ": body null").increment(1);

			return;

		}

		final String mergedInCF = DedupUtils.getDedupCF_mergedIn(Type.result);

		final Map<byte[], byte[]> mergedIn = value.getFamilyMap(Bytes.toBytes(mergedInCF));

		final byte[] outKey = getEmitKey(context, key, mergedIn);

		emit(context, outKey, body.toByteArray(), type);

	}

	private byte[] getEmitKey(final Context context, final ImmutableBytesWritable key, final Map<byte[], byte[]> mergedIn) {

		if (MapUtils.isNotEmpty(mergedIn)) {

			context.getCounter(Type.result.name(), RelName.isMergedIn.name()).increment(1);

			return Iterables.getOnlyElement(mergedIn.keySet());

		} else {

			return key.copyBytes();

		}

	}

}

package eu.dnetlib.data.mapreduce.hbase.broker.enrich;

import java.io.IOException;

import java.util.Arrays;

import java.util.Map;

import com.google.common.base.Function;

import com.google.common.base.Predicate;

import com.google.common.collect.Iterables;

import com.google.common.collect.Lists;

import eu.dnetlib.data.mapreduce.util.DedupUtils;

import eu.dnetlib.data.mapreduce.util.OafDecoder;

import eu.dnetlib.data.mapreduce.util.OafRowKeyDecoder;

import eu.dnetlib.data.proto.OafProtos.Oaf;

import eu.dnetlib.data.proto.TypeProtos.Type;

import org.apache.commons.collections.MapUtils;

import org.apache.hadoop.hbase.client.Result;

import org.apache.hadoop.hbase.io.ImmutableBytesWritable;

import org.apache.hadoop.hbase.util.Bytes;

/**

 * Created by claudio on 08/07/16.

 */

public class ProjectEnrichmentMapper extends AbstractEnrichmentMapper {

	@Override

	protected String counterGroup() {

		return "Broker Enrichment projects";

	}

	@Override

	protected void map(final ImmutableBytesWritable key, final Result value, final Context context) throws IOException, InterruptedException {

		final Type type = getEntityType(key);

		final byte[] body = value.getValue(Bytes.toBytes(type.toString()), Bytes.toBytes("body"));

		if (body == null) {

			context.getCounter(counterGroup(), "missing metadata").increment(1);

			return;

		}

		switch (type) {

		case project:

			for (final byte[] resultId : listRelatedIds(value, "resultProject_outcome_produces")) {

				emit(context, resultId, body, "project");

			}

			break;

		case result:

			final Oaf.Builder oafBuilder = Oaf.newBuilder(OafDecoder.decode(body).getOaf());

			for (final String relName : Arrays.asList("resultProject_outcome_isProducedBy")) { // Â TODO add dataset rels

				for (final Oaf rel : listRelations(value, relName)) {

					oafBuilder.getEntityBuilder().addCachedOafRel(rel);

					context.getCounter(counterGroup(), "rel: " + relName).increment(1);

				}

			}

			final Map<byte[], byte[]> mergedIn = value.getFamilyMap(DedupUtils.getDedupCF_mergedInBytes(Type.result));

			if (MapUtils.isEmpty(mergedIn) & !DedupUtils.isRoot(key)) {

				emit(context, key.copyBytes(), oafBuilder.build().toByteArray(), "result not deduped");

			} else if (DedupUtils.isRoot(key)) {

				emit(context, key.copyBytes(), oafBuilder.build().toByteArray(), "result merges");

			} else {

				emit(context, getRootId(mergedIn), oafBuilder.build().toByteArray(), "result mergedIn");

			}

			break;

		default:

			throw new IllegalArgumentException("invalid type: " + type);

		}

	}

	private Iterable<Oaf> listRelations(final Result value, final String relType) {

		//TODO consider only relationshipt not deletedbyinference

		final Map<byte[], byte[]> map = value.getFamilyMap(Bytes.toBytes(relType));

		if (MapUtils.isNotEmpty(map)) {

			return Iterables.filter(Iterables.transform(map.values(), new Function<byte[], Oaf>() {

				@Override

				public Oaf apply(final byte[] input) {

					return OafDecoder.decode(input).getOaf();

				}

			}), new Predicate<Oaf>() {

				@Override

				public boolean apply(final Oaf rel) {

					return !rel.getRel().getTarget().contains("unidentified");

				}

			});

		} else {

			return Lists.newArrayList();

		}

	}

	private Iterable<byte[]> listRelatedIds(final Result value, final String relType) {

		//TODO consider only relationshipt not deletedbyinference

		final Map<byte[], byte[]> map = value.getFamilyMap(Bytes.toBytes(relType));

		if (MapUtils.isNotEmpty(map)) {

			return map.keySet();

		} else {

			return Lists.newArrayList();

		}

	}

	private Type getEntityType(final ImmutableBytesWritable key) {

		return OafRowKeyDecoder.decode(key.copyBytes()).getType();

	}

	private byte[] getRootId(final Map<byte[], byte[]> mergedIn) {

		return Iterables.getOnlyElement(mergedIn.keySet());

	}

}

package eu.dnetlib.data.mapreduce.hbase.broker;

import java.io.IOException;

import java.util.List;

import com.google.common.collect.Lists;

import eu.dnetlib.broker.objects.OpenAireEventPayload;

import eu.dnetlib.data.mapreduce.hbase.broker.mapping.HighlightFactory;

import eu.dnetlib.data.mapreduce.hbase.broker.mapping.OpenAireEventPayloadFactory;

import eu.dnetlib.data.mapreduce.hbase.broker.model.EventWrapper;

import eu.dnetlib.data.proto.FieldTypeProtos.StringField;

import eu.dnetlib.data.proto.OafProtos.Oaf;

import org.apache.commons.lang.StringUtils;

import org.dom4j.DocumentException;

import static eu.dnetlib.data.mapreduce.hbase.broker.mapping.EventFactory.asEvent;

/**

 * Created by claudio on 26/07/16.

 */

public class AbstractEventFactory {

	public static List<EventWrapper> process(final Oaf current, final Oaf other, final float trust)

			throws IOException, InterruptedException, DocumentException {

		return new AbstractEventFactory().processAbstract(current, other, trust);

	}

	private List<EventWrapper> processAbstract(final Oaf current, final Oaf other, final float trust)

			throws IOException, InterruptedException, DocumentException {

		final List<EventWrapper> events = Lists.newArrayList();

		if (hasAbstract(other)) {

			// doProcessAbstract(context, current, other, Topic.MORE_ABSTRACT);

			if (!hasAbstract(current)) {

				events.add(doProcessAbstract(current, other, Topic.ENRICH_MISSING_ABSTRACT, trust));

			}

		}

		return events;

	}

	private EventWrapper doProcessAbstract(final Oaf current, final Oaf other, final Topic topic, final float trust)

			throws IOException, InterruptedException, DocumentException {

		final Oaf.Builder prototype = Oaf.newBuilder(current);

		final List<StringField> descriptionList = other.getEntity().getResult().getMetadata().getDescriptionList();

		prototype.getEntityBuilder().getResultBuilder().getMetadataBuilder().addAllDescription(descriptionList);

		final Oaf oaf = prototype.build();

		final OpenAireEventPayload payload =

				HighlightFactory.highlightEnrichAbstract(OpenAireEventPayloadFactory.fromOAF(oaf.getEntity(), other.getEntity(), trust), descriptionList);

		return EventWrapper.newInstance(

				asEvent(oaf.getEntity(), topic, payload, other.getEntity(), trust),

				topic.getValue());

	}

	private boolean hasAbstract(final Oaf oaf) {

		return oaf.getEntity().getResult().getMetadata().getDescriptionList()

				.stream()

				.anyMatch(s -> StringUtils.isNotBlank(s.getValue()));

	}

}

<?xml version="1.0" ?>

<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/maven-v4_0_0.xsd">

	<parent>

		<groupId>eu.dnetlib</groupId>

		<artifactId>dnet45-parent</artifactId>

		<version>1.0.0</version>

		<relativePath />

	</parent>

	<modelVersion>4.0.0</modelVersion>

	<groupId>eu.dnetlib</groupId>

	<artifactId>dnet-mapreduce-jobs</artifactId>

	<version>1.1.1-PROD</version>

	<packaging>jar</packaging>

	<scm>

		<developerConnection>scm:svn:https://svn.driver.research-infrastructures.eu/driver/dnet45/modules/dnet-mapreduce-jobs/tags/dnet-mapreduce-jobs-1.1.1-PROD</developerConnection>

	</scm>

	<build>

		<plugins>

			<plugin>

				<artifactId>maven-assembly-plugin</artifactId>

				<configuration>

					<archive>

						<manifest>

							<mainClass>eu.dnetlib.data.mapreduce.hbase.dataimport.ImportRecordsJob</mainClass>

						</manifest>

					</archive>

					<descriptorRefs>

						<descriptorRef>jar-with-dependencies</descriptorRef>

					</descriptorRefs>

				</configuration>

			</plugin>

		</plugins>

	</build>

	<repositories>

		<!-- Cloudera Repositories -->

		<repository>

			<snapshots>

				<enabled>false</enabled>

			</snapshots>

			<id>cloudera-central</id>

			<name>cloundera-libs-release</name>

			<url>http://maven.research-infrastructures.eu/nexus/content/repositories/cloudera-central</url>

		</repository>

		<repository>

			<id>cloudera-snapshots</id>

			<name>cloudera-libs-snapshot</name>

			<url>http://maven.research-infrastructures.eu/nexus/content/repositories/cloudera-snapshots</url>

		</repository>

	</repositories>

	<dependencies>

		<dependency>

			<groupId>commons-logging</groupId>

			<artifactId>commons-logging</artifactId>

			<version>${commons.logging.version}</version>

		</dependency>

		<dependency>

			<groupId>junit</groupId>

			<artifactId>junit</artifactId>

			<version>${junit.version}</version>

			<scope>test</scope>

		</dependency>

		<dependency>

			<groupId>com.google.code.gson</groupId>

			<artifactId>gson</artifactId>

			<version>${google.gson.version}</version>

		</dependency>

		<dependency>

			<groupId>eu.dnetlib</groupId>

			<artifactId>dnet-index-solr-common</artifactId>

			<version>[1.0.0,2.0.0)</version>

		</dependency>

		<dependency>

			<groupId>eu.dnetlib</groupId>

			<artifactId>dnet-openaire-broker-common</artifactId>

			<version>[1.0.0,2.0.0)</version>

		</dependency>

		<dependency>

			<groupId>com.mycila</groupId>

			<artifactId>xmltool</artifactId>

			<version>3.3</version>

		</dependency>

		<dependency>

			<groupId>eu.dnetlib</groupId>

			<artifactId>cnr-misc-utils</artifactId>

			<version>[1.0.0,2.0.0)</version>

			<exclusions>

				<exclusion>

					<groupId>apache</groupId>

					<artifactId>commons-lang</artifactId>

				</exclusion>

			</exclusions>

		</dependency>

		<dependency>

			<groupId>eu.dnetlib</groupId>

			<artifactId>dnet-hadoop-commons</artifactId>

			<version>[2.0.0,3.0.0)</version>

			<exclusions>

				<exclusion>

					<artifactId>commons-httpclient</artifactId>

					<groupId>commons-httpclient</groupId>

				</exclusion>

			</exclusions>

		</dependency>

		<dependency>

			<groupId>org.apache.hbase</groupId>

			<artifactId>hbase</artifactId>

			<version>${apache.hbase.version}</version>

			<exclusions>

				<exclusion>

					<groupId>tomcat</groupId>

					<artifactId>jasper-runtime</artifactId>

				</exclusion>

				<exclusion>

					<groupId>tomcat</groupId>

					<artifactId>jasper-compiler</artifactId>

				</exclusion>

				<exclusion>

					<artifactId>slf4j-api</artifactId>

					<groupId>org.slf4j</groupId>

				</exclusion>

				<exclusion>

					<artifactId>slf4j-log4j12</artifactId>

					<groupId>org.slf4j</groupId>

				</exclusion>

				<exclusion>

					<artifactId>commons-lang</artifactId>

					<groupId>commons-lang</groupId>

				</exclusion>

				<exclusion>

					<artifactId>commons-httpclient</artifactId>

					<groupId>commons-httpclient</groupId>

				</exclusion>

				<exclusion>

					<artifactId>httpclient</artifactId>

					<groupId>org.apache.httpcomponents</groupId>

				</exclusion>

				<exclusion>

					<artifactId>httpcore</artifactId>

					<groupId>org.apache.httpcomponents</groupId>

				</exclusion>

			</exclusions>

		</dependency>

		<dependency>

			<groupId>com.googlecode.protobuf-java-format</groupId>

			<artifactId>protobuf-java-format</artifactId>

			<version>1.2</version>

		</dependency>

		<dependency>

			<groupId>eu.dnetlib</groupId>

			<artifactId>dnet-openaireplus-mapping-utils</artifactId>

			<version>[6.0.0,6.1.5]</version>

		</dependency>

		<dependency>

			<groupId>org.antlr</groupId>

			<artifactId>stringtemplate</artifactId>

			<version>3.2</version>

		</dependency>

		<dependency>

			<groupId>org.json</groupId>

			<artifactId>json</artifactId>

			<version>20140107</version>

		</dependency>

		<dependency>

			<groupId>com.typesafe</groupId>

			<artifactId>config</artifactId>

			<version>1.2.1</version>

		</dependency>

		<dependency>

			<groupId>eu.dnetlib</groupId>

			<artifactId>dnet-pace-core</artifactId>

			<version>[2.0.0,3.0.0)</version>

		</dependency>

		<dependency>

			<groupId>org.mongodb</groupId>

			<artifactId>mongo-java-driver</artifactId>

			<version>${mongodb.driver.version}</version>

		</dependency>

		<dependency>

			<groupId>eu.dnetlib</groupId>

			<artifactId>dnet-actionmanager-common</artifactId>

			<version>[6.0.0,7.0.0)</version>

			<exclusions>

				<exclusion>

					<groupId>commons-httpclient</groupId>

					<artifactId>commons-httpclient</artifactId>

				</exclusion>

			</exclusions>

		</dependency>

		<dependency>

			<groupId>org.elasticsearch</groupId>

			<artifactId>elasticsearch-hadoop-mr</artifactId>

			<version>5.2.0</version>

			<exclusions>

				<exclusion>

					<groupId>tomcat</groupId>

					<artifactId>jasper-compiler</artifactId>

				</exclusion>

				<exclusion>

					<groupId>org.antlr</groupId>

					<artifactId>antlr-runtime</artifactId>

				</exclusion>

				<exclusion>

					<groupId>org.eclipse.jetty.aggregate</groupId>

					<artifactId>jetty-all</artifactId>

				</exclusion>

				<exclusion>

					<groupId>org.slf4j</groupId>

					<artifactId>slf4j-log4j12</artifactId>

				</exclusion>

				<exclusion>

					<groupId>org.glassfish.jersey.core</groupId>

					<artifactId>jersey-client</artifactId>

				</exclusion>

				<exclusion>

					<groupId>org.glassfish.jersey.core</groupId>

					<artifactId>jersey-common</artifactId>

				</exclusion>

				<exclusion>

					<groupId>org.glassfish.jersey.core</groupId>

					<artifactId>jersey-server</artifactId>

				</exclusion>

				<exclusion>

					<groupId>org.glassfish.jersey.containers</groupId>

					<artifactId>jersey-container-servlet</artifactId>

				</exclusion>

				<exclusion>

					<groupId>org.glassfish.jersey.containers</groupId>

					<artifactId>jersey-container-servlet-core</artifactId>

				</exclusion>

				<exclusion>

					<groupId>org.codehaus.groovy</groupId>

					<artifactId>groovy-all</artifactId>

				</exclusion>

				<exclusion>

					<groupId>org.apache.hive</groupId>

					<artifactId>hive-service</artifactId>

				</exclusion>

				<exclusion>

					<groupId>org.apache.spark</groupId>

					<artifactId>spark-core_2.10</artifactId>

				</exclusion>

				<exclusion>

					<groupId>org.apache.spark</groupId>

					<artifactId>spark-sql_2.10</artifactId>

				</exclusion>

				<exclusion>

					<groupId>org.apache.spark</groupId>

					<artifactId>spark-streaming_2.10</artifactId>

				</exclusion>

				<exclusion>

					<groupId>cascading</groupId>

					<artifactId>cascading-hadoop</artifactId>

				</exclusion>

				<exclusion>

					<groupId>cascading</groupId>

					<artifactId>cascading-local</artifactId>

				</exclusion>

				<exclusion>

					<groupId>org.apache.storm</groupId>

					<artifactId>storm-core</artifactId>

				</exclusion>

				<exclusion>

					<groupId>org.apache.pig</groupId>

					<artifactId>pig</artifactId>

				</exclusion>

			</exclusions>

		</dependency>

		<dependency>

			<groupId>org.mockito</groupId>

			<artifactId>mockito-core</artifactId>

			<version>${mockito.version}</version>

			<scope>test</scope>

		</dependency>

		<dependency>

			<groupId>eu.dnetlib</groupId>

			<artifactId>dnet-openaireplus-profiles</artifactId>

			<version>[1.0.0,2.0.0)</version>

			<scope>test</scope>

		</dependency>

		<dependency>

			<groupId>com.google.guava</groupId>

			<artifactId>guava</artifactId>

			<version>${google.guava.version}</version>

		</dependency>

	</dependencies>

</project>

package eu.dnetlib.data.mapreduce.hbase.broker.add;

import static eu.dnetlib.data.mapreduce.hbase.broker.mapping.EventFactory.asEvent;

import static eu.dnetlib.data.mapreduce.util.OafHbaseUtils.getPropertyValues;

import static eu.dnetlib.data.mapreduce.util.OafHbaseUtils.listKeys;

import java.io.IOException;

import java.util.HashSet;

import java.util.Map;

import java.util.Set;

import org.apache.commons.collections.MapUtils;

import org.apache.commons.lang.StringUtils;

import org.apache.commons.lang.math.RandomUtils;

import org.apache.hadoop.hbase.client.HTable;

import org.apache.hadoop.hbase.client.Result;

import org.apache.hadoop.hbase.client.ResultScanner;

import org.apache.hadoop.hbase.client.Scan;

import org.apache.hadoop.hbase.filter.FilterList;

import org.apache.hadoop.hbase.filter.FilterList.Operator;

import org.apache.hadoop.hbase.filter.PrefixFilter;

import org.apache.hadoop.hbase.io.ImmutableBytesWritable;

import org.apache.hadoop.hbase.mapreduce.TableMapper;

import org.apache.hadoop.hbase.util.Bytes;

import org.apache.hadoop.io.Text;

import com.google.common.base.Function;

import com.google.common.collect.Iterables;

import com.google.common.collect.Maps;

import com.google.common.collect.Sets;

import eu.dnetlib.broker.objects.OpenAireEventPayload;

import eu.dnetlib.data.mapreduce.hbase.broker.Topic;

import eu.dnetlib.data.mapreduce.hbase.broker.mapping.OpenAireEventPayloadFactory;

import eu.dnetlib.data.mapreduce.hbase.broker.model.EventMessage;

import eu.dnetlib.data.mapreduce.util.OafDecoder;

import eu.dnetlib.data.proto.OafProtos.Oaf;

import eu.dnetlib.data.proto.OafProtos.OafEntity;

/**

 * Created by claudio on 08/07/16.

 */

public class AdditionMapper extends TableMapper<Text, Text> {

	private Text outKey;

	private Text outValue;

	/**

	 * Map ProjectID -> Set of Organization info (id, name)

	 */

	private Map<String, Set<EntityInfo>> projectOrganization;

	/**

	 * Map OrganizationID -> Set of Datasource info (id, name)

	 */

	private Map<String, Set<EntityInfo>> organizationDatasource;

	private Set<String> organizationPrefixBlacklist = Sets.newHashSet();

	// White list for datasource typologies.

	private Set<String> dsTypeWhitelist = Sets.newHashSet();

	@Override

	protected void setup(final Context context) throws IOException {

		organizationPrefixBlacklist = Sets.newHashSet("nsf_________");

		dsTypeWhitelist.addAll(getPropertyValues(context, "broker.datasource.type.whitelist"));

		projectOrganization = getRelMap(context, "20", "organization", "projectOrganization_participation_isParticipant", organizationPrefixBlacklist);

		organizationDatasource = getRelMap(context, "10", "datasource", "datasourceOrganization_provision_provides", dsTypeWhitelist);

		outKey = new Text("");

		outValue = new Text();

	}

	class EntityInfo {

		private String id;

		private String name;

		public EntityInfo(final String id, final String name) {

			this.id = id;

			this.name = name;

		}

		@Override

		public int hashCode() {

			return getId().hashCode();

		}

		@Override

		public boolean equals(final Object obj) {

			return getId().equals(obj);

		}

		public String getId() {

			return id;

		}

		public void setId(final String id) {

			this.id = id;

		}

		public String getName() {

			return name;

		}

		public void setName(final String name) {

			this.name = name;

		}

	}

	@Override

	protected void map(final ImmutableBytesWritable key, final Result value, final Context context) throws IOException, InterruptedException {

		try {

			final Map<byte[], byte[]> map = value.getFamilyMap(Bytes.toBytes("result"));

			final byte[] bodyB = map.get(Bytes.toBytes("body"));

			if (MapUtils.isEmpty(map) || (bodyB == null)) {

				context.getCounter("result", "empty body").increment(1);

				return;

			}

			final Oaf oaf = Oaf.parseFrom(bodyB);

			if (oaf.getDataInfo().getDeletedbyinference()) {

				context.getCounter("result", "deletedbyinference = true").increment(1);

				return;

			}

			final Set<String> currentDatasourceIds = Sets.newHashSet(listKeys(oaf.getEntity().getCollectedfromList()));

			final Map<byte[], byte[]> resultProject = value.getFamilyMap(Bytes.toBytes("resultProject_outcome_isProducedBy"));

			if (!MapUtils.isEmpty(resultProject)) {

				for (final String projectId : asStringID(resultProject.keySet())) {

					final Set<EntityInfo> organizations = projectOrganization.get(projectId);

					if ((organizations != null) && !organizations.isEmpty()) {

						for (final EntityInfo organization : organizations) {

							final Set<EntityInfo> datasources = organizationDatasource.get(organization.getId());

							if ((datasources != null) && !datasources.isEmpty()) {

								for (final EntityInfo datasource : datasources) {

									if (!currentDatasourceIds.contains(datasource.getId())) {

										// emit event for datasourceId

										final float trust = RandomUtils.nextFloat();

										final OpenAireEventPayload payload = OpenAireEventPayloadFactory.fromOAF(oaf.getEntity(), oaf.getEntity(), trust);

										// event.setPayload(HighlightFactory.highlightEnrichPid(payload,

										// Lists.newArrayList(pids)).toJSON());

										final EventMessage event =

												asEvent(oaf.getEntity(), Topic.ADD_BY_PROJECT, payload, datasource.getId(), datasource.getName(), trust);

										emit(event, context);

										context.getCounter("event", Topic.ADD_BY_PROJECT.getValue()).increment(1);

									}

								}

							}

						}

					}

				}

			}

		} catch (final Exception e) {

			throw new RuntimeException(e);

		}

	}

	private void emit(final EventMessage e, final Context context) throws IOException, InterruptedException {

		// tKey.set(e.getMap().get("id"));

		outValue.set(e.toString());

		context.write(outKey, outValue);

	}

	private Iterable<String> asStringID(final Iterable<byte[]> in) {

		return Iterables.transform(in, new Function<byte[], String>() {

			@Override

			public String apply(final byte[] input) {

				return getID(new String(input));

			}

		});

	}

	private Map<String, Set<EntityInfo>> getRelMap(final Context context,

			final String prefixFilter,

			final String entity,

			final String columnFamily,

			final Set<String> filter) throws IOException {

		System.out.println(String.format("loading %s, %s", entity, columnFamily));

		final Map<String, Set<EntityInfo>> out = Maps.newHashMap();

		final String tableName = context.getConfiguration().get("hbase.mapred.inputtable");

		System.out.println(String.format("table name: '%s'", tableName));

		try (final HTable table = new HTable(context.getConfiguration(), tableName);

				final ResultScanner res = scanTable(table, prefixFilter, entity, columnFamily)) {

			for (final Result r : res) {

				final byte[] bodyB = r.getValue(Bytes.toBytes(entity), Bytes.toBytes("body"));

				if (bodyB == null) {

					context.getCounter("missing body", entity).increment(1);

				} else {

					final OafEntity oafEntity = OafDecoder.decode(bodyB).getEntity();

					final EntityInfo kv = getEntityInfo(oafEntity, filter);

					if (kv != null) {

						final Map<byte[], byte[]> relMap = r.getFamilyMap(Bytes.toBytes(columnFamily));

						if (MapUtils.isNotEmpty(relMap)) {

							for (final String id : asStringID(relMap.keySet())) {

								if (!out.containsKey(id)) {

									out.put(id, new HashSet<EntityInfo>());

								}

								out.get(id).add(kv);

							}

						} else {

							context.getCounter("skipped", entity).increment(1);

						}

					}

				}

			}

		}

		System.out.println(String.format("loaded map for %s, %s, size: %s", entity, columnFamily, out.size()));

		return out;

	}

	private EntityInfo getEntityInfo(final OafEntity entity, final Set<String> filter) {

		final String id = getID(entity.getId());

		switch (entity.getType()) {

		case datasource:

			final String dsType = entity.getDatasource().getMetadata().getDatasourcetype().getClassid();

			if (!filter.contains(dsType)) { return null; }

			return new EntityInfo(id, entity.getDatasource().getMetadata().getOfficialname().getValue());

		case organization:

			if (filter.contains(prefix(id))) { return null; }

			return new EntityInfo(id, entity.getOrganization().getMetadata().getLegalname().getValue());

		default:

			throw new IllegalArgumentException("invalid entity: " + entity);

		}

	}

	private ResultScanner scanTable(final HTable table, final String prefixFilter, final String entity, final String columnFamily) throws IOException {

		final Scan scan = new Scan();

		final FilterList fl = new FilterList(Operator.MUST_PASS_ALL);

		fl.addFilter(new PrefixFilter(Bytes.toBytes(prefixFilter)));

		scan.setFilter(fl);

		scan.addFamily(Bytes.toBytes(entity));

		scan.addFamily(Bytes.toBytes(columnFamily));

		return table.getScanner(scan);

	}

	private String getID(final String s) {

		return StringUtils.substringAfter(s, "|");

	}

	private String prefix(final String s) {

		return StringUtils.substringBefore(s, "::");

	}

}

package eu.dnetlib.data.mapreduce.hbase.dedup;

import java.io.IOException;

import java.util.List;

import com.google.common.collect.Iterables;

import com.google.common.collect.Lists;

import eu.dnetlib.data.mapreduce.JobParams;

import eu.dnetlib.data.mapreduce.hbase.index.config.RelClasses;

import eu.dnetlib.data.mapreduce.util.DedupUtils;

import eu.dnetlib.data.mapreduce.util.OafDecoder;

import eu.dnetlib.data.mapreduce.util.OafHbaseUtils;

import eu.dnetlib.data.mapreduce.util.OafRelDecoder;

import eu.dnetlib.data.proto.OafProtos.Oaf;

import eu.dnetlib.data.transform.OafEntityMerger;

import eu.dnetlib.pace.config.DedupConfig;

import org.apache.commons.lang.StringUtils;

import org.apache.hadoop.hbase.client.Put;

import org.apache.hadoop.hbase.io.ImmutableBytesWritable;

import org.apache.hadoop.hbase.mapreduce.TableReducer;

import org.apache.hadoop.hbase.util.Bytes;

import org.apache.hadoop.io.Text;

public class DedupBuildRootsReducer extends TableReducer<Text, ImmutableBytesWritable, ImmutableBytesWritable> {

	private enum OafPatch {

		rootToEntity, entityToRoot

	}

	private DedupConfig dedupConf;

	private RelClasses relClasses;

	@Override

	protected void setup(final Context context) throws IOException, InterruptedException {

		super.setup(context);

		dedupConf = DedupConfig.load(context.getConfiguration().get(JobParams.DEDUP_CONF));

		System.out.println("dedup buildRoots reducer\n\nwf conf: " + dedupConf.toString());

		final String relClassJson = context.getConfiguration().get("relClasses");

		System.out.println("relClassesJson:\n" + relClassJson);

		relClasses = RelClasses.fromJSon(relClassJson);

		System.out.println("relClasses:\n" + relClasses);

	}

	@Override

	protected void reduce(final Text key, final Iterable<ImmutableBytesWritable> values, final Context context) throws IOException, InterruptedException {

		// ensures we're dealing with a root, otherwise returns

		if (!DedupUtils.isRoot(key.toString())) {

			System.err.println("aborting DedupBuildRootsReducer, found non-root key: " + key);

			context.getCounter("DedupBuildRootsReducer", "aborted").increment(1);

			return;

		}

		final byte[] rowkey = Bytes.toBytes(key.toString());

		final List<Oaf> entities = Lists.newArrayList();

		for (final Oaf oaf : toOaf(values)) {

			switch (oaf.getKind()) {

			case entity:

				entities.add(oaf);

				break;

			case relation:

				handleRels(context, rowkey, oaf);

				break;

			default:

				break;

			}

		}

		// build and emit the root body

		final Oaf.Builder builder = OafEntityMerger.merge(dedupConf, key.toString(), entities);

		if (entities.size() < JobParams.MAX_COUNTERS) {

			context.getCounter(dedupConf.getWf().getEntityType() + " root group size", lpad(entities.size())).increment(1);

		} else {

			context.getCounter(dedupConf.getWf().getEntityType() + " root group size", "> " + JobParams.MAX_COUNTERS).increment(1);

		}

		emit(context, rowkey, dedupConf.getWf().getEntityType(), DedupUtils.BODY_S, builder.build().toByteArray(), "root");

	}

	private Iterable<Oaf> toOaf(final Iterable<ImmutableBytesWritable> values) {

		return Iterables.transform(values, OafHbaseUtils.oafDecoder());

	}

	private void handleRels(final Context context, final byte[] rowkey, final Oaf oaf) throws IOException, InterruptedException {

		// emit relation from the root to the related entities

		OafDecoder decoder = rootToEntity(rowkey, oaf);

		emit(context, rowkey, decoder.getCFQ(), decoder.relTargetId(), decoder.toByteArray(), "[root -> entity]");

		// emit relation from the related entities to the root

		decoder = entityToRoot(rowkey, oaf);

		byte[] revKey = Bytes.toBytes(decoder.relSourceId());

		emit(context, revKey, decoder.getCFQ(), new String(rowkey), decoder.toByteArray(), "[entity -> root]");

		// mark relation from the related entities to the duplicate as deleted

		decoder = markDeleted(oaf, true);

		revKey = Bytes.toBytes(decoder.relSourceId());

		emit(context, revKey, decoder.getCFQ(), decoder.relTargetId(), decoder.toByteArray(), "mark deleted [dup -> entity]");

		// mark relation from the related entities to the duplicate as deleted

		decoder = markDeleted(oaf, false);

		revKey = Bytes.toBytes(decoder.relSourceId());

		emit(context, revKey, decoder.getCFQ(), decoder.relTargetId(), decoder.toByteArray(), "mark deleted [entity -> dup]");

	}

	private void emit(final Context context, final byte[] rowkey, final String family, final String qualifier, final byte[] value, final String label)

			throws IOException, InterruptedException {

		final Put put = new Put(rowkey).add(Bytes.toBytes(family), Bytes.toBytes(qualifier), value);

		put.setWriteToWAL(JobParams.WRITE_TO_WAL);

		context.write(new ImmutableBytesWritable(rowkey), put);

		context.getCounter(family, label).increment(1);

	}

	// /////////////////

	private OafDecoder rootToEntity(final byte[] rootRowkey, final Oaf rel) {

		return patchRelations(rootRowkey, rel, OafPatch.rootToEntity);

	}

	private OafDecoder entityToRoot(final byte[] rootRowkey, final Oaf rel) {

		return patchRelations(rootRowkey, rel, OafPatch.entityToRoot);

	}

	private OafDecoder markDeleted(final Oaf rel, final boolean reverse) {

		return deleteRelations(rel, reverse);

	}

	// patches relation objects setting the source field with the root id

	private OafDecoder patchRelations(final byte[] rootRowkey, final Oaf rel, final OafPatch patchKind) {

		final String id = new String(rootRowkey);

		final OafRelDecoder decoder = OafRelDecoder.decode(rel.getRel());

		final Oaf.Builder builder = Oaf.newBuilder(rel);

		builder.getDataInfoBuilder().setInferred(true).setDeletedbyinference(false);

		switch (patchKind) {

		case rootToEntity:

			// builder.getDataInfoBuilder().setInferenceprovenance("dedup (BuildRoots p:rootToEntity)");

			builder.getRelBuilder().setSource(id);

			break;

		case entityToRoot:

			final String relClass = rel.getRel().getRelClass();

			/*

			if(StringUtils.isBlank(relClass)) {

				throw new IllegalStateException(String.format("missing relation term for %s in row %s", rel.getRel().getRelType().name(), id));

			}

			*/

			final String inverse = relClasses.getInverse(relClass);

			if(StringUtils.isBlank(inverse)) {

				throw new IllegalStateException(String.format("missing inverse relation for %s in row %s", relClass, id));

			}

			builder.setRel(decoder.setClassId(inverse));

			// builder.getDataInfoBuilder().setInferenceprovenance("dedup (BuildRoots p:entityToRoot)");

			builder.getRelBuilder().setSource(builder.getRel().getTarget());

			builder.getRelBuilder().setTarget(id);

			break;

		default:

			break;

		}

		return OafDecoder.decode(builder.build());

	}

	private OafDecoder deleteRelations(final Oaf rel, final boolean reverse) {

		final Oaf.Builder builder = Oaf.newBuilder(rel);

		// builder.getDataInfoBuilder().setInferenceprovenance("dedup (BuildRoots d: " + reverse + ")");

		builder.getDataInfoBuilder().setDeletedbyinference(true);

		if (reverse) {

			final OafRelDecoder decoder = OafRelDecoder.decode(rel.getRel());

			builder.setRel(decoder.setClassId(relClasses.getInverse(rel.getRel().getRelClass())));

			// swap source and target

			final String tmp = builder.getRel().getSource();

			builder.getRelBuilder().setSource(builder.getRel().getTarget());

			builder.getRelBuilder().setTarget(tmp);

		}

		return OafDecoder.decode(builder.build());

	}

	private String lpad(final int s) {

		return StringUtils.leftPad(String.valueOf(s), 5);

	}

}

package eu.dnetlib.data.mapreduce.hbase.dedup.cc;

/**

 * Created by claudio on 14/10/15.

 */

import java.io.IOException;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Reducer;

public class MindistSearchReducer extends Reducer<Text, VertexWritable, Text, VertexWritable> {

	private static final Log log = LogFactory.getLog(MindistSearchReducer.class);

	public static final String UPDATE_COUNTER = "UpdateCounter";

	public static final String SKIPPED = "SKIPPED";

	public static final String UPDATED = "UPDATED";

	private boolean depthOne;

	private boolean debug = false;

	@Override

	protected void setup(Context context) throws IOException, InterruptedException {

		super.setup(context);

		final String recursionDepth = context.getConfiguration().get("mindist_recursion_depth");

		log.info("got recursion depth: " + recursionDepth);

		if (Integer.parseInt(recursionDepth) == 0) {

			depthOne = true;

		}

		debug = context.getConfiguration().getBoolean("mindist_DEBUG", false);

		log.info("debug mode: " + debug);

	}

	@Override

	protected void reduce(Text key, Iterable<VertexWritable> values, Context context) throws IOException, InterruptedException {

		VertexWritable realVertex = null;

		Text currentMinimalKey = null;

		//boolean foundEdges = false;

		if (depthOne) {

			for (VertexWritable vertex : values) {

				if (!vertex.isMessage()) {

					//log.info(String.format("found vertex with edges: %s", key.toString()));

					realVertex = vertex.clone();

				}

			}

			if (realVertex == null) {

				context.getCounter(UPDATE_COUNTER, SKIPPED).increment(1);

				return;

			}

			realVertex.setActivated(true);

			realVertex.setVertexId(realVertex.getEdges().first());

			if (key.compareTo(realVertex.getVertexId()) < 0) {

				realVertex.setVertexId(key);

			}

			context.getCounter(UPDATE_COUNTER, UPDATED).increment(1);

		} else {

			for (VertexWritable vertex : values) {

				if (!vertex.isMessage()) {

					if (realVertex == null) {

						realVertex = vertex.clone();

					}

				} else {

					if (currentMinimalKey == null) {

						currentMinimalKey = new Text(vertex.getVertexId());

					} else {

						if (currentMinimalKey.compareTo(vertex.getVertexId()) > 0) {

							currentMinimalKey = new Text(vertex.getVertexId());

						}

					}

				}

			}

			if (realVertex == null) {

				context.getCounter(UPDATE_COUNTER, SKIPPED).increment(1);

				return;

			}

			if (currentMinimalKey != null && currentMinimalKey.compareTo(realVertex.getVertexId()) < 0) {

				realVertex.setVertexId(currentMinimalKey);

				realVertex.setActivated(true);

				context.getCounter(UPDATE_COUNTER, UPDATED).increment(1);

			} else {

				realVertex.setActivated(false);

			}

		}

		context.write(key, realVertex);

		if (debug) {

			log.info(realVertex.toJSON());

		}

	}

}

package eu.dnetlib.data.mapreduce.hbase.dedup.cc;

import java.io.IOException;

import org.apache.hadoop.hbase.KeyValue;

import org.apache.hadoop.hbase.client.Result;

import org.apache.hadoop.hbase.io.ImmutableBytesWritable;

import org.apache.hadoop.hbase.mapreduce.TableMapper;

import org.apache.hadoop.io.Text;

/**

 * Created by claudio on 14/10/15.

 */

public class HBaseToSimilarityGraphMapper extends TableMapper<Text, VertexWritable> {

	@Override

	protected void map(final ImmutableBytesWritable keyIn, final Result value, final Context context) throws IOException, InterruptedException {

		final VertexWritable vertex = new VertexWritable();

		final Text realKey = new Text(keyIn.copyBytes());

		vertex.checkAndSetMinimalVertex(realKey);

		vertex.addVertex(realKey);

		for (KeyValue kv : value.list()) {

			Text tmp = new Text(kv.getQualifier());

			vertex.checkAndSetMinimalVertex(tmp);

			vertex.addVertex(tmp);

		}

		context.write(realKey, vertex);

		for (Text edge : vertex.getEdges()) {

			context.write(edge, vertex.makeMessage());

		}

	}

}

package eu.dnetlib.data.mapreduce.util;

import java.io.StringReader;

import java.io.StringWriter;

import java.util.List;

import java.util.Map;

import java.util.Map.Entry;

import java.util.Set;

import javax.xml.transform.*;

import javax.xml.transform.dom.DOMSource;

import javax.xml.transform.stream.StreamResult;

import com.google.common.base.Joiner;

import com.google.common.base.Predicate;

import com.google.common.base.Splitter;

import com.google.common.collect.Iterables;

import com.google.common.collect.Lists;

import com.google.common.collect.Maps;

import com.google.common.collect.Sets;

import com.google.protobuf.Descriptors.EnumValueDescriptor;

import com.google.protobuf.Descriptors.FieldDescriptor;

import com.google.protobuf.GeneratedMessage;

import com.mycila.xmltool.XMLDoc;

import com.mycila.xmltool.XMLTag;

import eu.dnetlib.data.mapreduce.hbase.index.config.*;

import eu.dnetlib.data.proto.FieldTypeProtos.*;

import eu.dnetlib.data.proto.OafProtos.OafEntity;

import eu.dnetlib.data.proto.OafProtos.OafRel;

import eu.dnetlib.data.proto.ProjectProtos.Project;

import eu.dnetlib.data.proto.RelMetadataProtos.RelMetadata;

import eu.dnetlib.data.proto.ResultProtos.Result;

import eu.dnetlib.data.proto.ResultProtos.Result.Context;

import eu.dnetlib.data.proto.ResultProtos.Result.ExternalReference;

import eu.dnetlib.data.proto.ResultProtos.Result.Instance;

import eu.dnetlib.data.proto.ResultProtos.Result.Journal;

import eu.dnetlib.data.proto.TypeProtos;

import eu.dnetlib.data.proto.TypeProtos.Type;

import org.apache.commons.lang.StringUtils;

import org.dom4j.Document;

import org.dom4j.DocumentException;

import org.dom4j.Element;

import org.dom4j.Node;

import org.dom4j.io.SAXReader;

import static eu.dnetlib.miscutils.collections.MappedCollection.listMap;

public class XmlRecordFactory {

	// private static final Log log = LogFactory.getLog(XmlRecordFactory.class); // NOPMD by marko on 11/24/08 5:02 PM

	private final Map<String, Integer> relCounters = Maps.newHashMap();

	protected Set<String> specialDatasourceTypes;

	protected TemplateFactory templateFactory = new TemplateFactory();

	protected OafDecoder mainEntity = null;

	protected String key = null;

	protected List<OafDecoder> relations = Lists.newLinkedList();

	protected List<OafDecoder> children = Lists.newLinkedList();

	protected EntityConfigTable entityConfigTable;

	protected ContextMapper contextMapper;

	protected RelClasses relClasses;

	protected String schemaLocation;

	protected boolean entityDefaults;

	protected boolean relDefaults;

	protected boolean childDefaults;

	protected Set<String> contextes = Sets.newHashSet();

	protected List<String> extraInfo = Lists.newArrayList();

	protected Map<String, Integer> counters = Maps.newHashMap();

	protected Transformer transformer;

	protected static Predicate<String> instanceFilter = new Predicate<String>() {

		final Set<String> instanceFieldFilter = Sets.newHashSet("instancetype", "hostedby", "license", "accessright", "collectedfrom", "dateofacceptance");

		@Override

		public boolean apply(final String s) {

			return instanceFieldFilter.contains(s);

		}

	};

	public XmlRecordFactory(final EntityConfigTable entityConfigTable, final ContextMapper contextMapper, final RelClasses relClasses,

			final String schemaLocation, final boolean entityDefaults, final boolean relDefaults, final boolean childDefeaults, final Set<String> otherDatasourceTypesUForUI)

			throws TransformerConfigurationException, TransformerFactoryConfigurationError {

		this.entityConfigTable = entityConfigTable;

		this.contextMapper = contextMapper;

		this.relClasses = relClasses;

		this.schemaLocation = schemaLocation;

		this.entityDefaults = entityDefaults;

		this.relDefaults = relDefaults;

		this.childDefaults = childDefeaults;

		this.specialDatasourceTypes = otherDatasourceTypesUForUI;

		transformer = TransformerFactory.newInstance().newTransformer();

		transformer.setOutputProperty(OutputKeys.OMIT_XML_DECLARATION, "yes");

	}

	public static String removePrefix(final String s) {

		if (s.contains("|")) return StringUtils.substringAfter(s, "|");

		return s;

	}

	public static String escapeXml(final String value) {

		return value.replaceAll("&", "&amp;").replaceAll("<", "&lt;").replaceAll(">", "&gt;").replaceAll("\"", "&quot;").replaceAll("'", "&apos;");

	}

	public Map<String, Integer> getRelCounters() {

		return relCounters;

	}

	public RelClasses getRelClasses() {

		return relClasses;

	}

	public String getId() {

		return key;

	}

	public boolean isValid() {

		return mainEntity != null;

	}

	public void setMainEntity(final OafDecoder mainEntity) {

		this.mainEntity = mainEntity;

		this.key = mainEntity.decodeEntity().getId();

	}

	public void addRelation(final Type type, final OafDecoder rel) {

		addRelOrChild(type, relations, rel);

	}

	public void addChild(final Type type, final OafDecoder child) {

		addRelOrChild(type, children, child);

	}

	private void addRelOrChild(final Type type, final List<OafDecoder> list, final OafDecoder decoder) {

		final OafRel oafRel = decoder.getOafRel();

		final String rd = oafRel.getRelType().toString() + "_" + oafRel.getSubRelType() + "_" + relClasses.getInverse(oafRel.getRelClass());

		final LinkDescriptor ld = entityConfigTable.getDescriptor(type, new RelDescriptor(rd));

		if (getRelCounters().get(rd) == null) {

			getRelCounters().put(rd, 0);

		}

		if (ld == null) {

			list.add(decoder);

			return;

		}

		if (ld.getMax() < 0) {

			list.add(decoder);

			return;

		}

		if (getRelCounters().get(rd) < ld.getMax()) {

			getRelCounters().put(rd, getRelCounters().get(rd) + 1);

			list.add(decoder);

		}

	}

	public String build() {

		try {

			final OafEntityDecoder entity = mainEntity.decodeEntity();

			// log.info("building");

			// log.info("main: " + mainEntity);

			// log.info("rel:  " + relations);

			// log.info("chi:  " + children);

			// log.info("=============");

			final Predicate<String> filter = entityConfigTable.getFilter(entity.getType());

			final List<String> metadata = decodeType(entity, filter, entityDefaults, false);

			// rels has to be processed before the contexts because they enrich the contextMap with the funding info.

			final List<String> rels = listRelations();

			metadata.addAll(buildContexts(entity.getType()));

			metadata.add(parseDataInfo(mainEntity));

			final String body = templateFactory.buildBody(entity.getType(), metadata, rels, listChildren(), extraInfo);

			return templateFactory

					.buildRecord(key, entity.getDateOfCollection(), entity.getDateOfTransformation(), schemaLocation, body, countersAsXml());

		} catch (final Throwable e) {

			throw new RuntimeException(String.format("error building record '%s'", this.key), e);

		}

	}

	private String parseDataInfo(final OafDecoder decoder) {

		final DataInfo dataInfo = decoder.getOaf().getDataInfo();

		final StringBuilder sb = new StringBuilder();

		sb.append("<datainfo>");

		sb.append(asXmlElement("inferred", dataInfo.getInferred() + "", null, null));

		sb.append(asXmlElement("deletedbyinference", dataInfo.getDeletedbyinference() + "", null, null));

		sb.append(asXmlElement("trust", dataInfo.getTrust() + "", null, null));

		sb.append(asXmlElement("inferenceprovenance", dataInfo.getInferenceprovenance() + "", null, null));

		sb.append(asXmlElement("provenanceaction", null, dataInfo.getProvenanceaction(), null));

		sb.append("</datainfo>");

		return sb.toString();

	}

	private List<String> decodeType(final OafEntityDecoder decoder, final Predicate<String> filter, final boolean defaults, final boolean expandingRel) {

		final List<String> metadata = Lists.newArrayList();

		metadata.addAll(listFields(decoder.getMetadata(), filter, defaults, expandingRel));

		metadata.addAll(listFields(decoder.getOafEntity(), filter, defaults, expandingRel));

		if ((decoder.getEntity() instanceof Result) && !expandingRel) {

			metadata.add(asXmlElement("bestaccessright", "", getBestAccessright(), null));

			metadata.addAll(listFields(decoder.getEntity(), filter, defaults, expandingRel));

		}

		if ((decoder.getEntity() instanceof Project) && !expandingRel) {

			metadata.addAll(listFields(decoder.getEntity(), filter, defaults, expandingRel));

		}

		return metadata;

	}

	private Qualifier getBestAccessright() {

		Qualifier bestAccessRight = getQualifier("UNKNOWN", "not available", "dnet:access_modes");

		final LicenseComparator lc = new LicenseComparator();

		for (final Instance instance : ((Result) mainEntity.decodeEntity().getEntity()).getInstanceList()) {

			if (lc.compare(bestAccessRight, instance.getAccessright()) > 0) {

				bestAccessRight = instance.getAccessright();

			}

		}

		return bestAccessRight;

	}

	public Qualifier getQualifier(final String classid, final String classname, final String schemename) {

		return Qualifier.newBuilder().setClassid(classid).setClassname(classname).setSchemeid(schemename).setSchemename(schemename).build();

	}

	private List<String> listRelations() {

		final List<String> rels = Lists.newArrayList();

		for (final OafDecoder decoder : this.relations) {

			final OafRel rel = decoder.getOafRel();

			final OafEntity cachedTarget = rel.getCachedTarget();

			final OafRelDecoder relDecoder = OafRelDecoder.decode(rel);

			// if (!relDecoder.getRelType().equals(RelType.personResult) || relDecoder.getRelTargetId().equals(key)) {

			if (relDecoder.getRelSourceId().equals(key) || relDecoder.getRelTargetId().equals(key)) {

				final List<String> metadata = Lists.newArrayList();

				final TypeProtos.Type targetType = relDecoder.getTargetType(mainEntity.getEntity().getType());

				//final Set<String> relFilter = entityConfigTable.getFilter(targetType, relDecoder.getRelDescriptor());

				metadata.addAll(listFields(relDecoder.getSubRel(), entityConfigTable.getIncludeFilter(targetType, relDecoder.getRelDescriptor()), false, true));

				String semanticclass = "";

				String semanticscheme = "";

				final RelDescriptor relDescriptor = relDecoder.getRelDescriptor();

				if ((cachedTarget != null) && cachedTarget.isInitialized()) {

					//final Set<String> filter = entityConfigTable.getFilter(targetType, relDescriptor);

					final OafEntityDecoder d = OafEntityDecoder.decode(cachedTarget);

					metadata.addAll(decodeType(d, entityConfigTable.getIncludeFilter(targetType, relDescriptor), relDefaults, true));

					if (d.getType().equals(Type.result)) {

						for(Instance i : cachedTarget.getResult().getInstanceList()) {

							final List<String> fields = listFields(i, entityConfigTable.getIncludeFilter(targetType, relDecoder.getRelDescriptor()), false, true);

							metadata.addAll(fields);

						}

					}

				}

				final RelMetadata relMetadata = relDecoder.getRelMetadata();

				// debug

				if (relMetadata == null) {

					// System.err.println(this);

					semanticclass = semanticscheme = "UNKNOWN";

				} else {

					semanticclass = relClasses.getInverse(relMetadata.getSemantics().getClassname());

					semanticscheme = relMetadata.getSemantics().getSchemename();

				}

				final String rd = relDescriptor.getSubRelType().toString();

				incrementCounter(rd);

				final DataInfo info = decoder.getOaf().getDataInfo();

				if (info.getInferred()) {

					incrementCounter(rd + "_inferred");

				} else if(StringUtils.startsWith(info.getProvenanceaction().getClassid(), "sysimport:crosswalk")) {

					incrementCounter(rd + "_collected");

				} else if(StringUtils.startsWith(info.getProvenanceaction().getClassid(), "user:")) {

					incrementCounter(rd + "_claimed");

				}

				final LinkDescriptor ld = entityConfigTable.getDescriptor(relDecoder.getTargetType(mainEntity.getEntity().getType()), relDescriptor);

				final String relId = (ld != null) && !ld.isSymmetric() ? relDecoder.getRelTargetId() : relDecoder.getRelSourceId();

				rels.add(templateFactory.getRel(targetType, relId, Sets.newHashSet(metadata), semanticclass, semanticscheme, info.getInferred(), info.getTrust(),

						info.getInferenceprovenance(), info.getProvenanceaction().getClassid()));

			}

		}

		return rels;

	}

	// //////////////////////////////////

	private List<String> listChildren() {

		final List<String> children = Lists.newArrayList();

		for (final OafDecoder decoder : this.children) {

			final OafEntity cachedTarget = decoder.getOafRel().getCachedTarget();

			addChildren(children, cachedTarget, decoder.getRelDescriptor());

		}