/*
* Copyright (c) 2009 Concurrent, Inc.
*
* This work has been released into the public domain
* by the copyright holder. This applies worldwide.
*
* In case this is not legally possible:
* The copyright holder grants any entity the right
* to use this work for any purpose, without any
* conditions, unless such conditions are required by law.
*/

package parallelai.spyglass.hbase;

import java.io.IOException;
import java.util.UUID;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.MasterNotRunningException;
import org.apache.hadoop.hbase.ZooKeeperConnectionException;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.mapreduce.TableOutputFormat;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.RecordReader;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import cascading.flow.FlowProcess;
import cascading.tap.SinkMode;
import cascading.tap.Tap;
import cascading.tap.hadoop.io.HadoopTupleEntrySchemeIterator;
import cascading.tuple.TupleEntryCollector;
import cascading.tuple.TupleEntryIterator;

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

/**
* The HBaseRawTap class is a {@link Tap} subclass. It is used in conjunction with
* the {@HBaseRawScheme} to allow for the reading and writing
* of data to and from a HBase cluster.
*/
@SuppressWarnings({ "deprecation", "rawtypes" })
public class HBaseRawTap extends Tap<JobConf, RecordReader, OutputCollector> {
	/**
	 *
	 */
	private static final long serialVersionUID = 8019189493428493323L;

	/** Field LOG */
	private static final Logger LOG = LoggerFactory.getLogger(HBaseRawTap.class);

	private final String id = UUID.randomUUID().toString();

	/** Field SCHEME */
	public static final String SCHEME = "hbase";

	/** Field hBaseAdmin */
	private transient HBaseAdmin hBaseAdmin;

	/** Field hostName */
	private String quorumNames;
	/** Field tableName */
	private String tableName;
	private String base64Scan;

	/**
	 * Constructor HBaseTap creates a new HBaseTap instance.
	 *
	 * @param tableName
	 *            of type String
	 * @param HBaseFullScheme
	 *            of type HBaseFullScheme
	 */
	public HBaseRawTap(String tableName, HBaseRawScheme HBaseFullScheme) {
		super(HBaseFullScheme, SinkMode.UPDATE);
		this.tableName = tableName;
	}

	/**
	 * Constructor HBaseTap creates a new HBaseTap instance.
	 *
	 * @param tableName
	 *            of type String
	 * @param HBaseFullScheme
	 *            of type HBaseFullScheme
	 * @param sinkMode
	 *            of type SinkMode
	 */
	public HBaseRawTap(String tableName, HBaseRawScheme HBaseFullScheme, SinkMode sinkMode) {
		super(HBaseFullScheme, sinkMode);
		this.tableName = tableName;
	}

	/**
	 * Constructor HBaseTap creates a new HBaseTap instance.
	 *
	 * @param tableName
	 *            of type String
	 * @param HBaseFullScheme
	 *            of type HBaseFullScheme
	 */
	public HBaseRawTap(String quorumNames, String tableName, HBaseRawScheme HBaseFullScheme) {
		super(HBaseFullScheme, SinkMode.UPDATE);
		this.quorumNames = quorumNames;
		this.tableName = tableName;
	}

	/**
	 * Constructor HBaseTap creates a new HBaseTap instance.
	 *
	 * @param tableName
	 *            of type String
	 * @param HBaseFullScheme
	 *            of type HBaseFullScheme
	 * @param sinkMode
	 *            of type SinkMode
	 */
	public HBaseRawTap(String quorumNames, String tableName, HBaseRawScheme HBaseFullScheme, SinkMode sinkMode) {
		super(HBaseFullScheme, sinkMode);
		this.quorumNames = quorumNames;
		this.tableName = tableName;
	}

	/**
	 * Constructor HBaseTap creates a new HBaseTap instance.
	 *
	 * @param quorumNames		HBase quorum
	 * @param tableName			The name of the HBase table to read
	 * @param HBaseFullScheme
	 * @param base64Scan		An optional base64 encoded scan object
	 * @param sinkMode			If REPLACE the output table will be deleted before writing to
	 */
	public HBaseRawTap(String quorumNames, String tableName, HBaseRawScheme HBaseFullScheme, String base64Scan, SinkMode sinkMode) {
		super(HBaseFullScheme, sinkMode);
		this.quorumNames = quorumNames;
		this.tableName = tableName;
		this.base64Scan = base64Scan;
	}

	/**
	 * Method getTableName returns the tableName of this HBaseTap object.
	 *
	 * @return the tableName (type String) of this HBaseTap object.
	 */
	public String getTableName() {
		return tableName;
	}

	public Path getPath() {
		return new Path(SCHEME + ":/" + tableName.replaceAll(":", "_"));
	}

	protected HBaseAdmin getHBaseAdmin(JobConf conf) throws MasterNotRunningException, ZooKeeperConnectionException {
		if (hBaseAdmin == null) {
			Configuration hbaseConf = HBaseConfiguration.create(conf);
			hBaseAdmin = new HBaseAdmin(hbaseConf);
		}

		return hBaseAdmin;
	}

	@Override
	public void sinkConfInit(FlowProcess<JobConf> process, JobConf conf) {
		if (quorumNames != null) {
			conf.set("hbase.zookeeper.quorum", quorumNames);
		}

		LOG.debug("sinking to table: {}", tableName);

		if (isReplace() && conf.get("mapred.task.partition") == null) {
			try {
				deleteResource(conf);

			} catch (IOException e) {
				throw new RuntimeException("could not delete resource: " + e);
			}
		}

		else if (isUpdate() || isReplace()) {
			try {
				createResource(conf);
			} catch (IOException e) {
				throw new RuntimeException(tableName + " does not exist !", e);
			}

		}

		conf.set(TableOutputFormat.OUTPUT_TABLE, tableName);
		super.sinkConfInit(process, conf);
	}

	@Override
	public String getIdentifier() {
		return id;
	}

	@Override
	public TupleEntryIterator openForRead(FlowProcess<JobConf> jobConfFlowProcess, RecordReader recordReader)
			throws IOException {
		return new HadoopTupleEntrySchemeIterator(jobConfFlowProcess, this, recordReader);
	}

	@Override
	public TupleEntryCollector openForWrite(FlowProcess<JobConf> jobConfFlowProcess, OutputCollector outputCollector)
			throws IOException {
		HBaseTapCollector hBaseCollector = new HBaseTapCollector(jobConfFlowProcess, this);
		hBaseCollector.prepare();
		return hBaseCollector;
	}

	@Override
	public boolean createResource(JobConf jobConf) throws IOException {
		HBaseAdmin hBaseAdmin = getHBaseAdmin(jobConf);

		if (hBaseAdmin.tableExists(tableName)) {
			return true;
		}

		LOG.info("creating hbase table: {}", tableName);

		HTableDescriptor tableDescriptor = new HTableDescriptor(tableName);

		String[] familyNames = ((HBaseRawScheme) getScheme()).getFamilyNames();

		for (String familyName : familyNames) {
			tableDescriptor.addFamily(new HColumnDescriptor(familyName));
		}

		hBaseAdmin.createTable(tableDescriptor);

		return true;
	}

	@Override
	public boolean deleteResource(JobConf jobConf) throws IOException {
		if (getHBaseAdmin(jobConf).tableExists(tableName)) {
			if (getHBaseAdmin(jobConf).isTableEnabled(tableName))
				getHBaseAdmin(jobConf).disableTable(tableName);
			getHBaseAdmin(jobConf).deleteTable(tableName);
		}
		return true;
	}

	@Override
	public boolean resourceExists(JobConf jobConf) throws IOException {
		return getHBaseAdmin(jobConf).tableExists(tableName);
	}

	@Override
	public long getModifiedTime(JobConf jobConf) throws IOException {
		return System.currentTimeMillis(); // currently unable to find last mod
											// time
											// on a table
	}

	@Override
	public void sourceConfInit(FlowProcess<JobConf> process, JobConf conf) {
		// a hack for MultiInputFormat to see that there is a child format
		FileInputFormat.setInputPaths(conf, getPath());

		if (quorumNames != null) {
			conf.set("hbase.zookeeper.quorum", quorumNames);
		}

		LOG.debug("sourcing from table: {}", tableName);
		conf.set(TableInputFormat.INPUT_TABLE, tableName);
		if (null != base64Scan)
			conf.set(TableInputFormat.SCAN, base64Scan);

		super.sourceConfInit(process, conf);
	}

	@Override
	public boolean equals(Object object) {
		if (this == object) {
			return true;
		}
		if (object == null || getClass() != object.getClass()) {
			return false;
		}
		if (!super.equals(object)) {
			return false;
		}

		HBaseRawTap hBaseTap = (HBaseRawTap) object;

		if (tableName != null ? !tableName.equals(hBaseTap.tableName) : hBaseTap.tableName != null) {
			return false;
		}

		if (base64Scan != null ? !base64Scan.equals(hBaseTap.base64Scan) : hBaseTap.base64Scan != null) {
			return false;
		}

		return true;
	}

	@Override
	public int hashCode() {
		int result = super.hashCode();
		result = 31 * result + (tableName != null ? tableName.hashCode() : 0) + (base64Scan != null ? base64Scan.hashCode() : 0);
		return result;
	}
}