D-Net

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.graph.model.DNGFDecoder;

import eu.dnetlib.data.graph.model.DNGFRelDecoder;

import eu.dnetlib.data.mapreduce.JobParams;

import eu.dnetlib.data.proto.DNGFProtos.DNGF;

import eu.dnetlib.data.proto.DNGFProtos.DNGFEntity;

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

import eu.dnetlib.data.transform.DNGFEntityMerger;

import eu.dnetlib.data.transform.Ontologies;

import eu.dnetlib.data.transform.OntologyLoader;

import eu.dnetlib.pace.config.DedupConfig;

import org.apache.commons.lang3.StringUtils;

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;

import static eu.dnetlib.data.mapreduce.util.dao.HBaseTableDAO.*;

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

    public static final String FIXED_RELATION = "Fixed Relation";

    private DedupConfig dedupConf;

    private Ontologies ontologies;

	@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());

		ontologies = OntologyLoader.loadOntologies(context.getConfiguration().get(JobParams.ONTOLOGIES));

		System.out.println("ontologies: " + ontologies.toJson(true));

	}

	@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 (!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<DNGF> entities = Lists.newArrayList();

		for (final DNGF oaf : toDNGF(values)) {

			switch (oaf.getKind()) {

			case entity:

				entities.add(oaf);

				break;

			case relation:

				handleRels(context, rowkey, oaf, true);

				break;

			default:

				break;

			}

		}

		// build and emit the root body

		final DNGF.Builder builder = DNGFEntityMerger.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);

		}

		final DNGF dngf = builder.build();

		final DNGFEntity entity = dngf.getEntity();

		try {

			context.write(new ImmutableBytesWritable(Bytes.toBytes(entity.getId())), asPut(dngf));

		} catch (Throwable e) {

			System.out.println("Exception dngf = " + dngf.toString());

			context.getCounter(entity.getType().toString(), e.getClass().getName()).increment(1);

		}

		context.getCounter(entity.getType().toString(), "root").increment(1);

		// add person rels TODO: remove this hack

		// context.getCounter("hack", "personResult out").increment(personMap.size());

	}

	private Iterable<DNGF> toDNGF(final Iterable<ImmutableBytesWritable> values) {

		return Iterables.transform(values, ibw -> DNGFDecoder.decode(ibw.copyBytes()).getDNGF());

	}

	private void handleRels(final Context context, final byte[] rowkey, final DNGF rel, final boolean hack) throws IOException, InterruptedException {

		if (hack && checkHack(new String(rowkey), rel)) {

			context.getCounter("hack", "personResult in").increment(1);

		} else {

			// emit relation from the root to the related entities

			DNGFDecoder decoder = rootToEntity(rowkey, rel, context);

			context.write(new ImmutableBytesWritable(rowkey), asPutByCollectedFrom(decoder.getDNGF()));

			// emit relation from the related entities to the root

			decoder = entityToRoot(rowkey, rel, context);

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

			context.write(new ImmutableBytesWritable(revKey), asPutByCollectedFrom(decoder.getDNGF()));

            context.getCounter(FIXED_RELATION, decoder.getRelDescriptor().shortQualifier() + " [entity <-> root]").increment(2);

        }

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

		DNGFDecoder decoder = markDeleted(rel, true, context);

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

		context.write(new ImmutableBytesWritable(revKey), asPut(decoder.getDNGF()));

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

		decoder = markDeleted(rel, false, context);

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

		context.write(new ImmutableBytesWritable(revKey), asPut(decoder.getDNGF()));

        context.getCounter(FIXED_RELATION, decoder.getRelDescriptor().shortQualifier() + " mark deleted [dup <-> entity]").increment(2);

    }

	public boolean checkHack(final String root, final DNGF oaf) {

		boolean res;

		final String type = dedupConf.getWf().getEntityType();

		if ((type.equals(Type.publication.toString()) || type.equals(Type.dataset.toString())) &&

				oaf.getRel().getTargetType().equals(Type.person) && !md5matches(root, oaf.getRel().getSource())) {

			res = true;

		} else {

			res = false;

		}

		// if (root.equals("50|dedup_wf_001::92f6197ea6f16ae554755aced832fb6f")) {

		// System.out.println("##################");

		// System.out.println("root  : " + root);

		// System.out.println("source: " + oaf.getRel().getSource());

		// System.out.println("ckeck:  " + res);

		// }

		return res;

	}

	private boolean md5matches(final String id1, final String id2) {

		return id1.replaceAll("^.*\\:\\:", "").equals(id2.replaceAll("^.*\\:\\:", ""));

	}

	private DNGFDecoder rootToEntity(final byte[] rootRowkey, final DNGF rel, final Context context) {

		return patchRelations(rootRowkey, rel, DNGFPatch.rootToEntity, context);

	}

	private DNGFDecoder entityToRoot(final byte[] rootRowkey, final DNGF rel, final Context context) {

		return patchRelations(rootRowkey, rel, DNGFPatch.entityToRoot, context);

	}

	private DNGFDecoder markDeleted(final DNGF rel, final boolean reverse, final Context context) {

		return deleteRelations(rel, reverse, context);

	}

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

	private DNGFDecoder patchRelations(final byte[] rootRowkey, final DNGF rel, final DNGFPatch patchKind, final Context context) {

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

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

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

		switch (patchKind) {

		case rootToEntity:

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

			builder.getRelBuilder().setSource(new String(rootRowkey));

			break;

		case entityToRoot:

			builder.setRel(decoder.setClassId(getInverse(decoder, context)));

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

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

			builder.getRelBuilder().setTarget(new String(rootRowkey));

            final Type sourceType = builder.getRelBuilder().getSourceType();

            builder.getRelBuilder().setSourceType(builder.getRel().getTargetType());

            builder.getRelBuilder().setTargetType(sourceType);

            break;

		default:

			break;

		}

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

	}

	private String getInverse(final DNGFRelDecoder decoder, final Context context) {

		final String inverse = ontologies.inverseOf(decoder.getRelDescriptor());

		if (StringUtils.isBlank(inverse)) {

            //context.getCounter("unmapped relationship", decoder.getRelDescriptor().shortQualifier()).increment(1);

            return "unknown";

		}

		return inverse;

	}

	private DNGFDecoder deleteRelations(final DNGF rel, final boolean reverse, final Context context) {

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

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

		builder.getDataInfoBuilder().setDeletedbyinference(true);

		if (reverse) {

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

			builder.setRel(decoder.setClassId(getInverse(decoder, context)));

			// swap source and target

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

            final Type sType = builder.getRel().getSourceType();

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

			builder.getRelBuilder().setTarget(tmp);

            builder.getRelBuilder().setSourceType(builder.getRel().getTargetType());

            builder.getRelBuilder().setTargetType(sType);

        }

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

	}

	private String lpad(final int s) {

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

	}

	private enum DNGFPatch {

		rootToEntity, entityToRoot

	}

}