// Path: themis-client/src/main/java/org/apache/hadoop/hbase/themis/ThemisPut.java
// public class ThemisPut extends ThemisMutation {
//   private Put put;
//   
//   public ThemisPut(byte[] row) {
//     put = new Put(row);
//   }
//   
//   public ThemisPut(Put put) throws IOException {
//     checkContainingPreservedColumns(put.getFamilyMap());
//     setHBasePut(put);
//   }
// 
//   public byte[] getRow() {
//     return put.getRow();
//   }
// 
//   // must specify both the family and qualifier when add mutation
//   public ThemisPut add(byte[] family, byte[] qualifier, byte[] value) throws IOException {
//     checkContainingPreservedColumn(family, qualifier);
//     this.put.add(family, qualifier, Long.MIN_VALUE, value);
//     return this;
//   }
//   
//   protected Put getHBasePut() {
//     return this.put;
//   }
//   
//   protected void setHBasePut(Put put) {
//     this.put = put;
//   }
// 
//   @Override
//   protected boolean hasColumn() {
//     return put.getFamilyMap() != null && put.getFamilyMap().size() != 0;
//   }
// 
//   @Override
//   public Map<byte[], List<KeyValue>> getFamilyMap() {
//     return put.getFamilyMap();
//   }
// }


import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.FirstKeyOnlyFilter;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.RowCounter;
import org.apache.hadoop.hbase.themis.ThemisPut;
import org.apache.hadoop.hbase.themis.Transaction;
import org.apache.hadoop.hbase.themis.mapreduce.ThemisRowCounter.RowCounterMapper;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.mapreduce.Counter;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.output.NullOutputFormat;
import org.junit.Test;
import org.junit.experimental.categories.Category;
        String qualifier = StringUtils.substringAfter(columnName, ":");
        if (StringUtils.isBlank(qualifier)) {
          throw new IOException("must specify qualifier to read themis table");
        } else {
          scan.addColumn(Bytes.toBytes(family), Bytes.toBytes(qualifier));
        }
      }
    }
    
    List<Scan> scans = new ArrayList<Scan>();
    for (String tableName : tableNames) {
      Scan thisScan = new Scan(scan);
      thisScan.setAttribute(Scan.SCAN_ATTRIBUTES_TABLE_NAME, Bytes.toBytes(tableName));
      scans.add(thisScan);
    }
    
    job.setOutputFormatClass(NullOutputFormat.class);
    ThemisTableMapReduceUtil.initTableMapperJob(scans, RowCounterMapper.class,
      ImmutableBytesWritable.class, Result.class, job);
    job.setNumReduceTasks(0);
    return job;
  }
  
  protected void writeTestData() throws IOException {
    Transaction transaction = new Transaction(connection);
    for (byte[] tableName : new byte[][] { TABLENAME, ANOTHER_TABLENAME }) {
      transaction.put(
        tableName,
        new ThemisPut(ROW).add(FAMILY, QUALIFIER, VALUE).add(ANOTHER_FAMILY, QUALIFIER, VALUE)
            .add(FAMILY, ANOTHER_FAMILY, VALUE));
// Path: themis-client/src/main/java/org/apache/hadoop/hbase/themis/ThemisPut.java
// public class ThemisPut extends ThemisMutation {
//   private Put put;
//   
//   public ThemisPut(byte[] row) {
//     put = new Put(row);
//   }
//   
//   public ThemisPut(Put put) throws IOException {
//     checkContainingPreservedColumns(put.getFamilyMap());
//     setHBasePut(put);
//   }
// 
//   public byte[] getRow() {
//     return put.getRow();
//   }
// 
//   // must specify both the family and qualifier when add mutation
//   public ThemisPut add(byte[] family, byte[] qualifier, byte[] value) throws IOException {
//     checkContainingPreservedColumn(family, qualifier);
//     this.put.add(family, qualifier, Long.MIN_VALUE, value);
//     return this;
//   }
//   
//   protected Put getHBasePut() {
//     return this.put;
//   }
//   
//   protected void setHBasePut(Put put) {
//     this.put = put;
//   }
// 
//   @Override
//   protected boolean hasColumn() {
//     return put.getFamilyMap() != null && put.getFamilyMap().size() != 0;
//   }
// 
//   @Override
//   public Map<byte[], List<KeyValue>> getFamilyMap() {
//     return put.getFamilyMap();
//   }
// }


// Path: themis-client/src/test/java/org/apache/hadoop/hbase/themis/mapreduce/TestMultiThemisTableInputFormat.java
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.FirstKeyOnlyFilter;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.RowCounter;
import org.apache.hadoop.hbase.themis.ThemisPut;
import org.apache.hadoop.hbase.themis.Transaction;
import org.apache.hadoop.hbase.themis.mapreduce.ThemisRowCounter.RowCounterMapper;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.mapreduce.Counter;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.output.NullOutputFormat;
import org.junit.Test;
import org.junit.experimental.categories.Category;

        String qualifier = StringUtils.substringAfter(columnName, ":");
        if (StringUtils.isBlank(qualifier)) {
          throw new IOException("must specify qualifier to read themis table");
        } else {
          scan.addColumn(Bytes.toBytes(family), Bytes.toBytes(qualifier));
        }
      }
    }
    
    List<Scan> scans = new ArrayList<Scan>();
    for (String tableName : tableNames) {
      Scan thisScan = new Scan(scan);
      thisScan.setAttribute(Scan.SCAN_ATTRIBUTES_TABLE_NAME, Bytes.toBytes(tableName));
      scans.add(thisScan);
    }
    
    job.setOutputFormatClass(NullOutputFormat.class);
    ThemisTableMapReduceUtil.initTableMapperJob(scans, RowCounterMapper.class,
      ImmutableBytesWritable.class, Result.class, job);
    job.setNumReduceTasks(0);
    return job;
  }
  
  protected void writeTestData() throws IOException {
    Transaction transaction = new Transaction(connection);
    for (byte[] tableName : new byte[][] { TABLENAME, ANOTHER_TABLENAME }) {
      transaction.put(
        tableName,
        new ThemisPut(ROW).add(FAMILY, QUALIFIER, VALUE).add(ANOTHER_FAMILY, QUALIFIER, VALUE)
            .add(FAMILY, ANOTHER_FAMILY, VALUE));
      transaction.put(tableName, new ThemisPut(ANOTHER_ROW).add(FAMILY, QUALIFIER, VALUE));