D-Net

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

import java.io.IOException;

import java.util.Collection;

import java.util.Map;

import java.util.Map.Entry;

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.hbase.util.Bytes;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Counter;

import com.google.protobuf.InvalidProtocolBufferException;

import eu.dnetlib.data.mapreduce.JobParams;

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

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

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

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

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

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

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

import eu.dnetlib.data.proto.KindProtos.Kind;

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

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

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

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

import eu.dnetlib.data.proto.RelTypeProtos.SubRelType;

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

import eu.dnetlib.miscutils.functional.xml.IndentXmlString;

public class PrepareFeedMapper extends TableMapper<Text, ImmutableBytesWritable> {

	private EntityConfigTable entityConfigTable;

	private RelClasses relClasses;

	private Text outKey;

	private ImmutableBytesWritable ibw;

	@Override

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

		final String json = context.getConfiguration().get(JobParams.INDEX_ENTITY_LINKS);

		System.out.println(JobParams.INDEX_ENTITY_LINKS + ":\n" + json);

		entityConfigTable = IndexConfig.load(json).getConfigMap();

		final String contextMap = context.getConfiguration().get("contextmap");

		System.out.println("contextmap:\n" + IndentXmlString.apply(contextMap));

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

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

		relClasses = RelClasses.fromJSon(relClassJson);

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

		outKey = new Text();

		ibw = new ImmutableBytesWritable();

	}

	@Override

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

		final OafRowKeyDecoder keyDecoder = OafRowKeyDecoder.decode(keyIn.copyBytes());

		final Type type = keyDecoder.getType();

		final Oaf oaf = mergeUpdates(value, context, type, keyDecoder);

		if (isValid(oaf)) {

			if (!isInvisible(oaf)) {

				if (!deletedByInference(oaf) || entityConfigTable.includeDuplicates(type)) {

					emit(new String(keyIn.copyBytes()), context, oaf);

					incrementCounter(context, Kind.entity.toString(), getEntityType(oaf, type), 1);

					final Collection<LinkDescriptor> lds = entityConfigTable.getDescriptors(type);

					if (lds.isEmpty()) {

						context.getCounter(type.name(), "missing link descriptor").increment(1);

					}

					for (final LinkDescriptor ld : lds) {

						final Map<byte[], byte[]> columnMap = value.getFamilyMap(Bytes.toBytes(ld.getRelDescriptor().getIt()));

						if (hasData(columnMap)) {

							emitRelationship(oaf.getEntity(), context, columnMap, ld);

							incrementCounter(context, type.toString(), ld.getRelDescriptor().getIt(), columnMap.size());

						} // else {

						// incrementCounter(context, type.toString(), ld.getRelDescriptor().getIt() + "_empty", 1);

						// }

					}

				} else {

					incrementCounter(context, "deleted by inference", type.toString(), 1);

				}

			} else {

				incrementCounter(context, "invisible", type.toString(), 1);

			}

		} else {

			incrementCounter(context, "missing body (map)", type.toString(), 1);

		}

	}

	private Oaf mergeUpdates(final Result value, final Context context, final Type type, final OafRowKeyDecoder keyDecoder)

			throws InvalidProtocolBufferException {

		try {

			return UpdateMerger.mergeBodyUpdates(context, value.getFamilyMap(Bytes.toBytes(type.toString())));

		} catch (final InvalidProtocolBufferException e) {

			System.err.println(String.format("Unable to parse proto (Type: %s) in row: %s", type.toString(), keyDecoder.getKey()));

			throw e;

		}

	}

	private void emitRelationship(final OafEntity cachedTarget, final Context context, final Map<byte[], byte[]> columnMap, final LinkDescriptor ld)

			throws IOException, InterruptedException {

		final Oaf.Builder oafBuilder = Oaf.newBuilder().setKind(Kind.relation);

		// iterates the column map

		for (final Entry<byte[], byte[]> e : columnMap.entrySet()) {

			final Oaf oaf = decodeProto(context, e.getValue());

			if (!isValid(oaf)) {

				incrementCounter(context, "invalid oaf rel", ld.getRelDescriptor().getIt(), 1);

			} else if (!deletedByInference(oaf)) {

				final OafRel.Builder relBuilder = OafRel.newBuilder(oaf.getRel());

				if (ld.isSymmetric()) {

					final RelDescriptor rd = ld.getRelDescriptor();

					relBuilder.setCachedTarget(cachedTarget).setRelType(rd.getRelType()).setSubRelType(rd.getSubRelType());

				} else {

					final String target = relBuilder.getSource();

					relBuilder.setSource(relBuilder.getTarget());

					relBuilder.setTarget(target);

				}

				if (ld.getRelDescriptor().getIt().contains(SubRelType.dedup.toString()) && isDedupSelf(relBuilder)) {

					incrementCounter(context, "avoid to emit dedup self", ld.getRelDescriptor().getIt(), 1);

					continue;

				}

				final OafRel oafRel = relBuilder.setChild(ld.isChild()).build();

				// String rowKey = patchTargetId(target, OafRelDecoder.decode(oafRel).getRelTargetId());

				emit(oafRel.getTarget(), context, merge(oafBuilder, oaf).setRel(oafRel).build());

			} else {

				incrementCounter(context, "deleted by inference", ld.getRelDescriptor().getIt(), 1);

			}

		}

	}

	private Oaf.Builder merge(final Oaf.Builder builder, final Oaf prototype) {

		return builder.setDataInfo(prototype.getDataInfo()).setLastupdatetimestamp(prototype.getLastupdatetimestamp());

	}

	private boolean isDedupSelf(final OafRelOrBuilder rel) {

		return rel.getSource().contains(rel.getTarget());

	}

	private boolean hasData(final Map<byte[], byte[]> columnMap) {

		return (columnMap != null) && !columnMap.isEmpty();

	}

	private boolean isValid(final Oaf oaf) {

		return (oaf != null) && oaf.isInitialized();

	}

	private boolean isInvisible(final Oaf oaf) {

		return oaf.getDataInfo().hasInvisible() ? oaf.getDataInfo().getInvisible() : false;

	}

	private boolean deletedByInference(final Oaf oaf) {

		return oaf.getDataInfo().getDeletedbyinference();

	}

	private Oaf decodeProto(final Context context, final byte[] body) {

		try {

			return Oaf.parseFrom(body);

		} catch (final InvalidProtocolBufferException e) {

			e.printStackTrace(System.err);

			context.getCounter("decodeProto", e.getClass().getName()).increment(1);

		}

		return null;

	}

	// private byte[] prefix(final Type name, final byte[] bytes) {

	// return concat(concat(name.toString().getBytes(), PrepareFeedJob.bSEPARATOR), bytes);

	// }

	private void emit(final String key, final Context context, final Oaf oaf) throws IOException, InterruptedException {

		// Text keyOut = new Text(Hashing.murmur3_128().hashString(key).toString());

		outKey.set(key);

		ibw.set(oaf.toByteArray());

		context.write(outKey, ibw);

	}

	// protected byte[] concat(final byte[] a, final byte[] b) {

	// byte[] c = new byte[a.length + b.length];

	// System.arraycopy(a, 0, c, 0, a.length);

	// System.arraycopy(b, 0, c, a.length, b.length);

	//

	// return c;

	// }

	private void incrementCounter(final Context context, final String k, final String t, final int n) {

		getCounter(context, k, t).increment(n);

	}

	private Counter getCounter(final Context context, final String k, final String t) {

		return context.getCounter(k, t);

	}

	private String getEntityType(final Oaf oaf, final Type type) {

		switch (type) {

		case result:

			return oaf.getEntity().getResult().getMetadata().getResulttype().getClassid();

		default:

			return type.toString();

		}

	}

}