D-Net

package eu.dnetlib.data.mapreduce.util.dao;

import java.nio.ByteBuffer;

import java.util.*;

import java.util.Map.Entry;

import java.util.function.Function;

import java.util.function.Predicate;

import java.util.stream.Collectors;

import java.util.stream.Stream;

import com.google.protobuf.InvalidProtocolBufferException;

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

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

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

import eu.dnetlib.data.graph.utils.RelDescriptor;

import eu.dnetlib.data.mapreduce.JobParams;

import eu.dnetlib.data.proto.*;

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

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

import eu.dnetlib.data.proto.DNGFProtos.DNGFRel.Builder;

import eu.dnetlib.data.proto.DliFieldTypeProtos;

import eu.dnetlib.data.proto.DliProtos;

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

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

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

import eu.dnetlib.data.proto.WDSFieldTypeProtos;

import eu.dnetlib.data.proto.WdsDatasetProtos;

import eu.dnetlib.data.proto.WdsPublicationProtos;

import eu.dnetlib.data.transform.Ontologies;

import org.apache.commons.lang3.StringUtils;

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

import static eu.dnetlib.data.graph.utils.RelDescriptor.ONTOLOGY_SEPARATOR;

import static eu.dnetlib.data.graph.utils.RelDescriptor.QUALIFIER_SEPARATOR;

/**

 * Created by claudio on 16/01/2017.

 */

public class HBaseTableDAO {

	public static final String ROOT = "dedup_wf";

	public static byte[] cfMetadataByte() {

		return Bytes.toBytes(cfMetadata());

	}

	public static String cfMetadata() {

		return ColumnFamily.metadata.toString();

	}

	public static byte[] cfRelsByte() {

		return Bytes.toBytes(cfRels());

	}

	public static String cfRels() {

		return ColumnFamily.rels.toString();

	}

	public static RelDescriptor parseQualifier(final byte[] qualifier) {

		return parseQualifier(new String(qualifier));

	}

	private static RelDescriptor parseQualifier(final String qualifier) {

		return new RelDescriptor(qualifier);

	}

	public static String newId(final String id, final String dedupRun) {

		if ((dedupRun == null) || (dedupRun.length() != 3)) throw new IllegalArgumentException("wrong dedupRun param");

		return id.replaceFirst("\\|.*\\:\\:", dedupPrefix(dedupRun));

	}

	public static byte[] newIdBytes(final String s, final String dedupRun) {

		return newId(s, dedupRun).getBytes();

	}

	public static byte[] newIdBytes(final ByteBuffer b, final String dedupRun) {

		return newId(new String(b.array()), dedupRun).getBytes();

	}

	public static String dedupPrefix(final String dedupRun) {

		return "|" + ROOT + "_" + dedupRun + "::";

	}

	public static boolean isRoot(final String s) {

		return s.contains(ROOT);

	}

	public static DNGFRel.Builder getDedup(final String from, final String to, final String relQualifier) {

		final RelDescriptor rd = new RelDescriptor(relQualifier);

		final Builder rel =

				DNGFRel.newBuilder()

						.setRelType(Qualifier.newBuilder()

								.setClassid(rd.getTermCode()).setClassname(rd.getTermCode())

								.setSchemeid(rd.getOntologyCode()).setSchemename(rd.getOntologyCode()))

						.setSource(from).setSourceType(DNGFRowKeyDecoder.decode(from).getType())

						.setTarget(to).setTargetType(DNGFRowKeyDecoder.decode(to).getType())

						.setChild(false);

		return rel;

	}

	public static boolean isRoot(final ImmutableBytesWritable s) {

		return isRoot(s.copyBytes());

	}

	public static boolean isRoot(final byte[] s) {

		return isRoot(new String(s));

	}

	public static String getDedupQualifier_merges(final Type type, final String targetId) {

		return getDedupQualifier_merges(type) + QUALIFIER_SEPARATOR + targetId;

	}

	public static byte[] getDedupQualifier_mergesBytes(final Type type, final String targetId) {

		return Bytes.toBytes(getDedupQualifier_merges(type, targetId));

	}

	public static String getDedupQualifier_merges(final Type type) {

		return getDedupRelType(type) + ONTOLOGY_SEPARATOR + "merges";

	}

	public static String getDedupQualifier_mergedIn(final Type type) {

		return getDedupRelType(type) + ONTOLOGY_SEPARATOR + "isMergedIn";

	}

	public static String getDedupQualifier_mergedIn(final Type type, final String targetId) {

		return getDedupQualifier_mergedIn(type) + QUALIFIER_SEPARATOR + targetId;

	}

	public static String getDedupQualifier_merges(final String type) {

		return getDedupQualifier_merges(Type.valueOf(type));

	}

	public static byte[] getDedupQualifier_mergesBytes(final Type type) {

		return Bytes.toBytes(getDedupQualifier_merges(type));

	}

	public static byte[] getDedupQualifier_mergesBytes(final String type) {

		return getDedupQualifier_mergesBytes(Type.valueOf(type));

	}

	public static byte[] getDedupQualifier_mergesBytes(final String type, final String targetId) {

		return getDedupQualifier_mergesBytes(Type.valueOf(type), targetId);

	}

	public static String getDedupQualifier_mergedIn(final String type) {

		return getDedupQualifier_mergedIn(Type.valueOf(type));

	}

	public static byte[] getDedupQualifier_mergedInBytes(final Type type) {

		return Bytes.toBytes(getDedupQualifier_mergedIn(type));

	}

	public static byte[] getDedupQualifier_mergedInBytes(final Type type, final String targetId) {

		return Bytes.toBytes(getDedupQualifier_mergedIn(type, targetId));

	}

	public static byte[] getDedupQualifier_mergedInBytes(final String type) {

		return getDedupQualifier_mergedInBytes(Type.valueOf(type));

	}

	public static String getSimilarityQualifier(final Type type) {

		return getDedupRelType(type) + ONTOLOGY_SEPARATOR + "isSimilarTo";

	}

	public static String getSimilarityQualifier(final Type type, final String targetId) {

		return getSimilarityQualifier(type) + QUALIFIER_SEPARATOR + targetId;

	}

	public static String getSimilarityQualifier(final String type) {

		return getSimilarityQualifier(Type.valueOf(type));

	}

	public static byte[] getSimilarityQualifierBytes(final Type type) {

		return Bytes.toBytes(getSimilarityQualifier(type));

	}

	public static byte[] getSimilarityQualifierBytes(final Type type, final String targetId) {

		return Bytes.toBytes(getSimilarityQualifier(type, targetId));

	}

    public static String getInverseRelation(final DNGF.Builder inputRel, final Ontologies ontologies) {

        return getInverseRelation(DNGFRelDecoder.decode(inputRel.getRel()).getRelDescriptor(), ontologies);

    }

    public static String getInverseRelation(final RelDescriptor inputRel, final Ontologies ontologies) {

        try {

            final String inverseRelation = ontologies.inverseOf(inputRel);

            if (StringUtils.isBlank(inverseRelation)) {

                return "unknown";

            }

            return inverseRelation;

        } catch (Throwable e) {

            return "unknown";

        }

    }

	public static byte[] getSimilarityQualifierBytes(final String type) {

		return getSimilarityQualifierBytes(Type.valueOf(type));

	}

	public static String getRelTypeString(final Type type) {

		return getDedupRelType(type).toString();

	}

	public static List<String> getTargetIds(final Result result, final String qualifier) {

		final Map<byte[], byte[]> rels = result.getFamilyMap(cfRelsByte());

		if (rels == null) {

			return new ArrayList<>();

		}

		return getTargetIds(rels, qualifier);

	}

	public static List<String> getTargetIds(final Map<byte[], byte[]> rels, final String semantics) {

		return rels.keySet().stream()

				.map(b -> new String(b))

				.filter(s -> s.contains(semantics))

				.map(s -> StringUtils.substringAfter(s, QUALIFIER_SEPARATOR))

				.collect(Collectors.toList());

	}

	public static byte[] getMetadataB(final Result value, final Type type) {

		return value.getValue(cfMetadataByte(), Bytes.toBytes(type.toString()));

	}

	public static DNGF getMetadata(final Result value, final Type type) throws InvalidProtocolBufferException {

		final byte[] body = getMetadataB(value, type);

		return body != null ? parseProto(body) : null;

	}

	public static String getDedupRelType(final Type type) {

		switch (type) {

		case organization:

			return "organization_organization";

		case person:

			return "person_person";

		case publication:

			return "publication_publication";

		case dataset:

			return "dataset_dataset";

            case unknown:

                return "unknown_unknown";

		default:

			throw new IllegalArgumentException("Deduplication not supported for entity type: " + type);

		}

	}

	public static Put asPutByCollectedFrom(final DNGF dngf) {

		switch (dngf.getKind()) {

		case entity:

			return asPut(dngf, sumHashes(dngf.getEntity().getCollectedfromList()).longValue());

		case relation:

			return asPut(dngf, sumHashes(dngf.getRel().getCollectedfromList()).longValue());

		default:

			throw new IllegalArgumentException("invalid kind");

		}

	}

	private static Integer sumHashes(final List<KeyValue> cfs) {

		return cfs.stream()

				.map(kv -> kv.getValue())

				.map(s -> s.hashCode())

                .map(h -> Math.abs(h))

                .collect(Collectors.summingInt(Integer::intValue));

	}

	public static Put asPut(final DNGF dngf, final Long ts) {

		switch (dngf.getKind()) {

		case entity:

			final Put entity = getPut(dngf.getEntity().getId(), ts);

			return entity.add(cfMetadataByte(), Bytes.toBytes(dngf.getEntity().getType().toString()), dngf.toByteArray());

		case relation:

			final DNGFRel rel = dngf.getRel();

			final Put putRel = getPut(rel.getSource(), ts);

			final Qualifier relType = rel.getRelType();

			final String qualifier = relType.getSchemeid() + ONTOLOGY_SEPARATOR + relType.getClassid() + QUALIFIER_SEPARATOR + rel.getTarget();

			return putRel.add(cfRelsByte(), Bytes.toBytes(qualifier), dngf.toByteArray());

		default:

			throw new IllegalArgumentException("invalid kind");

		}

	}

	private static Put getPut(final String rowkey, final Long ts) {

		final Put put = ts != null ? new Put(Bytes.toBytes(rowkey), ts) : new Put(Bytes.toBytes(rowkey));

		put.setWriteToWAL(JobParams.WRITE_TO_WAL);

		return put;

	}

	public static Put asPut(final DNGF dngf) {

		return asPut(dngf, null);

	}

	public static byte[] rowKey(final DNGF dngf) {

		switch (dngf.getKind()) {

		case entity:

			return dngf.getEntity().getId().getBytes();

		case relation:

			final DNGFRel rel = dngf.getRel();

			return rel.getSource().getBytes();

		default:

			throw new IllegalArgumentException("invalid kind");

		}

	}

	public static Map<String, Map<Long, DNGF>> relVersions(final Result value, final String... filter) {

		final NavigableMap<byte[], NavigableMap<Long, byte[]>> map = value.getMap().get(cfRelsByte());

		return filterMapKeys(map, filter)

				.collect(Collectors.toMap(

						key -> key,

						key -> map.get(key.getBytes()).entrySet()

								.stream()

								.collect(Collectors.toMap(

										Entry::getKey,

										inner -> parseProto(inner.getValue()))

								)));

	}

	public static Map<String, DNGF> rel(final Result value, final String... filter) {

		final NavigableMap<byte[], byte[]> map = value.getFamilyMap(cfRelsByte());

		return filterMapKeys(map, filter)

				.collect(Collectors.toMap(

						key -> key,

						key -> parseProto(map.get(key.getBytes()))));

	}

	private static Stream<String> filterMapKeys(final Map<byte[], ?> map, final String... filter) {

		return map.keySet().stream()

				.map(Bytes::toString)

				.filter(asFilter(filter));

	}

	private static Predicate<String> asFilter(final String... filters) {

		return qualifier -> Arrays.asList(filters).stream().allMatch(filter -> !qualifier.contains(filter));

	}

	public static DNGF parseProto(final byte[] value) {

		final DNGFDecoder d = DNGFDecoder.decode(value, DliFieldTypeProtos.completionStatus,

				DliProtos.completionStatus, DliProtos.resolvedfrom, DliProtos.typedIdentifier,

				WdsDatasetProtos.WdsDataset.geolocation, WdsDatasetProtos.WdsDataset.otherRels, WdsPublicationProtos.WdsPublication.projects,

				WdsDatasetProtos.WdsDataset.projects, WDSFieldTypeProtos.identifierType,WDSFieldTypeProtos.identifierValue, WDSFieldTypeProtos.relationSemantic

		);

		return d.getDNGF();

	}

	public static Function<DNGFDecoder, String> idDecoder() {

		return input -> input.getEntityId();

	}

	public static ImmutableBytesWritable ibw(final String targetId) {

		return new ImmutableBytesWritable(Bytes.toBytes(targetId));

	}

    public enum ColumnFamily {metadata, rels}

}