package tsmc;

// TSMC serial program number 4
// 0. after TSMC3CalculateMR
// 1. run addIndexToTurnover() would add index table to tsmc.
//    if you have indexed the table before, don't do it again.
// 2. tune the filter_value in main() and run readSortedValGreater() would show report

import java.io.IOException;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.tableindexed.IndexSpecification;
import org.apache.hadoop.hbase.client.tableindexed.IndexedTable;
import org.apache.hadoop.hbase.client.tableindexed.IndexedTableAdmin;
import org.apache.hadoop.hbase.filter.CompareFilter;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.util.Bytes;

public class TSMC4SortTurnover {

  public void addIndexToTurnover(String OriTable, String IndexID,
      String OriColumn) throws IOException {
    HBaseConfiguration conf = new HBaseConfiguration();
    conf.addResource(new Path("/opt/hbase/conf/hbase-site.xml"));
    IndexedTableAdmin admin = new IndexedTableAdmin(conf);
    admin.addIndex(Bytes.toBytes(OriTable), new IndexSpecification(IndexID,
        Bytes.toBytes(OriColumn)));

  }

  public void readSortedValGreater(String filter_val) throws IOException {
    HBaseConfiguration conf = new HBaseConfiguration();
    conf.addResource(new Path("/opt/hbase/conf/hbase-site.xml"));
    // the id of the index to use
    String tablename = "tsmc";
    String indexId = "Sum";
    byte[] column_1 = Bytes.toBytes("Turnover:Sum");
    byte[] column_2 = Bytes.toBytes("Detail:Name");

    byte[] indexStartRow = HConstants.EMPTY_START_ROW;
    byte[] indexStopRow = null;
    byte[][] indexColumns = null;

    SingleColumnValueFilter indexFilter = new SingleColumnValueFilter(Bytes
        .toBytes("Turnover"), Bytes.toBytes("Sum"),
        CompareFilter.CompareOp.GREATER_OR_EQUAL, Bytes
            .toBytes(filter_val));

    byte[][] baseColumns = new byte[][] { column_1, column_2 };

    IndexedTable table = new IndexedTable(conf, Bytes.toBytes(tablename));

    ResultScanner scanner = table.getIndexedScanner(indexId, indexStartRow,
        indexStopRow, indexColumns, indexFilter, baseColumns);

    for (Result rowResult : scanner) {
      String sum = Bytes.toString(rowResult.getValue(column_1));
      String name = Bytes.toString(rowResult.getValue(column_2));
      System.out.println(name + " 's turnover is " + sum + " $.");
    }
    table.close();
  }

  public static void main(String[] args) throws IOException {

    TSMC4SortTurnover tt = new TSMC4SortTurnover();
    tt.addIndexToTurnover("tsmc", "Sum", "Turnover:Sum");
    tt.readSortedValGreater("130");

  }
}