1 files changed, 17 insertions, 19 deletions

diff --git a/src/main/java/parallelai/spyglass/hbase/HBaseRecordReaderGranular.java b/src/main/java/parallelai/spyglass/hbase/HBaseRecordReaderGranular.java
index 2155d99..fb10615 100644
--- a/src/main/java/parallelai/spyglass/hbase/HBaseRecordReaderGranular.java
+++ b/src/main/java/parallelai/spyglass/hbase/HBaseRecordReaderGranular.java
@@ -1,35 +1,24 @@
 package parallelai.spyglass.hbase;
 
-import static org.apache.hadoop.hbase.mapreduce.TableRecordReaderImpl.LOG_PER_ROW_COUNT;
-
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.TreeSet;
-import java.util.Vector;
-
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
-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.client.ScannerCallable;
-import org.apache.hadoop.hbase.filter.Filter;
 import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
 import org.apache.hadoop.hbase.mapreduce.TableInputFormat;
-import org.apache.hadoop.hbase.util.Bytes; 
+import org.apache.hadoop.hbase.util.Bytes;
 import org.apache.hadoop.hbase.util.Writables;
-import org.apache.hadoop.mapred.RecordReader;
 import org.apache.hadoop.util.StringUtils;
 
-import org.jruby.javasupport.util.RuntimeHelpers;
-import parallelai.spyglass.hbase.HBaseConstants.SourceMode;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Vector;
 
 public class HBaseRecordReaderGranular extends HBaseRecordReaderBase {
 
@@ -72,6 +61,9 @@ public class HBaseRecordReaderGranular extends HBaseRecordReaderBase {
         scan.setFilter(trrRowFilter);
         scan.setCacheBlocks(true);
         scan.setCaching(scanCaching);
+        if(timestamp != -1) {
+            scan.setTimeStamp(timestamp);
+        }
         this.scanner = this.htable.getScanner(scan);
         currentScan = scan;
       } else {
@@ -82,6 +74,9 @@ public class HBaseRecordReaderGranular extends HBaseRecordReaderBase {
         TableInputFormat.addColumns(scan, trrInputColumns);
           scan.setCacheBlocks(true);
           scan.setCaching(scanCaching);
+          if(timestamp != -1) {
+              scan.setTimeStamp(timestamp);
+          }
         this.scanner = this.htable.getScanner(scan);
         currentScan = scan;
       }
@@ -93,7 +88,10 @@ public class HBaseRecordReaderGranular extends HBaseRecordReaderBase {
       TableInputFormat.addColumns(scan, trrInputColumns);
       scan.setFilter(trrRowFilter);
       scan.setCacheBlocks(true);
-        scan.setCaching(scanCaching);
+      scan.setCaching(scanCaching);
+      if(timestamp != -1) {
+        scan.setTimeStamp(timestamp);
+      }
       this.scanner = this.htable.getScanner(scan);
       currentScan = scan;
     }