/**Name		:	KcnaufHadoopManagement
 *Author	:	Zhang Bing 
 *Created	:	2013-7-31
 *Function	:	delete of hbase
 */
package com.exercise.hbase.command;

import java.util.LinkedList;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.PrefixFilter;

public class HBaseQuery {
	public Configuration config;
	private String tablename;

	public String getTablename() {
		return tablename;
	}

	public void setTablename(String tablename) {
		this.tablename = tablename;
	}

	public HBaseQuery(String tablename) {
		this.tablename = tablename;
		config = ConfigurationHandle.getInstance();
	}
	
	/**
	 * This method can get all rows data from hbase by the same rowkey 
	 * @param rowkey
	 * @return
	 * @throws Exception
	 */
	public List<Record> getDataByRowkey(String rowkey) throws Exception {
		List<Record> recordList = null;
		try {
			recordList = new LinkedList<Record>();
			HTable table = new HTable(config, tablename);
			Get get = new Get(rowkey.getBytes());
			Result rs = null;
			rs = table.get(get);
			
			KeyValue[] kv = rs.raw();
			for (int i = 0; i < kv.length; i++) {
				recordList.add(new Record(
						new String(kv[i].getRow(), "UTF-8"),
						new String(kv[i].getFamily(), "UTF-8"),
						new String(kv[i].getQualifier(), "UTF-8"),
						new String(kv[i].getValue(), "UTF-8"),
						kv[i].getTimestamp()));
			}
			/*
			 * for(KeyValue kv:rs.raw()){ record=new
			 * String(kv.getValue(),"UTF-8"); }
			 */
			return recordList;
		} catch (Exception e) {
			e.printStackTrace();
		}
		return null;
	}

	/**
	 * This method can get all rows data from hbase by the same rowkeyPrifix
	 * @param rowkeyPrifix
	 * @return
	 * @throws Exception
	 */
	public List<Record> getDataByRowkeyPrifix(String rowkeyPrifix) throws Exception {
		List<Record> recordList = new LinkedList<Record>();
		try {
			HTable table = new HTable(config, tablename);
			Scan scan = new Scan();
			scan.setFilter(new PrefixFilter(rowkeyPrifix.getBytes()));
			ResultScanner rs = table.getScanner(scan);
			
			for (Result r : rs) {
				KeyValue[] kv = r.raw();
				for (int i = 0; i < kv.length; i++) {
					recordList.add(new Record(
							new String(kv[i].getRow(), "UTF-8"),
							new String(kv[i].getFamily(), "UTF-8"),
							new String(kv[i].getQualifier(), "UTF-8"),
							new String(kv[i].getValue(), "UTF-8"),
							kv[i].getTimestamp()));
				}
			}
			rs.close();
			return recordList;
		} catch (Exception e) {
			e.printStackTrace();
			return null;
		}

	}

	/**
	 * This method can get all rows data from hbase by the same family and qualify
	 * @param family
	 * @param qualify
	 * @return
	 */
	public List<Record> getDataByFamily(String family, String qualify) {
		List<Record> recordList = new LinkedList<Record>();
		try {
			HTable table = new HTable(config, tablename);
			ResultScanner rs = table.getScanner(family.getBytes(),
					qualify.getBytes());
			for (Result r : rs) {
				KeyValue[] kv = r.raw();
				for (int i = 0; i < kv.length; i++) {
					recordList.add(new Record(
							new String(kv[i].getRow(), "UTF-8"),
							new String(kv[i].getFamily(), "UTF-8"),
							new String(kv[i].getQualifier(), "UTF-8"),
							new String(kv[i].getValue(), "UTF-8"),
							kv[i].getTimestamp()));
				}
			}
			rs.close();
			return recordList;
		} catch (Exception e) {
			e.printStackTrace();
			return null;
		}

	}

	/**
	 * This method can get data by the same timestamp
	 * @param timestamp
	 * @return
	 */
	public List<Record> getDataByTimestamp(long timestamp) {
		List<Record> recordList = new LinkedList<Record>();
		try {
			HTable table = new HTable(config, tablename);
			ResultScanner rs = table.getScanner(new Scan()
					.setTimeStamp(timestamp));
			for (Result r : rs) {
				KeyValue[] kv = r.raw();
				for (int i = 0; i < kv.length; i++) {
					recordList.add(new Record(
							new String(kv[i].getRow(), "UTF-8"),
							new String(kv[i].getFamily(), "UTF-8"),
							new String(kv[i].getQualifier(), "UTF-8"),
							new String(kv[i].getValue(), "UTF-8"),
							kv[i].getTimestamp()));
				}
			}
			rs.close();
			return recordList;
		} catch (Exception e) {
			e.printStackTrace();
			return null;
		}
	}
}