1 files changed, 140 insertions, 0 deletions
diff --git a/src/main/java/parallelai/spyglass/hbase/HBaseRecordReaderBase.java b/src/main/java/parallelai/spyglass/hbase/HBaseRecordReaderBase.java
new file mode 100644
index 0000000..37858ad
--- /dev/null
+++ b/src/main/java/parallelai/spyglass/hbase/HBaseRecordReaderBase.java
@@ -0,0 +1,140 @@
+package parallelai.spyglass.hbase;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.client.HTable;
+import org.apache.hadoop.hbase.client.Result;
+import org.apache.hadoop.hbase.client.ScannerCallable;
+import org.apache.hadoop.hbase.filter.Filter;
+import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.mapred.RecordReader;
+
+import java.util.TreeSet;
+
+import static org.apache.hadoop.hbase.mapreduce.TableRecordReaderImpl.LOG_PER_ROW_COUNT;
+
+/**
+ * Created with IntelliJ IDEA.
+ * User: chand_000
+ * Date: 29/08/13
+ * Time: 15:42
+ * To change this template use File | Settings | File Templates.
+ */
+public abstract class HBaseRecordReaderBase implements
+        RecordReader<ImmutableBytesWritable, Result> {
+
+    protected TreeSet<String> keyList;
+    protected HBaseConstants.SourceMode sourceMode;
+    protected boolean endRowInclusive = true;
+    protected int versions = 1;
+    protected boolean useSalt = false;
+
+    protected byte[] startRow;
+    protected byte[] endRow;
+
+    protected HTable htable;
+    protected byte[][] trrInputColumns;
+
+    protected Filter trrRowFilter;
+
+    protected boolean logScannerActivity = false;
+    protected int logPerRowCount = 100;
+
+    @Override
+    public String toString() {
+        StringBuffer sbuf = new StringBuffer();
+
+        sbuf.append("".format("HBaseRecordReaderRegional : startRow [%s] endRow [%s] endRowInc [%s] ",
+                Bytes.toString(startRow), Bytes.toString(endRow), endRowInclusive));
+        sbuf.append("".format(" sourceMode [%s] salt [%s] versions [%s] ",
+                sourceMode, useSalt, versions));
+
+        return sbuf.toString();
+    }
+
+    byte[] getStartRow() {
+        return this.startRow;
+    }
+
+    /**
+     * @param htable
+     *          the {@link org.apache.hadoop.hbase.client.HTable} to scan.
+     */
+    public void setHTable(HTable htable) {
+        Configuration conf = htable.getConfiguration();
+        logScannerActivity = conf.getBoolean(ScannerCallable.LOG_SCANNER_ACTIVITY,
+                false);
+        logPerRowCount = conf.getInt(LOG_PER_ROW_COUNT, 100);
+        this.htable = htable;
+    }
+
+    /**
+     * @param inputColumns
+     *          the columns to be placed in {@link Result}.
+     */
+    public void setInputColumns(final byte[][] inputColumns) {
+        this.trrInputColumns = inputColumns;
+    }
+
+    /**
+     * @param startRow
+     *          the first row in the split
+     */
+    public void setStartRow(final byte[] startRow) {
+        this.startRow = startRow;
+    }
+
+    /**
+     *
+     * @param endRow
+     *          the last row in the split
+     */
+    public void setEndRow(final byte[] endRow) {
+        this.endRow = endRow;
+    }
+
+    /**
+     * @param rowFilter
+     *          the {@link org.apache.hadoop.hbase.filter.Filter} to be used.
+     */
+    public void setRowFilter(Filter rowFilter) {
+        this.trrRowFilter = rowFilter;
+    }
+
+    public TreeSet<String> getKeyList() {
+        return keyList;
+    }
+
+    public void setKeyList(TreeSet<String> keyList) {
+        this.keyList = keyList;
+    }
+
+    public void setVersions(int versions) {
+        this.versions = versions;
+    }
+
+    public void setUseSalt(boolean useSalt) {
+        this.useSalt = useSalt;
+    }
+
+    public HBaseConstants.SourceMode getSourceMode() {
+        return sourceMode;
+    }
+
+    public void setSourceMode(HBaseConstants.SourceMode sourceMode) {
+        this.sourceMode = sourceMode;
+    }
+
+    public byte[] getEndRow() {
+        return endRow;
+    }
+
+    public void setEndRowInclusive(boolean isInclusive) {
+        endRowInclusive = isInclusive;
+    }
+
+    public boolean getEndRowInclusive() {
+        return endRowInclusive;
+    }
+
+}