From a9482a23820a74328703d558189af1407b134204 Mon Sep 17 00:00:00 2001
From: Andrew Kyle Purtell <apurtell@apache.org>
Date: Thu, 7 Jan 2010 17:34:15 +0000
Subject: [PATCH] HBASE-2095 TIF should support more confs for the scanner

git-svn-id: https://svn.apache.org/repos/asf/hadoop/hbase/trunk@896935 13f79535-47bb-0310-9956-ffa450edef68
---
 CHANGES.txt                                   |  9 ++-
 .../hbase/mapreduce/TableInputFormat.java     | 63 +++++++++++++++++--
 2 files changed, 63 insertions(+), 9 deletions(-)

diff --git a/CHANGES.txt b/CHANGES.txt
index 20ef6f380ef..0586eb6a201 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -277,9 +277,10 @@ Release 0.21.0 - Unreleased
    HBASE-2080  [EC2] Support multivolume local instance storage
    HBASE-2083  [EC2] HDFS DataNode no longer required on master
    HBASE-2084  [EC2] JAVA_HOME handling broken
-   HBASE-2036  Use Configuration instead of HBaseConfiguration (Enis Soztutar via Stack)
-   HBASE-2085  StringBuffer -> StringBuilder - conversion of references as necessary
-               (Kay Kay via Stack)
+   HBASE-2036  Use Configuration instead of HBaseConfiguration (Enis Soztutar
+               via Stack)
+   HBASE-2085  StringBuffer -> StringBuilder - conversion of references as
+               necessary (Kay Kay via Stack)
    HBASE-2052  Upper bound of outstanding WALs can be overrun
    HBASE-2086  Job(configuration,String) deprecated (Kay Kay via Stack)
    HBASE-1996  Configure scanner buffer in bytes instead of number of rows
@@ -287,6 +288,8 @@ Release 0.21.0 - Unreleased
    HBASE-2090  findbugs issues (Kay Kay via Stack)
    HBASE-2089  HBaseConfiguration() ctor. deprecated (Kay Kay via Stack)
    HBASE-2035  Binary values are formatted wrong in shell
+   HBASE-2095  TIF shuold support more confs for the scanner (Bassam Tabbara
+               via Andrew Purtell)
 
   NEW FEATURES
    HBASE-1901  "General" partitioner for "hbase-48" bulk (behind the api, write
diff --git a/src/java/org/apache/hadoop/hbase/mapreduce/TableInputFormat.java b/src/java/org/apache/hadoop/hbase/mapreduce/TableInputFormat.java
index 825a437cfd2..7dba46cc028 100644
--- a/src/java/org/apache/hadoop/hbase/mapreduce/TableInputFormat.java
+++ b/src/java/org/apache/hadoop/hbase/mapreduce/TableInputFormat.java
@@ -38,10 +38,24 @@ implements Configurable {
   
   private final Log LOG = LogFactory.getLog(TableInputFormat.class);
   
-  /** Job parameter that specifies the output table. */
+  /** Job parameter that specifies the input table. */
   public static final String INPUT_TABLE = "hbase.mapreduce.inputtable";
-  /** Space delimited list of columns. */
+  /** Base-64 encoded scanner. All other SCAN_ confs are ignored if this is specified. */
   public static final String SCAN = "hbase.mapreduce.scan";
+  /** Space delimited list of columns to scan. */
+  public static final String SCAN_COLUMNS = "hbase.mapreduce.scan.columns";
+  /** The timestamp used to filter columns with a specific timestamp. */
+  public static final String SCAN_TIMESTAMP = "hbase.mapreduce.scan.timestamp";
+  /** The starting timestamp used to filter columns with a specific range of versions. */
+  public static final String SCAN_TIMERANGE_START = "hbase.mapreduce.scan.timerange.start";
+  /** The ending timestamp used to filter columns with a specific range of versions. */
+  public static final String SCAN_TIMERANGE_END = "hbase.mapreduce.scan.timerange.end";
+  /** The maximum number of version to return. */
+  public static final String SCAN_MAXVERSIONS = "hbase.mapreduce.scan.maxversions";
+  /** Set to false to disable server-side caching of blocks for this scan. */
+  public static final String SCAN_CACHEBLOCKS = "hbase.mapreduce.scan.cacheblocks";
+  /** The number of rows for caching that will be passed to scanners. */
+  public static final String SCAN_CACHEDROWS = "hbase.mapreduce.scan.cachedrows";
   
   /** The configuration. */
   private Configuration conf = null;
@@ -74,12 +88,49 @@ implements Configurable {
     } catch (Exception e) {
       LOG.error(StringUtils.stringifyException(e));
     }
+    
     Scan scan = null;
-    try {
-      scan = TableMapReduceUtil.convertStringToScan(conf.get(SCAN));
-    } catch (IOException e) {
-      LOG.error("An error occurred.", e);
+    
+    if (conf.get(SCAN) != null) {
+      try {
+        scan = TableMapReduceUtil.convertStringToScan(conf.get(SCAN));
+      } catch (IOException e) {
+        LOG.error("An error occurred.", e);
+      }
+    } else {
+      try {
+        scan = new Scan();
+        
+        if (conf.get(SCAN_COLUMNS) != null) {
+          scan.addColumns(conf.get(SCAN_COLUMNS));
+        }
+        
+        if (conf.get(SCAN_TIMESTAMP) != null) {
+          scan.setTimeStamp(Long.parseLong(conf.get(SCAN_TIMESTAMP)));
+        }
+        
+        if (conf.get(SCAN_TIMERANGE_START) != null && conf.get(SCAN_TIMERANGE_END) != null) {
+          scan.setTimeRange(
+              Long.parseLong(conf.get(SCAN_TIMERANGE_START)), 
+              Long.parseLong(conf.get(SCAN_TIMERANGE_END)));
+        }
+
+        if (conf.get(SCAN_MAXVERSIONS) != null) {
+          scan.setMaxVersions(Integer.parseInt(conf.get(SCAN_MAXVERSIONS)));
+        }
+        
+        if (conf.get(SCAN_CACHEBLOCKS) != null) {
+          scan.setCacheBlocks(Boolean.parseBoolean(conf.get(SCAN_CACHEBLOCKS)));
+        }
+
+        if (conf.get(SCAN_CACHEDROWS) != null) {
+          scan.setCaching(Integer.parseInt(conf.get(SCAN_CACHEDROWS)));
+        }
+      } catch (Exception e) {
+          LOG.error(StringUtils.stringifyException(e));
+      }
     }
+
     setScan(scan);
   }