HBASE-4938 Create a HRegion.getScanner public method that allows reading from a specified readPoint (Dhruba)

git-svn-id: https://svn.apache.org/repos/asf/hbase/trunk@1222890 13f79535-47bb-0310-9956-ffa450edef68

src/main/java/org/apache/hadoop/hbase/client/IsolationLevel.java

@@ -0,0 +1,54 @@
+/*
+ * Copyright The Apache Software Foundation
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.hbase.client;
+
+/**
+ * Specify Isolation levels in Scan operations.
+ * <p>
+ * There are two isolation levels. A READ_COMMITTED isolation level
+ * indicates that only data that is committed be returned in a scan.
+ * An isolation level of READ_UNCOMMITTED indicates that a scan
+ * should return data that is being modified by transactions that might
+ * not have been committed yet.
+ */
+public enum IsolationLevel {
+
+  READ_COMMITTED(1),
+  READ_UNCOMMITTED(2);
+
+  IsolationLevel(int value) {}
+
+  public byte [] toBytes() {
+    return new byte [] { toByte() };
+  }
+
+  public byte toByte() {
+    return (byte)this.ordinal();
+  }
+
+  public static IsolationLevel fromBytes(byte [] bytes) {
+    return IsolationLevel.fromByte(bytes[0]);
+  }
+
+  public static IsolationLevel fromByte(byte vbyte) {
+    return IsolationLevel.values()[vbyte];
+  }
+}

src/main/java/org/apache/hadoop/hbase/client/Scan.java

@@ -82,6 +82,7 @@ import java.util.TreeSet;
  */
 public class Scan extends OperationWithAttributes implements Writable {
   private static final String RAW_ATTR = "_raw_";
+  private static final String ISOLATION_LEVEL = "_isolationlevel_";
 
   private static final byte SCAN_VERSION = (byte)2;
   private byte [] startRow = HConstants.EMPTY_START_ROW;
@@ -632,4 +633,30 @@ public class Scan extends OperationWithAttributes implements Writable {
     byte[] attr = getAttribute(RAW_ATTR);
     return attr == null ? false : Bytes.toBoolean(attr);
   }
+
+  /*
+   * Set the isolation level for this scan. If the
+   * isolation level is set to READ_UNCOMMITTED, then
+   * this scan will return data from committed and
+   * uncommitted transactions. If the isolation level 
+   * is set to READ_COMMITTED, then this scan will return 
+   * data from committed transactions only. If a isolation
+   * level is not explicitly set on a Scan, then it 
+   * is assumed to be READ_COMMITTED.
+   * @param level IsolationLevel for this scan
+   */
+  public void setIsolationLevel(IsolationLevel level) {
+    setAttribute(ISOLATION_LEVEL, level.toBytes());
+  }
+  /*
+   * @return The isolation level of this scan.
+   * If no isolation level was set for this scan object, 
+   * then it returns READ_COMMITTED.
+   * @return The IsolationLevel for this scan
+   */
+  public IsolationLevel getIsolationLevel() {
+    byte[] attr = getAttribute(ISOLATION_LEVEL);
+    return attr == null ? IsolationLevel.READ_COMMITTED :
+                          IsolationLevel.fromBytes(attr);
+  }
 }

src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java

@@ -74,6 +74,7 @@ import org.apache.hadoop.hbase.client.Append;
 import org.apache.hadoop.hbase.client.Delete;
 import org.apache.hadoop.hbase.client.Get;
 import org.apache.hadoop.hbase.client.Increment;
+import org.apache.hadoop.hbase.client.IsolationLevel;
 import org.apache.hadoop.hbase.client.Put;
 import org.apache.hadoop.hbase.client.Result;
 import org.apache.hadoop.hbase.client.Row;
@@ -3154,8 +3155,15 @@ public class HRegion implements HeapSize { // , Writable{
 
       // synchronize on scannerReadPoints so that nobody calculates
       // getSmallestReadPoint, before scannerReadPoints is updated.
+      IsolationLevel isolationLevel = scan.getIsolationLevel();
       synchronized(scannerReadPoints) {
-        this.readPt = MultiVersionConsistencyControl.resetThreadReadPoint(mvcc);
+        if (isolationLevel == IsolationLevel.READ_UNCOMMITTED) {
+          // This scan can read even uncommitted transactions
+          this.readPt = Long.MAX_VALUE;
+          MultiVersionConsistencyControl.setThreadReadPoint(this.readPt);
+        } else {
+          this.readPt = MultiVersionConsistencyControl.resetThreadReadPoint(mvcc);
+        }
         scannerReadPoints.put(this, this.readPt);
       }