D-Net

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

import java.io.IOException;

import eu.dnetlib.data.mapreduce.JobParams;

import org.apache.hadoop.hbase.KeyValue;

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

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

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

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 com.google.common.collect.Lists;

import com.google.protobuf.GeneratedMessage;

import com.google.protobuf.InvalidProtocolBufferException;

import eu.dnetlib.data.mapreduce.hbase.VolatileColumnFamily;

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

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

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

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

public class HBaseResetMapper extends TableMapper<ImmutableBytesWritable, Mutation> {

	@Override

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

		super.setup(context);

		String table = null;

		for (final String s : Lists.newArrayList("hbase.mapreduce.inputtable", "hbase.mapred.inputtable")) {

			try {

				table = context.getConfiguration().get(s).trim();

			} catch (final NullPointerException e) {}

		}

		if (table == null) throw new IllegalStateException("unable to find table name");

		System.out.println("I start to reset table '" + table + "'");

	}

	@Override

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

		// System.out.println("EVALUATING " + Bytes.toString(key.get()));

		final byte[] bKey = key.copyBytes();

		if (DedupUtils.isRoot(new String(bKey))) {

			deleteRow(key, context);

			return;

		}

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

			if (Bytes.toString(kv.getQualifier()).equals(DedupUtils.BODY_S) || isColumn(kv.getFamily())) {

				final Oaf oaf = Oaf.parseFrom(kv.getValue());

				if (isColumnToUpdate(oaf)) {

					updateColumn(key, kv, resetOaf(oaf), context);

				}

			} else if (isColumnToDelete(kv)) {

				deleteColumn(key, kv, context);

			}

		}

	}

	private boolean isColumn(final byte[] family) {

		try {

			final String s = Bytes.toString(family);

			return !s.contains(SubRelType.dedup.toString()) && !s.contains(SubRelType.dedupSimilarity.toString()) && !VolatileColumnFamily.isVolatile(s)

					&& (DedupUtils.decodeCF(family) != null);

		} catch (final IllegalArgumentException e) {

			return false;

		}

	}

	private boolean isColumnToUpdate(final Oaf oaf) {

		return oaf.getDataInfo().hasDeletedbyinference() && oaf.getDataInfo().getDeletedbyinference();

	}

	private Oaf resetOaf(final Oaf oaf) {

		// I reset deletedbyinference

		final DataInfo.Builder dataInfoBuilder = DataInfo.newBuilder(oaf.getDataInfo());

		dataInfoBuilder.setDeletedbyinference(false).setInferenceprovenance("");

		final Oaf.Builder oafBuilder = Oaf.newBuilder(oaf);

		oafBuilder.setDataInfo(dataInfoBuilder);

		return oafBuilder.build();

	}

	private boolean isColumnToDelete(final KeyValue kv) throws InvalidProtocolBufferException {

		// Columns in "dedupRel" and "similarRel" column families and with the attribute inferred=true must be deleted

		final String cf = Bytes.toString(kv.getFamily());

		if (cf.contains(SubRelType.dedup.toString()) || cf.contains(SubRelType.dedupSimilarity.toString()) || VolatileColumnFamily.isVolatile(cf)) return true;

		final Oaf oaf = Oaf.parseFrom(kv.getValue());

		return oaf.getDataInfo().hasInferred() && oaf.getDataInfo().getInferred();

	}

	private void updateColumn(final ImmutableBytesWritable key, final KeyValue col, final GeneratedMessage newValue, final Context context) throws IOException,

	InterruptedException {

		final byte[] bKey = key.copyBytes();

		final Put put = new Put(bKey);

		put.add(col.getFamily(), col.getQualifier(), newValue.toByteArray());

		put.setWriteToWAL(JobParams.WRITE_TO_WAL);

		context.write(key, put);

		context.getCounter("reset job", "update body").increment(1);

		// System.out.println("   --- UPDATED ROW key=" + Bytes.toString(bKey));

	}

	private void deleteRow(final ImmutableBytesWritable key, final Context context) throws IOException, InterruptedException {

		final byte[] bKey = key.copyBytes();

		final Delete d = new Delete(bKey);

		d.setWriteToWAL(JobParams.WRITE_TO_WAL);

		context.write(key, d);

		context.getCounter("reset job", "delete row").increment(1);

		// System.out.println("   --- DELETED ROW key=" + Bytes.toString(bKey));

	}

	private void deleteColumn(final ImmutableBytesWritable key, final KeyValue col, final Context context) throws IOException, InterruptedException {

		final byte[] bKey = key.copyBytes();

		final Delete d = new Delete(bKey);

		d.deleteColumns(col.getFamily(), col.getQualifier());

		d.setWriteToWAL(JobParams.WRITE_TO_WAL);

		context.write(key, d);

		context.getCounter("reset job", "delete column").increment(1);

		// System.out.println("   --- DELETED COLUMN key=" + Bytes.toString(bKey) + " cf=" + Bytes.toString(col.getFamily()) + ", q="

		// + Bytes.toString(col.getQualifier()));

	}

}