fluo-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ktur...@apache.org
Subject [fluo] branch master updated: fixes #917 Added read locks (#953)
Date Wed, 01 Nov 2017 14:15:10 GMT
This is an automated email from the ASF dual-hosted git repository.

kturner pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/fluo.git


The following commit(s) were added to refs/heads/master by this push:
     new a18c045  fixes #917 Added read locks (#953)
a18c045 is described below

commit a18c04525bd41cf4140f66d26d4405030a96a1e9
Author: Keith Turner <kturner@apache.org>
AuthorDate: Wed Nov 1 10:13:29 2017 -0400

    fixes #917 Added read locks (#953)
---
 .../apache/fluo/accumulo/format/FluoFormatter.java |  16 +
 .../iterators/GarbageCollectionIterator.java       |  35 ++
 .../accumulo/iterators/OpenReadLockIterator.java   | 133 +++++
 .../fluo/accumulo/iterators/PrewriteIterator.java  |  65 ++-
 .../accumulo/iterators/RollbackCheckIterator.java  |   3 +
 .../fluo/accumulo/iterators/SnapshotIterator.java  |  96 +++-
 .../apache/fluo/accumulo/util/ColumnConstants.java |   1 +
 .../{ColumnConstants.java => ReadLockUtil.java}    |  37 +-
 .../fluo/accumulo/values/DelReadLockValue.java     |  58 +++
 .../apache/fluo/accumulo/values/ReadLockValue.java |  76 +++
 .../fluo/accumulo/iterators/CountingIterator.java  |  63 +++
 .../iterators/GarbageCollectionIteratorTest.java   | 100 ++++
 .../iterators/OpenReadLockIteratorTest.java        |  90 ++++
 .../accumulo/iterators/PrewriteIteratorTest.java   | 238 ++++++++-
 .../accumulo/iterators/SnapshotIteratorTest.java   |  81 ++-
 .../apache/fluo/accumulo/iterators/TestData.java   |  37 +-
 .../fluo/api/client/AbstractSnapshotBase.java      |  19 +
 .../fluo/api/client/AbstractTransactionBase.java   |   3 +
 .../org/apache/fluo/api/client/SnapshotBase.java   |   1 +
 .../apache/fluo/api/client/TransactionBase.java    |  26 +-
 .../java/org/apache/fluo/command/ScanTest.java     |   2 +-
 .../org/apache/fluo/core/impl/LockResolver.java    | 193 +++++---
 .../fluo/core/impl/ParallelSnapshotScanner.java    |  25 +-
 .../apache/fluo/core/impl/ReadLockSnapshot.java    |  79 +++
 .../org/apache/fluo/core/impl/SnapshotScanner.java |  27 +-
 .../org/apache/fluo/core/impl/TransactionImpl.java | 233 +++++++--
 .../apache/fluo/core/log/TracingTransaction.java   |  95 +++-
 .../java/org/apache/fluo/core/util/ColumnUtil.java |  12 +-
 .../java/org/apache/fluo/integration/ITBase.java   |  13 +-
 .../apache/fluo/integration/TestTransaction.java   |   6 +
 .../impl/GarbageCollectionIteratorIT.java          |  83 ++++
 .../fluo/integration/impl/ReadLockFailureIT.java   | 427 ++++++++++++++++
 .../apache/fluo/integration/impl/ReadLockIT.java   | 548 +++++++++++++++++++++
 .../apache/fluo/integration/impl/ScannerIT.java    |   2 +-
 .../org/apache/fluo/integration/log/LogIT.java     |  69 +++
 35 files changed, 2791 insertions(+), 201 deletions(-)

diff --git a/modules/accumulo/src/main/java/org/apache/fluo/accumulo/format/FluoFormatter.java b/modules/accumulo/src/main/java/org/apache/fluo/accumulo/format/FluoFormatter.java
index 056fa8d..d87975a 100644
--- a/modules/accumulo/src/main/java/org/apache/fluo/accumulo/format/FluoFormatter.java
+++ b/modules/accumulo/src/main/java/org/apache/fluo/accumulo/format/FluoFormatter.java
@@ -22,8 +22,11 @@ import org.apache.accumulo.core.data.Key;
 import org.apache.accumulo.core.data.Value;
 import org.apache.fluo.accumulo.util.ColumnConstants;
 import org.apache.fluo.accumulo.util.NotificationUtil;
+import org.apache.fluo.accumulo.util.ReadLockUtil;
 import org.apache.fluo.accumulo.values.DelLockValue;
+import org.apache.fluo.accumulo.values.DelReadLockValue;
 import org.apache.fluo.accumulo.values.LockValue;
+import org.apache.fluo.accumulo.values.ReadLockValue;
 import org.apache.fluo.accumulo.values.WriteValue;
 import org.apache.fluo.api.data.Bytes;
 import org.apache.fluo.api.data.Column;
@@ -104,6 +107,15 @@ public class FluoFormatter {
       if ((ts & ColumnConstants.PREFIX_MASK) == ColumnConstants.ACK_PREFIX) {
         type = "ACK";
       }
+      if ((ts & ColumnConstants.PREFIX_MASK) == ColumnConstants.RLOCK_PREFIX) {
+        if (ReadLockUtil.isDelete(ts)) {
+          type = "DEL_RLOCK";
+        } else {
+          type = "RLOCK";
+        }
+        ts = ReadLockUtil.decodeTs(ts);
+      }
+
 
       StringBuilder sb = new StringBuilder();
 
@@ -115,6 +127,10 @@ public class FluoFormatter {
       } else if (type.equals("LOCK")) {
         // TODO can Value be made to extend Bytes w/o breaking API?
         val = new LockValue(entry.getValue().get()).toString();
+      } else if (type.equals("RLOCK")) {
+        val = new ReadLockValue(entry.getValue().get()).toString();
+      } else if (type.equals("DEL_RLOCK")) {
+        val = new DelReadLockValue(entry.getValue().get()).toString();
       } else {
         encNonAscii(sb, entry.getValue().get());
         val = sb.toString();
diff --git a/modules/accumulo/src/main/java/org/apache/fluo/accumulo/iterators/GarbageCollectionIterator.java b/modules/accumulo/src/main/java/org/apache/fluo/accumulo/iterators/GarbageCollectionIterator.java
index 9ca23e9..1814dfa 100644
--- a/modules/accumulo/src/main/java/org/apache/fluo/accumulo/iterators/GarbageCollectionIterator.java
+++ b/modules/accumulo/src/main/java/org/apache/fluo/accumulo/iterators/GarbageCollectionIterator.java
@@ -35,8 +35,10 @@ import org.apache.accumulo.core.iterators.IteratorEnvironment;
 import org.apache.accumulo.core.iterators.IteratorUtil.IteratorScope;
 import org.apache.accumulo.core.iterators.SortedKeyValueIterator;
 import org.apache.fluo.accumulo.util.ColumnConstants;
+import org.apache.fluo.accumulo.util.ReadLockUtil;
 import org.apache.fluo.accumulo.util.ZookeeperUtil;
 import org.apache.fluo.accumulo.values.DelLockValue;
+import org.apache.fluo.accumulo.values.DelReadLockValue;
 import org.apache.fluo.accumulo.values.WriteValue;
 
 /**
@@ -159,6 +161,7 @@ public class GarbageCollectionIterator implements SortedKeyValueIterator<Key, Va
     boolean oldestSeen = false;
     boolean sawAck = false;
     long firstWrite = -1;
+    long lastReadLockDeleteTs = -1;
 
     truncationTime = -1;
 
@@ -243,6 +246,38 @@ public class GarbageCollectionIterator implements SortedKeyValueIterator<Key, Va
         } else if (complete) {
           completeTxs.remove(txDoneTs);
         }
+      } else if (colType == ColumnConstants.RLOCK_PREFIX) {
+        boolean keep = false;
+        long rlts = ReadLockUtil.decodeTs(ts);
+        boolean isDelete = ReadLockUtil.isDelete(ts);
+
+        if (isDelete) {
+          lastReadLockDeleteTs = rlts;
+        }
+
+        if (rlts > invalidationTime) {
+          if (isFullMajc) {
+            if (isDelete) {
+              if (DelReadLockValue.isRollback(source.getTopValue().get())) {
+                // can drop rolled back read lock delete markers on any full majc, do not need to consider gcTimestamp
+                keep = false;
+              } else {
+                long rlockCommitTs =
+                    DelReadLockValue.getCommitTimestamp(source.getTopValue().get());
+                keep = rlockCommitTs >= gcTimestamp;
+              }
+            } else {
+              keep = lastReadLockDeleteTs != rlts;
+            }
+          } else {
+            // can drop deleted read lock entries.. keep the delete entry.
+            keep = isDelete || lastReadLockDeleteTs != rlts;
+          }
+        }
+
+        if (keep) {
+          keys.add(source.getTopKey(), source.getTopValue());
+        }
       } else if (colType == ColumnConstants.LOCK_PREFIX) {
         if (ts > invalidationTime) {
           keys.add(source.getTopKey(), source.getTopValue());
diff --git a/modules/accumulo/src/main/java/org/apache/fluo/accumulo/iterators/OpenReadLockIterator.java b/modules/accumulo/src/main/java/org/apache/fluo/accumulo/iterators/OpenReadLockIterator.java
new file mode 100644
index 0000000..dd4de54
--- /dev/null
+++ b/modules/accumulo/src/main/java/org/apache/fluo/accumulo/iterators/OpenReadLockIterator.java
@@ -0,0 +1,133 @@
+/*
+ * 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.fluo.accumulo.iterators;
+
+import java.io.IOException;
+import java.util.Collection;
+import java.util.Map;
+
+import org.apache.accumulo.core.data.ByteSequence;
+import org.apache.accumulo.core.data.Key;
+import org.apache.accumulo.core.data.PartialKey;
+import org.apache.accumulo.core.data.Range;
+import org.apache.accumulo.core.data.Value;
+import org.apache.accumulo.core.iterators.IteratorEnvironment;
+import org.apache.accumulo.core.iterators.SortedKeyValueIterator;
+import org.apache.fluo.accumulo.util.ColumnConstants;
+import org.apache.fluo.accumulo.util.ReadLockUtil;
+
+import static org.apache.fluo.accumulo.util.ColumnConstants.ACK_PREFIX;
+import static org.apache.fluo.accumulo.util.ColumnConstants.DATA_PREFIX;
+import static org.apache.fluo.accumulo.util.ColumnConstants.DEL_LOCK_PREFIX;
+import static org.apache.fluo.accumulo.util.ColumnConstants.LOCK_PREFIX;
+import static org.apache.fluo.accumulo.util.ColumnConstants.RLOCK_PREFIX;
+import static org.apache.fluo.accumulo.util.ColumnConstants.TIMESTAMP_MASK;
+import static org.apache.fluo.accumulo.util.ColumnConstants.TX_DONE_PREFIX;
+import static org.apache.fluo.accumulo.util.ColumnConstants.WRITE_PREFIX;
+
+public class OpenReadLockIterator implements SortedKeyValueIterator<Key, Value> {
+
+  private TimestampSkippingIterator source;
+
+  private Key lastDelete;
+
+  private void findTop() throws IOException {
+    while (source.hasTop()) {
+
+      long colType = source.getTopKey().getTimestamp() & ColumnConstants.PREFIX_MASK;
+
+      if (colType == TX_DONE_PREFIX || colType == WRITE_PREFIX || colType == DEL_LOCK_PREFIX) {
+        source.skipToPrefix(source.getTopKey(), RLOCK_PREFIX);
+        continue;
+      } else if (colType == RLOCK_PREFIX) {
+        if (ReadLockUtil.isDelete(source.getTopKey())) {
+          lastDelete.set(source.getTopKey());
+        } else {
+          if (lastDelete.equals(source.getTopKey(), PartialKey.ROW_COLFAM_COLQUAL_COLVIS)) {
+            long ts1 = ReadLockUtil.decodeTs(source.getTopKey().getTimestamp() & TIMESTAMP_MASK);
+            long ts2 = ReadLockUtil.decodeTs(lastDelete.getTimestamp() & TIMESTAMP_MASK);
+
+            if (ts1 != ts2) {
+              // found a read lock that is not suppressed by a delete read lock entry
+              return;
+            }
+          } else {
+            // found a read lock that is not suppressed by a delete read lock entry
+            return;
+          }
+        }
+        source.next();
+        continue;
+      } else if (colType == DATA_PREFIX || colType == LOCK_PREFIX || colType == ACK_PREFIX) {
+        source.skipColumn(source.getTopKey());
+        continue;
+      } else {
+        throw new IllegalArgumentException("Unknown column type " + source.getTopKey());
+      }
+    }
+  }
+
+  @Override
+  public void init(SortedKeyValueIterator<Key, Value> source, Map<String, String> options,
+      IteratorEnvironment env) throws IOException {
+    this.source = new TimestampSkippingIterator(source);
+  }
+
+  @Override
+  public boolean hasTop() {
+    return source.hasTop();
+  }
+
+  @Override
+  public void next() throws IOException {
+    source.next();
+    findTop();
+  }
+
+  @Override
+  public void seek(Range range, Collection<ByteSequence> columnFamilies, boolean inclusive)
+      throws IOException {
+
+    lastDelete = new Key();
+
+    Collection<ByteSequence> fams;
+    if (columnFamilies.isEmpty() && !inclusive) {
+      fams = SnapshotIterator.NOTIFY_CF_SET;
+      inclusive = false;
+    } else {
+      fams = columnFamilies;
+    }
+
+    source.seek(range, fams, inclusive);
+    findTop();
+  }
+
+  @Override
+  public Key getTopKey() {
+    return source.getTopKey();
+  }
+
+  @Override
+  public Value getTopValue() {
+    return source.getTopValue();
+  }
+
+  @Override
+  public SortedKeyValueIterator<Key, Value> deepCopy(IteratorEnvironment env) {
+    throw new UnsupportedOperationException();
+  }
+
+}
diff --git a/modules/accumulo/src/main/java/org/apache/fluo/accumulo/iterators/PrewriteIterator.java b/modules/accumulo/src/main/java/org/apache/fluo/accumulo/iterators/PrewriteIterator.java
index a5a160d..b6f9a48 100644
--- a/modules/accumulo/src/main/java/org/apache/fluo/accumulo/iterators/PrewriteIterator.java
+++ b/modules/accumulo/src/main/java/org/apache/fluo/accumulo/iterators/PrewriteIterator.java
@@ -28,6 +28,8 @@ import org.apache.accumulo.core.data.Value;
 import org.apache.accumulo.core.iterators.IteratorEnvironment;
 import org.apache.accumulo.core.iterators.SortedKeyValueIterator;
 import org.apache.fluo.accumulo.util.ColumnConstants;
+import org.apache.fluo.accumulo.util.ReadLockUtil;
+import org.apache.fluo.accumulo.values.DelReadLockValue;
 import org.apache.fluo.accumulo.values.WriteValue;
 
 /**
@@ -37,6 +39,7 @@ public class PrewriteIterator implements SortedKeyValueIterator<Key, Value> {
   private static final String TIMESTAMP_OPT = "timestampOpt";
   private static final String CHECK_ACK_OPT = "checkAckOpt";
   private static final String NTFY_TIMESTAMP_OPT = "ntfyTsOpt";
+  private static final String READ_LOCK_OPT = "readLock";
 
   private TimestampSkippingIterator source;
   private long snaptime;
@@ -44,6 +47,7 @@ public class PrewriteIterator implements SortedKeyValueIterator<Key, Value> {
   boolean hasTop = false;
   boolean checkAck = false;
   long ntfyTimestamp = -1;
+  boolean readlock;
 
   public static void setSnaptime(IteratorSetting cfg, long time) {
     if (time < 0 || (ColumnConstants.PREFIX_MASK & time) != 0) {
@@ -52,8 +56,12 @@ public class PrewriteIterator implements SortedKeyValueIterator<Key, Value> {
     cfg.addOption(TIMESTAMP_OPT, time + "");
   }
 
+  public static void setReadlock(IteratorSetting cfg) {
+    cfg.addOption(READ_LOCK_OPT, Boolean.TRUE.toString());
+  }
+
   public static void enableAckCheck(IteratorSetting cfg, long timestamp) {
-    cfg.addOption(CHECK_ACK_OPT, "true");
+    cfg.addOption(CHECK_ACK_OPT, Boolean.TRUE.toString());
     cfg.addOption(NTFY_TIMESTAMP_OPT, timestamp + "");
   }
 
@@ -66,6 +74,8 @@ public class PrewriteIterator implements SortedKeyValueIterator<Key, Value> {
       this.checkAck = Boolean.parseBoolean(options.get(CHECK_ACK_OPT));
       this.ntfyTimestamp = Long.parseLong(options.get(NTFY_TIMESTAMP_OPT));
     }
+
+    this.readlock = Boolean.parseBoolean(options.getOrDefault(READ_LOCK_OPT, "false"));
   }
 
   @Override
@@ -140,7 +150,58 @@ public class PrewriteIterator implements SortedKeyValueIterator<Key, Value> {
           }
         }
 
-        source.skipToPrefix(seekRange.getStartKey(), ColumnConstants.LOCK_PREFIX);
+        if (readlock) {
+          source.skipToPrefix(seekRange.getStartKey(), ColumnConstants.LOCK_PREFIX);
+        } else {
+          source.skipToPrefix(seekRange.getStartKey(), ColumnConstants.RLOCK_PREFIX);
+        }
+      } else if (colType == ColumnConstants.RLOCK_PREFIX) {
+
+        long lastDeleteTs = -1;
+        long rlts = ReadLockUtil.decodeTs(ts);
+
+        if (!readlock) {
+          while (rlts > invalidationTime && colType == ColumnConstants.RLOCK_PREFIX) {
+            if (ReadLockUtil.isDelete(ts)) {
+              // ignore rolled back read locks, these should never prevent a write lock
+              if (!DelReadLockValue.isRollback(source.getTopValue().get())) {
+                if (rlts >= snaptime) {
+                  hasTop = true;
+                  return;
+                } else {
+                  long rlockCommitTs =
+                      DelReadLockValue.getCommitTimestamp(source.getTopValue().get());
+                  if (rlockCommitTs > snaptime) {
+                    hasTop = true;
+                    return;
+                  }
+                }
+              }
+
+
+              lastDeleteTs = rlts;
+            } else {
+              if (rlts != lastDeleteTs) {
+                // this read lock is active
+                hasTop = true;
+                return;
+              }
+            }
+
+            source.next();
+            if (source.hasTop()) {
+              colType = source.getTopKey().getTimestamp() & ColumnConstants.PREFIX_MASK;
+              ts = source.getTopKey().getTimestamp() & ColumnConstants.TIMESTAMP_MASK;
+              rlts = ReadLockUtil.decodeTs(ts);
+            } else {
+              break;
+            }
+          }
+        }
+
+        if (source.hasTop() && (colType == ColumnConstants.RLOCK_PREFIX)) {
+          source.skipToPrefix(seekRange.getStartKey(), ColumnConstants.LOCK_PREFIX);
+        }
       } else if (colType == ColumnConstants.LOCK_PREFIX) {
         if (ts > invalidationTime) {
           // nothing supersedes this lock, therefore the column is locked
diff --git a/modules/accumulo/src/main/java/org/apache/fluo/accumulo/iterators/RollbackCheckIterator.java b/modules/accumulo/src/main/java/org/apache/fluo/accumulo/iterators/RollbackCheckIterator.java
index 962f172..cae383f 100644
--- a/modules/accumulo/src/main/java/org/apache/fluo/accumulo/iterators/RollbackCheckIterator.java
+++ b/modules/accumulo/src/main/java/org/apache/fluo/accumulo/iterators/RollbackCheckIterator.java
@@ -129,6 +129,9 @@ public class RollbackCheckIterator implements SortedKeyValueIterator<Key, Value>
           continue;
         }
 
+      } else if (colType == ColumnConstants.RLOCK_PREFIX) {
+        source.skipToPrefix(curCol, ColumnConstants.LOCK_PREFIX);
+        continue;
       } else if (colType == ColumnConstants.LOCK_PREFIX) {
         if (ts > invalidationTime) {
           // nothing supersedes this lock, therefore the column is locked
diff --git a/modules/accumulo/src/main/java/org/apache/fluo/accumulo/iterators/SnapshotIterator.java b/modules/accumulo/src/main/java/org/apache/fluo/accumulo/iterators/SnapshotIterator.java
index 3c6939c..27e63ef 100644
--- a/modules/accumulo/src/main/java/org/apache/fluo/accumulo/iterators/SnapshotIterator.java
+++ b/modules/accumulo/src/main/java/org/apache/fluo/accumulo/iterators/SnapshotIterator.java
@@ -22,6 +22,7 @@ import java.util.Map;
 import java.util.Set;
 
 import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
 import org.apache.accumulo.core.client.IteratorSetting;
 import org.apache.accumulo.core.data.ArrayByteSequence;
 import org.apache.accumulo.core.data.ByteSequence;
@@ -38,6 +39,9 @@ public class SnapshotIterator implements SortedKeyValueIterator<Key, Value> {
 
   @VisibleForTesting
   static final String TIMESTAMP_OPT = "timestampOpt";
+
+  static final String RETURN_READLOCK_PRESENT_OPT = "rrlpOpt";
+
   private static final ByteSequence NOTIFY_CF_BS =
       new ArrayByteSequence(ColumnConstants.NOTIFY_CF.toArray());
 
@@ -47,10 +51,35 @@ public class SnapshotIterator implements SortedKeyValueIterator<Key, Value> {
   private long snaptime;
   private boolean hasTop = false;
 
+  private boolean returnReadLockPresent = false;
+
   private final Key curCol = new Key();
 
+  private Key readLockIgnore;
+  private Key readLockKey;
+  private Value readLockValue;
+
+  private void rememberReadLock(Key key, Value val) {
+    Preconditions.checkState(readLockKey == null && readLockValue == null);
+    if (readLockIgnore == null
+        || !key.equals(readLockIgnore, PartialKey.ROW_COLFAM_COLQUAL_COLVIS)) {
+      readLockKey = new Key(key);
+      readLockValue = new Value(val);
+    }
+  }
+
+  private void ignoreReadLock(Key key) {
+    readLockIgnore = key;
+  }
+
+  private void clearReadLock() {
+    readLockKey = null;
+    readLockValue = null;
+    readLockIgnore = null;
+  }
+
   private void findTop() throws IOException {
-    outer: while (source.hasTop()) {
+    outer: while (source.hasTop() && readLockKey == null) {
       long invalidationTime = -1;
       long dataPointer = -1;
 
@@ -89,6 +118,18 @@ public class SnapshotIterator implements SortedKeyValueIterator<Key, Value> {
           if (ts > invalidationTime) {
             invalidationTime = ts;
           }
+          if (returnReadLockPresent) {
+            source.skipToPrefix(curCol, ColumnConstants.RLOCK_PREFIX);
+          } else {
+            source.skipToPrefix(curCol, ColumnConstants.LOCK_PREFIX);
+          }
+          continue;
+
+        } else if (colType == ColumnConstants.RLOCK_PREFIX) {
+          if (returnReadLockPresent) {
+            rememberReadLock(source.getTopKey(), source.getTopValue());
+          }
+
           source.skipToPrefix(curCol, ColumnConstants.LOCK_PREFIX);
           continue;
         } else if (colType == ColumnConstants.LOCK_PREFIX) {
@@ -142,21 +183,26 @@ public class SnapshotIterator implements SortedKeyValueIterator<Key, Value> {
       IteratorEnvironment env) throws IOException {
     this.source = new TimestampSkippingIterator(source);
     this.snaptime = Long.parseLong(options.get(TIMESTAMP_OPT));
+    this.returnReadLockPresent =
+        Boolean.parseBoolean(options.getOrDefault(RETURN_READLOCK_PRESENT_OPT, "false"));
     // TODO could require client to send version as a sanity check
   }
 
   @Override
   public boolean hasTop() {
-    return hasTop && source.hasTop();
+    return hasTop && (readLockKey != null || source.hasTop());
   }
 
   @Override
   public void next() throws IOException {
-    curCol.set(source.getTopKey());
-    source.skipColumn(curCol);
-
-    findTop();
+    if (readLockKey != null) {
+      clearReadLock();
+    } else {
+      curCol.set(source.getTopKey());
+      source.skipColumn(curCol);
 
+      findTop();
+    }
   }
 
   @Override
@@ -167,16 +213,28 @@ public class SnapshotIterator implements SortedKeyValueIterator<Key, Value> {
     Collection<ByteSequence> cols;
     boolean inc;
 
+    clearReadLock();
+
     // handle continue case
     hasTop = true;
     if (range.getStartKey() != null && range.getStartKey().getTimestamp() != Long.MAX_VALUE
         && !range.isStartKeyInclusive()) {
-      Key nextCol = range.getStartKey().followingKey(PartialKey.ROW_COLFAM_COLQUAL_COLVIS);
-      if (range.afterEndKey(nextCol)) {
-        hasTop = false;
-        return;
+
+      if ((range.getStartKey().getTimestamp()
+          & ColumnConstants.PREFIX_MASK) == ColumnConstants.RLOCK_PREFIX) {
+        Key currCol = new Key(range.getStartKey());
+        currCol.setTimestamp(Long.MAX_VALUE);
+        newRange = new Range(currCol, true, range.getEndKey(), range.isEndKeyInclusive());
+        ignoreReadLock(currCol);
       } else {
-        newRange = new Range(nextCol, true, range.getEndKey(), range.isEndKeyInclusive());
+
+        Key nextCol = range.getStartKey().followingKey(PartialKey.ROW_COLFAM_COLQUAL_COLVIS);
+        if (range.afterEndKey(nextCol)) {
+          hasTop = false;
+          return;
+        } else {
+          newRange = new Range(nextCol, true, range.getEndKey(), range.isEndKeyInclusive());
+        }
       }
     } else {
       newRange = range;
@@ -196,12 +254,20 @@ public class SnapshotIterator implements SortedKeyValueIterator<Key, Value> {
 
   @Override
   public Key getTopKey() {
-    return source.getTopKey();
+    if (readLockKey != null) {
+      return readLockKey;
+    } else {
+      return source.getTopKey();
+    }
   }
 
   @Override
   public Value getTopValue() {
-    return source.getTopValue();
+    if (readLockValue != null) {
+      return readLockValue;
+    } else {
+      return source.getTopValue();
+    }
   }
 
   @Override
@@ -215,4 +281,8 @@ public class SnapshotIterator implements SortedKeyValueIterator<Key, Value> {
     }
     cfg.addOption(TIMESTAMP_OPT, time + "");
   }
+
+  public static void setReturnReadLockPresent(IteratorSetting cfg, boolean rrlp) {
+    cfg.addOption(RETURN_READLOCK_PRESENT_OPT, rrlp + "");
+  }
 }
diff --git a/modules/accumulo/src/main/java/org/apache/fluo/accumulo/util/ColumnConstants.java b/modules/accumulo/src/main/java/org/apache/fluo/accumulo/util/ColumnConstants.java
index 4da1a48..c38b9e6 100644
--- a/modules/accumulo/src/main/java/org/apache/fluo/accumulo/util/ColumnConstants.java
+++ b/modules/accumulo/src/main/java/org/apache/fluo/accumulo/util/ColumnConstants.java
@@ -26,6 +26,7 @@ public class ColumnConstants {
   public static final long TX_DONE_PREFIX = 0x6000000000000000L;
   public static final long WRITE_PREFIX = 0x4000000000000000L;
   public static final long DEL_LOCK_PREFIX = 0x2000000000000000L;
+  public static final long RLOCK_PREFIX = 0x0000000000000000L;
   public static final long LOCK_PREFIX = 0xe000000000000000L;
   public static final long ACK_PREFIX = 0xc000000000000000L;
   public static final long DATA_PREFIX = 0xa000000000000000L;
diff --git a/modules/accumulo/src/main/java/org/apache/fluo/accumulo/util/ColumnConstants.java b/modules/accumulo/src/main/java/org/apache/fluo/accumulo/util/ReadLockUtil.java
similarity index 54%
copy from modules/accumulo/src/main/java/org/apache/fluo/accumulo/util/ColumnConstants.java
copy to modules/accumulo/src/main/java/org/apache/fluo/accumulo/util/ReadLockUtil.java
index 4da1a48..654ebc1 100644
--- a/modules/accumulo/src/main/java/org/apache/fluo/accumulo/util/ColumnConstants.java
+++ b/modules/accumulo/src/main/java/org/apache/fluo/accumulo/util/ReadLockUtil.java
@@ -15,23 +15,30 @@
 
 package org.apache.fluo.accumulo.util;
 
-import org.apache.fluo.api.data.Bytes;
+import com.google.common.base.Preconditions;
+import org.apache.accumulo.core.data.Key;
 
-/**
- * Constants used extract data from columns
- */
-public class ColumnConstants {
+public class ReadLockUtil {
+  private static final long DEL_MASK = 0x0000000000000001L;
+
+  public static boolean isDelete(Key k) {
+    return isDelete(k.getTimestamp());
+  }
+
+  public static boolean isDelete(long ts) {
+    return (ts & DEL_MASK) == DEL_MASK;
+  }
 
-  public static final long PREFIX_MASK = 0xe000000000000000L;
-  public static final long TX_DONE_PREFIX = 0x6000000000000000L;
-  public static final long WRITE_PREFIX = 0x4000000000000000L;
-  public static final long DEL_LOCK_PREFIX = 0x2000000000000000L;
-  public static final long LOCK_PREFIX = 0xe000000000000000L;
-  public static final long ACK_PREFIX = 0xc000000000000000L;
-  public static final long DATA_PREFIX = 0xa000000000000000L;
-  public static final long TIMESTAMP_MASK = 0x1fffffffffffffffL;
-  public static final Bytes NOTIFY_CF = Bytes.of("ntfy");
+  public static long encodeTs(long ts, boolean isDelete) {
+    Preconditions.checkArgument((ts & ColumnConstants.PREFIX_MASK) == 0);
+    return ts << 1 | (isDelete ? 1 : 0);
+  }
 
-  private ColumnConstants() {}
+  public static long decodeTs(Key k) {
+    return decodeTs(k.getTimestamp());
+  }
 
+  public static long decodeTs(long ts) {
+    return ts >> 1;
+  }
 }
diff --git a/modules/accumulo/src/main/java/org/apache/fluo/accumulo/values/DelReadLockValue.java b/modules/accumulo/src/main/java/org/apache/fluo/accumulo/values/DelReadLockValue.java
new file mode 100644
index 0000000..58bf760
--- /dev/null
+++ b/modules/accumulo/src/main/java/org/apache/fluo/accumulo/values/DelReadLockValue.java
@@ -0,0 +1,58 @@
+/*
+ * 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.fluo.accumulo.values;
+
+import org.apache.fluo.accumulo.util.ByteArrayUtil;
+
+public class DelReadLockValue {
+
+  long commitTs = -1;
+  boolean rollback;
+
+  public DelReadLockValue(byte[] value) {
+    this.rollback = isRollback(value);
+    if (!this.rollback) {
+      this.commitTs = getCommitTimestamp(value);
+    }
+  }
+
+
+  public static byte[] encodeRollback() {
+    byte[] ba = new byte[1];
+    ba[0] = (byte) 1;
+    return ba;
+  }
+
+  public static byte[] encodeCommit(long commitTs) {
+    byte[] ba = new byte[9];
+    ba[0] = (byte) 0;
+    ByteArrayUtil.encode(ba, 1, commitTs);
+    return ba;
+  }
+
+  public static boolean isRollback(byte[] data) {
+    return (data[0] & 0x01) == 1;
+  }
+
+  public static long getCommitTimestamp(byte[] data) {
+    return ByteArrayUtil.decodeLong(data, 1);
+  }
+
+  @Override
+  public String toString() {
+    return commitTs + (rollback ? " ABORT" : " COMMIT");
+  }
+}
diff --git a/modules/accumulo/src/main/java/org/apache/fluo/accumulo/values/ReadLockValue.java b/modules/accumulo/src/main/java/org/apache/fluo/accumulo/values/ReadLockValue.java
new file mode 100644
index 0000000..73451ae
--- /dev/null
+++ b/modules/accumulo/src/main/java/org/apache/fluo/accumulo/values/ReadLockValue.java
@@ -0,0 +1,76 @@
+/*
+ * 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.fluo.accumulo.values;
+
+import java.util.List;
+
+import org.apache.fluo.accumulo.util.ByteArrayUtil;
+import org.apache.fluo.accumulo.util.LongUtil;
+import org.apache.fluo.api.data.Bytes;
+import org.apache.fluo.api.data.Column;
+
+import static org.apache.fluo.accumulo.format.FluoFormatter.encNonAscii;
+
+public class ReadLockValue {
+  private final Bytes prow;
+  private final Column pcol;
+  private final Long transactor;
+
+  public ReadLockValue(byte[] enc) {
+    List<Bytes> fields = ByteArrayUtil.split(enc);
+
+    if (fields.size() != 5) {
+      throw new IllegalArgumentException("more fields than expected");
+    }
+
+    this.prow = fields.get(0);
+    this.pcol = new Column(fields.get(1), fields.get(2), fields.get(3));
+    this.transactor = ByteArrayUtil.decodeLong(fields.get(4).toArray());
+  }
+
+  public Bytes getPrimaryRow() {
+    return prow;
+  }
+
+  public Column getPrimaryColumn() {
+    return pcol;
+  }
+
+  public Long getTransactor() {
+    return transactor;
+  }
+
+  public static byte[] encode(Bytes prow, Column pcol, Long transactor) {
+    return ByteArrayUtil.concat(prow, pcol.getFamily(), pcol.getQualifier(), pcol.getVisibility(),
+        Bytes.of(ByteArrayUtil.encode(transactor)));
+  }
+
+  @Override
+  public String toString() {
+    StringBuilder sb = new StringBuilder();
+    encNonAscii(sb, prow);
+    sb.append(" ");
+    encNonAscii(sb, pcol.getFamily());
+    sb.append(" ");
+    encNonAscii(sb, pcol.getQualifier());
+    sb.append(" ");
+    encNonAscii(sb, pcol.getVisibility());
+    sb.append(" ");
+    sb.append(LongUtil.toMaxRadixString(transactor));
+
+    return sb.toString();
+  }
+}
diff --git a/modules/accumulo/src/test/java/org/apache/fluo/accumulo/iterators/CountingIterator.java b/modules/accumulo/src/test/java/org/apache/fluo/accumulo/iterators/CountingIterator.java
new file mode 100644
index 0000000..a2517be
--- /dev/null
+++ b/modules/accumulo/src/test/java/org/apache/fluo/accumulo/iterators/CountingIterator.java
@@ -0,0 +1,63 @@
+/*
+ * 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.fluo.accumulo.iterators;
+
+import java.io.IOException;
+import java.util.Collection;
+import java.util.SortedMap;
+
+import org.apache.accumulo.core.data.ByteSequence;
+import org.apache.accumulo.core.data.Key;
+import org.apache.accumulo.core.data.Range;
+import org.apache.accumulo.core.data.Value;
+import org.apache.accumulo.core.iterators.SortedMapIterator;
+
+public class CountingIterator extends SortedMapIterator {
+
+  public static class Counter {
+    public int nextCalls;
+    public int seeks;
+
+    public void reset() {
+      nextCalls = 0;
+      seeks = 0;
+    }
+  }
+
+  private Counter counter;
+
+  CountingIterator(Counter counter, SortedMap<Key, Value> data) {
+    super(data);
+    this.counter = counter;
+  }
+
+  @Override
+  public void next() throws IOException {
+    super.next();
+    if (counter != null) {
+      counter.nextCalls++;
+    }
+  }
+
+  @Override
+  public void seek(Range range, Collection<ByteSequence> columnFamilies, boolean inclusive)
+      throws IOException {
+    super.seek(range, columnFamilies, inclusive);
+    if (counter != null) {
+      counter.seeks++;
+    }
+  }
+}
diff --git a/modules/accumulo/src/test/java/org/apache/fluo/accumulo/iterators/GarbageCollectionIteratorTest.java b/modules/accumulo/src/test/java/org/apache/fluo/accumulo/iterators/GarbageCollectionIteratorTest.java
index a1600d8..591fdce 100644
--- a/modules/accumulo/src/test/java/org/apache/fluo/accumulo/iterators/GarbageCollectionIteratorTest.java
+++ b/modules/accumulo/src/test/java/org/apache/fluo/accumulo/iterators/GarbageCollectionIteratorTest.java
@@ -597,4 +597,104 @@ public class GarbageCollectionIteratorTest {
 
     Assert.assertEquals(expected, output);
   }
+
+  @Test
+  public void testDeletedReadLocks() {
+
+    TestData input = new TestData();
+
+    input.add("0 f q RLOCK 42", "0 f q");
+    input.add("0 f q DEL_RLOCK 42", "50");
+    input.add("0 f q RLOCK 45", "0 f q");
+    input.add("0 f q DEL_RLOCK 45", "53");
+    input.add("0 f q RLOCK 49", "0 f q");
+    input.add("0 f q DEL_RLOCK 49", "ROLLBACK");
+
+    for (long oldestActiveTs : new long[] {20, 40, 42, 45, 46, 49, 50, 51, 52, 53, 54, 70}) {
+      TestData expected = new TestData();
+      if (oldestActiveTs <= 53) {
+        expected.add("0 f q DEL_RLOCK 45", "53");
+      }
+      if (oldestActiveTs <= 50) {
+        expected.add("0 f q DEL_RLOCK 42", "50");
+      }
+
+      TestData output = new TestData(newGCI(input, oldestActiveTs, true));
+      Assert.assertEquals(expected, output);
+
+
+      expected.add("0 f q DEL_RLOCK 45", "53");
+      expected.add("0 f q DEL_RLOCK 42", "50");
+      expected.add("0 f q DEL_RLOCK 49", "ROLLBACK");
+
+      output = new TestData(newGCI(input, oldestActiveTs, false));
+      Assert.assertEquals(expected, output);
+    }
+
+    input.add("0 f q RLOCK 47", "0 f q");
+    input.add("0 f q RLOCK 40", "0 f q");
+
+    for (long oldestActiveTs : new long[] {20, 40, 42, 45, 46, 47, 48, 49, 50, 51, 52, 53, 54,
+        70}) {
+      TestData expected = new TestData();
+      expected.add("0 f q RLOCK 47", "0 f q");
+      expected.add("0 f q RLOCK 40", "0 f q");
+
+      if (oldestActiveTs <= 53) {
+        expected.add("0 f q DEL_RLOCK 45", "53");
+      }
+      if (oldestActiveTs <= 50) {
+        expected.add("0 f q DEL_RLOCK 42", "50");
+      }
+
+      TestData output = new TestData(newGCI(input, oldestActiveTs, true));
+      Assert.assertEquals(expected, output);
+
+      expected.add("0 f q DEL_RLOCK 45", "53");
+      expected.add("0 f q DEL_RLOCK 42", "50");
+      expected.add("0 f q DEL_RLOCK 49", "ROLLBACK");
+      output = new TestData(newGCI(input, oldestActiveTs, false));
+      Assert.assertEquals(expected, output);
+    }
+  }
+
+  @Test
+  public void testInvalidatedReadLocks() {
+    // a write or delete lock invalidates all read locks, so they can be dropped on partial or full
+    // compactions
+    TestData input = new TestData();
+
+    input.add("0 f q RLOCK 42", "0 f q");
+    input.add("0 f q DEL_RLOCK 42", "50");
+    input.add("0 f q RLOCK 45", "0 f q");
+    input.add("0 f q DEL_RLOCK 45", "53");
+    input.add("0 f q RLOCK 60", "0 f q");
+    input.add("0 f q DEL_RLOCK 60", "70");
+    input.add("1 f q RLOCK 42", "0 f q");
+    input.add("1 f q DEL_RLOCK 42", "50");
+
+    TestData input2 = new TestData(input);
+    input2.add("0 f q WRITE 56", "55");
+    input2.add("0 f q DATA 55", "19");
+
+    TestData expected = new TestData();
+    expected.add("0 f q WRITE 56", "55");
+    expected.add("0 f q DATA 55", "19");
+
+    // invalidation time or oldest active should get rid of all read locks
+    TestData output = new TestData(newGCI(input2, 75, true));
+    Assert.assertEquals(expected, output);
+
+    // only invalidation should get rid of read locks in following three cases
+    expected.add("0 f q DEL_RLOCK 60", "70");
+    expected.add("1 f q DEL_RLOCK 42", "50"); // should not be invalidated, in diff column
+    output = new TestData(newGCI(input2, 75, false));
+    Assert.assertEquals(expected, output);
+
+    output = new TestData(newGCI(input2, 30, true));
+    Assert.assertEquals(expected, output);
+
+    output = new TestData(newGCI(input2, 30, false));
+    Assert.assertEquals(expected, output);
+  }
 }
diff --git a/modules/accumulo/src/test/java/org/apache/fluo/accumulo/iterators/OpenReadLockIteratorTest.java b/modules/accumulo/src/test/java/org/apache/fluo/accumulo/iterators/OpenReadLockIteratorTest.java
new file mode 100644
index 0000000..08b4ee0
--- /dev/null
+++ b/modules/accumulo/src/test/java/org/apache/fluo/accumulo/iterators/OpenReadLockIteratorTest.java
@@ -0,0 +1,90 @@
+/*
+ * 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.fluo.accumulo.iterators;
+
+import java.io.IOException;
+import java.util.Collections;
+
+import org.apache.accumulo.core.data.Key;
+import org.apache.accumulo.core.data.Range;
+import org.apache.accumulo.core.data.Value;
+import org.apache.accumulo.core.iterators.IteratorEnvironment;
+import org.apache.accumulo.core.iterators.IteratorUtil.IteratorScope;
+import org.apache.accumulo.core.iterators.SortedKeyValueIterator;
+import org.apache.accumulo.core.iterators.SortedMapIterator;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class OpenReadLockIteratorTest {
+  OpenReadLockIterator newORLI(TestData input) {
+    OpenReadLockIterator si = new OpenReadLockIterator();
+
+    IteratorEnvironment env = TestIteratorEnv.create(IteratorScope.scan, true);
+
+    try {
+      SortedKeyValueIterator<Key, Value> source = new SortedMapIterator(input.data);
+      si.init(source, Collections.emptyMap(), env);
+    } catch (IOException e) {
+      throw new RuntimeException(e);
+    }
+    return si;
+  }
+
+  @Test
+  public void testBasic() {
+    TestData input = new TestData();
+
+    input.add("0 f q LOCK 11", "1 f q");
+    input.add("0 f q WRITE 16", "11");
+    input.add("0 f q DATA 11", "15");
+    input.add("0 f q DEL_RLOCK 25", "36");
+    input.add("0 f q RLOCK 25", " 0 f q");
+    input.add("0 f q RLOCK 23", " 0 f q");
+    input.add("0 f q DEL_RLOCK 21", "30");
+    input.add("0 f q RLOCK 21", " 0 f q");
+
+    input.add("1 f q1 TX_DONE 16", "11");
+    input.add("1 f q LOCK 11", "1 f q");
+    input.add("1 f q WRITE 16", "11");
+    input.add("1 f q DATA 11", "15");
+    input.add("1 f q RLOCK 25", " 0 f q");
+    input.add("1 f q RLOCK 23", " 0 f q");
+    input.add("1 f q DEL_RLOCK 21", "30");
+    input.add("1 f q RLOCK 21", " 0 f q");
+    input.add("1 f q ACK 11", "");
+
+    input.add("3 f q LOCK 11", "1 f q");
+    input.add("3 f q WRITE 16", "11");
+    input.add("3 f q DATA 11", "15");
+    input.add("3 f q DEL_RLOCK 25", "36");
+    input.add("3 f q RLOCK 25", " 0 f q");
+    input.add("3 f q DEL_RLOCK 23", "33");
+    input.add("3 f q RLOCK 23", " 0 f q");
+    input.add("3 f q DEL_RLOCK 21", "30");
+    input.add("3 f q RLOCK 21", " 0 f q");
+
+    TestData expected = new TestData();
+    expected.add("0 f q RLOCK 23", " 0 f q");
+    expected.add("1 f q RLOCK 25", " 0 f q");
+    expected.add("1 f q RLOCK 23", " 0 f q");
+
+    TestData output = new TestData(newORLI(input));
+    Assert.assertEquals(expected, output);
+
+    output = new TestData(newORLI(input), new Range(), true);
+    Assert.assertEquals(expected, output);
+  }
+}
diff --git a/modules/accumulo/src/test/java/org/apache/fluo/accumulo/iterators/PrewriteIteratorTest.java b/modules/accumulo/src/test/java/org/apache/fluo/accumulo/iterators/PrewriteIteratorTest.java
index 959f224..ad601f5 100644
--- a/modules/accumulo/src/test/java/org/apache/fluo/accumulo/iterators/PrewriteIteratorTest.java
+++ b/modules/accumulo/src/test/java/org/apache/fluo/accumulo/iterators/PrewriteIteratorTest.java
@@ -21,12 +21,29 @@ import org.apache.accumulo.core.client.IteratorSetting;
 import org.apache.accumulo.core.data.Range;
 import org.apache.accumulo.core.iterators.IteratorEnvironment;
 import org.apache.accumulo.core.iterators.IteratorUtil.IteratorScope;
-import org.apache.accumulo.core.iterators.SortedMapIterator;
 import org.junit.Assert;
 import org.junit.Test;
 
 public class PrewriteIteratorTest {
 
+  PrewriteIterator newPI(TestData input, long snapTime, boolean readlock) {
+    PrewriteIterator ni = new PrewriteIterator();
+
+    IteratorEnvironment env = TestIteratorEnv.create(IteratorScope.scan, false);
+
+    try {
+      IteratorSetting cfg = new IteratorSetting(10, PrewriteIterator.class);
+      PrewriteIterator.setSnaptime(cfg, snapTime);
+      if (readlock) {
+        PrewriteIterator.setReadlock(cfg);
+      }
+      ni.init(input.getIterator(), cfg.getOptions(), env);
+    } catch (IOException e) {
+      throw new RuntimeException(e);
+    }
+    return ni;
+  }
+
   PrewriteIterator newPI(TestData input, long snapTime) {
     PrewriteIterator ni = new PrewriteIterator();
 
@@ -35,7 +52,7 @@ public class PrewriteIteratorTest {
     try {
       IteratorSetting cfg = new IteratorSetting(10, PrewriteIterator.class);
       PrewriteIterator.setSnaptime(cfg, snapTime);
-      ni.init(new SortedMapIterator(input.data), cfg.getOptions(), env);
+      ni.init(input.getIterator(), cfg.getOptions(), env);
     } catch (IOException e) {
       throw new RuntimeException(e);
     }
@@ -51,7 +68,7 @@ public class PrewriteIteratorTest {
       IteratorSetting cfg = new IteratorSetting(10, PrewriteIterator.class);
       PrewriteIterator.setSnaptime(cfg, snapTime);
       PrewriteIterator.enableAckCheck(cfg, ntfyTime);
-      ni.init(new SortedMapIterator(input.data), cfg.getOptions(), env);
+      ni.init(input.getIterator(), cfg.getOptions(), env);
     } catch (IOException e) {
       throw new RuntimeException(e);
     }
@@ -238,4 +255,219 @@ public class PrewriteIteratorTest {
     output = new TestData(newPI(input, 117), Range.exact("0", "f", "q"));
     Assert.assertEquals(0, output.data.size());
   }
+
+  @Test
+  public void testReadLockPreventsWriteLock() {
+
+    TestData[] initialInputs = new TestData[4];
+
+    initialInputs[0] = new TestData();
+
+    initialInputs[1] = new TestData();
+    initialInputs[1].add("0 f q WRITE 10", "5");
+    initialInputs[1].add("0 f q LOCK 5", "0 f q");
+    initialInputs[1].add("0 f q DATA 5", "15");
+
+    initialInputs[2] = new TestData();
+    initialInputs[2].add("0 f q WRITE 10", "5");
+    initialInputs[2].add("0 f q DATA 5", "15");
+
+    initialInputs[3] = new TestData();
+    initialInputs[3].add("0 f q DEL_LOCK 5", "ABORT");
+
+    for (TestData input : initialInputs) {
+
+      input.add("0 f q RLOCK 42", "0 f q");
+
+      TestData expected = new TestData();
+      expected.add("0 f q RLOCK 42", "0 f q");
+
+      for (int ts : new int[] {40, 45}) {
+        TestData output = new TestData(newPI(input, ts), Range.exact("0", "f", "q"));
+        Assert.assertEquals(expected, output);
+      }
+
+      input.add("0 f q DEL_RLOCK 42", "50");
+
+      expected = new TestData();
+      expected.add("0 f q DEL_RLOCK 42", "50");
+
+      for (int ts : new int[] {40, 45}) {
+        TestData output = new TestData(newPI(input, ts), Range.exact("0", "f", "q"));
+        Assert.assertEquals(expected, output);
+      }
+
+      TestData output = new TestData(newPI(input, 117), Range.exact("0", "f", "q"));
+      Assert.assertEquals(0, output.data.size());
+
+      input.add("0 f q RLOCK 30", "0 f q");
+
+      expected = new TestData();
+      expected.add("0 f q RLOCK 30", "0 f q");
+
+      output = new TestData(newPI(input, 117), Range.exact("0", "f", "q"));
+      Assert.assertEquals(expected, output);
+
+      input.add("0 f q DEL_RLOCK 30", "60");
+
+      output = new TestData(newPI(input, 117), Range.exact("0", "f", "q"));
+      Assert.assertEquals(0, output.data.size());
+
+      output = new TestData(newPI(input, 55), Range.exact("0", "f", "q"));
+      expected = new TestData();
+      expected.add("0 f q DEL_RLOCK 30", "60");
+      Assert.assertEquals(expected, output);
+
+      expected = new TestData();
+      expected.add("0 f q DEL_RLOCK 42", "50");
+
+      for (int ts : new int[] {20, 40, 45}) {
+        output = new TestData(newPI(input, ts), Range.exact("0", "f", "q"));
+        Assert.assertEquals(expected, output);
+      }
+
+    }
+  }
+
+  @Test
+  public void testOnlyDelReadLocks() {
+    //Garbage collection iter may drop read locks and leave del_read_lock entries.  Ensure this case works as expected.
+    TestData input = new TestData();
+    input.add("0 f q DEL_RLOCK 42", "50");
+    input.add("0 f q DEL_RLOCK 39", "44");
+
+    input.add("0 f q WRITE 10", "5");
+    input.add("0 f q LOCK 5", "0 f q");
+    input.add("0 f q DATA 5", "15");
+
+    for (long startTs : new long[] {31, 40, 47}) {
+      TestData expected = new TestData();
+      expected.add("0 f q DEL_RLOCK 42", "50");
+
+      TestData output = new TestData(newPI(input, startTs), Range.exact("0", "f", "q"));
+
+      Assert.assertEquals(expected, output);
+    }
+
+    TestData output = new TestData(newPI(input, 55), Range.exact("0", "f", "q"));
+    Assert.assertEquals(0, output.data.size());
+  }
+
+  @Test
+  public void testWriteLockPreventsReadLock() {
+    for (int i = 0; i < 2; i++) {
+      TestData input = new TestData();
+
+      if (i == 1) {
+        // this should have no impact
+        input.add("0 f q RLOCK 13", "0 f q");
+      }
+
+      input.add("0 f q LOCK 5", "0 f q");
+      input.add("0 f q DATA 5", "15");
+
+      TestData output = new TestData(newPI(input, 117, true), Range.exact("0", "f", "q"));
+      TestData expected = new TestData();
+      expected.add("0 f q LOCK 5", "0 f q");
+      Assert.assertEquals(expected, output);
+
+      input.add("0 f q WRITE 10", "5");
+      output = new TestData(newPI(input, 117, true), Range.exact("0", "f", "q"));
+      Assert.assertEquals(0, output.data.size());
+
+      output = new TestData(newPI(input, 11, true), Range.exact("0", "f", "q"));
+      Assert.assertEquals(0, output.data.size());
+
+      output = new TestData(newPI(input, 7, true), Range.exact("0", "f", "q"));
+      expected = new TestData();
+      expected.add("0 f q WRITE 10", "5");
+      Assert.assertEquals(expected, output);
+
+      output = new TestData(newPI(input, 4, true), Range.exact("0", "f", "q"));
+      Assert.assertEquals(expected, output);
+    }
+  }
+
+  @Test
+  public void testManyReadLocks() {
+    TestData input = new TestData();
+
+    input.add("0 f q WRITE 10", "5");
+    input.add("0 f q DATA 5", "15");
+
+    for (int i = 10; i < 2010; i += 2) {
+      input.add("0 f q DEL_RLOCK " + i, "" + (i + 1));
+      input.add("0 f q RLOCK " + i, "0 f q");
+    }
+
+    TestData output = new TestData(newPI(input, 3000, false), Range.exact("0", "f", "q"));
+    // scans all read locks looking for an active lock
+    Assert.assertEquals(2001, input.counter.nextCalls);
+    Assert.assertEquals(0, output.data.size());
+
+    // read locks do not need to scan for other read locks... this checks that read locks are
+    // skipped
+    input.counter.reset();
+    output = new TestData(newPI(input, 3000, true), Range.exact("0", "f", "q"));
+    // read locks should not scan everything looking for a read lock
+    Assert.assertEquals(12, input.counter.nextCalls); // skipping will read 11 before seeking
+    Assert.assertEquals(0, output.data.size());
+
+    // This write invalidates the 2000 read locks, so should skip read locks
+    input.add("0 f q WRITE 2500", "2490");
+    input.add("0 f q DATA 2490", "16");
+    input.counter.reset();
+    output = new TestData(newPI(input, 3000, false), Range.exact("0", "f", "q"));
+    Assert.assertEquals(13, input.counter.nextCalls);
+    Assert.assertEquals(0, output.data.size());
+  }
+
+  @Test
+  public void testGarbageCollectedReadLocks() {
+    // After a partial compaction the garbage collection iterator will drop read lock entries, but
+    // not del_read_locks entries. This test ensures that prewrite works w/ this.
+
+    TestData input = new TestData();
+
+    input.add("0 f q DEL_RLOCK 42", "50");
+    input.add("0 f q DEL_RLOCK 44", "51");
+
+
+    TestData output = new TestData(newPI(input, 47), Range.exact("0", "f", "q"));
+    TestData expected = new TestData();
+
+    expected.add("0 f q DEL_RLOCK 44", "51");
+
+    Assert.assertEquals(expected, output);
+
+    output = new TestData(newPI(input, 43), Range.exact("0", "f", "q"));
+    Assert.assertEquals(expected, output);
+  }
+
+  @Test
+  public void testAbortedReadLock() {
+    // A read lock that was aborted or rolledback should not prevent a write lock
+    TestData input = new TestData();
+
+    input.add("0 f q DEL_RLOCK 55", "ABORT");
+    input.add("0 f q DEL_RLOCK 42", "50");
+
+    for (int i = 0; i < 2; i++) {
+      for (long startTs : new long[] {31, 40, 47}) {
+        TestData expected = new TestData();
+        expected.add("0 f q DEL_RLOCK 42", "50");
+
+        TestData output = new TestData(newPI(input, startTs), Range.exact("0", "f", "q"));
+
+        Assert.assertEquals(expected, output);
+      }
+
+      for (long startTs : new long[] {51, 55, 56}) {
+        TestData output = new TestData(newPI(input, startTs), Range.exact("0", "f", "q"));
+        Assert.assertEquals(0, output.data.size());
+      }
+      //add this for 2nd iteration, should ignore because delete rolls back
+      input.add("0 f q RLOCK 55", "0 f q");
+    }
+  }
 }
diff --git a/modules/accumulo/src/test/java/org/apache/fluo/accumulo/iterators/SnapshotIteratorTest.java b/modules/accumulo/src/test/java/org/apache/fluo/accumulo/iterators/SnapshotIteratorTest.java
index 4605923..4e0c27d 100644
--- a/modules/accumulo/src/test/java/org/apache/fluo/accumulo/iterators/SnapshotIteratorTest.java
+++ b/modules/accumulo/src/test/java/org/apache/fluo/accumulo/iterators/SnapshotIteratorTest.java
@@ -31,10 +31,11 @@ import org.junit.Assert;
 import org.junit.Test;
 
 public class SnapshotIteratorTest {
-  SnapshotIterator newSI(TestData input, long startTs) {
+  SnapshotIterator newSI(TestData input, long startTs, boolean returnReadLocks) {
     SnapshotIterator si = new SnapshotIterator();
 
     Map<String, String> options = new HashMap<>();
+    options.put(SnapshotIterator.RETURN_READLOCK_PRESENT_OPT, returnReadLocks + "");
     options.put(SnapshotIterator.TIMESTAMP_OPT, startTs + "");
 
     IteratorEnvironment env = TestIteratorEnv.create(IteratorScope.scan, true);
@@ -48,6 +49,10 @@ public class SnapshotIteratorTest {
     return si;
   }
 
+  SnapshotIterator newSI(TestData input, long startTs) {
+    return newSI(input, startTs, true);
+  }
+
   @Test
   public void testBasic() {
     TestData input = new TestData();
@@ -61,13 +66,11 @@ public class SnapshotIteratorTest {
     TestData output = new TestData(newSI(input, 6));
     Assert.assertEquals(0, output.data.size());
 
-    output = new TestData(newSI(input, 11));
     TestData expected = new TestData().add("0 f q DATA 9", "14");
-    Assert.assertEquals(expected, output);
+    checkInput(input, expected, 11);
 
-    output = new TestData(newSI(input, 17));
     expected = new TestData().add("0 f q DATA 11", "15");
-    Assert.assertEquals(expected, output);
+    checkInput(input, expected, 17);
   }
 
   @Test
@@ -85,13 +88,11 @@ public class SnapshotIteratorTest {
     output = new TestData(newSI(input, 15));
     Assert.assertEquals(0, output.data.size());
 
-    output = new TestData(newSI(input, 17));
     TestData expected = new TestData().add("0 f q DATA 11", "15");
-    Assert.assertEquals(expected, output);
+    checkInput(input, expected, 17);
 
-    output = new TestData(newSI(input, 22));
     expected = new TestData().add("0 f q LOCK 21", "1 f q");
-    Assert.assertEquals(expected, output);
+    checkInput(input, expected, 22);
   }
 
   @Test
@@ -115,13 +116,11 @@ public class SnapshotIteratorTest {
     output = new TestData(newSI(input, 15));
     Assert.assertEquals(0, output.data.size());
 
-    output = new TestData(newSI(input, 17));
     TestData expected = new TestData().add("0 f q DATA 11", "15");
-    Assert.assertEquals(expected, output);
+    checkInput(input, expected, 17);
 
-    output = new TestData(newSI(input, 23));
     expected = new TestData().add("0 f q DATA 11", "15");
-    Assert.assertEquals(expected, output);
+    checkInput(input, expected, 23);
 
 
     // test case where there is newer lock thats not invalidated by DEL_LOCK
@@ -137,17 +136,14 @@ public class SnapshotIteratorTest {
     output = new TestData(newSI(input, 6));
     Assert.assertEquals(0, output.data.size());
 
-    output = new TestData(newSI(input, 17));
     expected = new TestData().add("0 f q DATA 11", "15");
-    Assert.assertEquals(expected, output);
+    checkInput(input, expected, 17);
 
-    output = new TestData(newSI(input, 19));
     expected = new TestData().add("0 f q DATA 11", "15");
-    Assert.assertEquals(expected, output);
+    checkInput(input, expected, 19);
 
-    output = new TestData(newSI(input, 23));
     expected = new TestData().add("0 f q LOCK 21", "1 f q");
-    Assert.assertEquals(expected, output);
+    checkInput(input, expected, 23);
   }
 
   @Test
@@ -253,7 +249,54 @@ public class SnapshotIteratorTest {
     for (Range range : ranges) {
       checkManyColumnData(input, numToWrite, range);
     }
+  }
 
+  @Test
+  public void testReadLock() {
+    TestData input = new TestData();
+
+    input.add("0 f q WRITE 16", "11 DELETE");
+    input.add("0 f q DATA 11", "15");
+    input.add("0 f q DEL_RLOCK 5", "6");
+    input.add("0 f q RLOCK 5", " 0 f q");
+
+    input.add("1 f q WRITE 16", "11");
+    input.add("1 f q DATA 11", "15");
+
+    input.add("2 f q WRITE 16", "11");
+    input.add("2 f q DATA 11", "17");
+    input.add("2 f q DEL_RLOCK 5", "6");
+    input.add("2 f q RLOCK 5", " 0 f q");
+
+
+    TestData expected = new TestData();
+    expected.add("0 f q DATA 11", "15");
+    expected.add("1 f q DATA 11", "15");
+    expected.add("2 f q DATA 11", "17");
+
+
+    checkInput(input, expected, 20, false);
+
+    expected.add("0 f q DEL_RLOCK 5", "6");
+    expected.add("2 f q DEL_RLOCK 5", "6");
+
+    checkInput(input, expected, 20);
+
+  }
+
+  private void checkInput(TestData input, TestData expected, long startTs) {
+    checkInput(input, expected, startTs, true);
+  }
+
+  private void checkInput(TestData input, TestData expected, long startTs,
+      boolean returnReadLocks) {
+    // run test with a single seek followed by many next calls
+    TestData output = new TestData(newSI(input, startTs, returnReadLocks), new Range());
+    Assert.assertEquals(expected, output);
+
+    // run test reseeking after each key
+    output = new TestData(newSI(input, startTs, returnReadLocks), new Range(), true);
+    Assert.assertEquals(expected, output);
   }
 
   private void checkManyColumnData(TestData input, int numToWrite, Range range) throws IOException {
diff --git a/modules/accumulo/src/test/java/org/apache/fluo/accumulo/iterators/TestData.java b/modules/accumulo/src/test/java/org/apache/fluo/accumulo/iterators/TestData.java
index 2ea42b3..262df15 100644
--- a/modules/accumulo/src/test/java/org/apache/fluo/accumulo/iterators/TestData.java
+++ b/modules/accumulo/src/test/java/org/apache/fluo/accumulo/iterators/TestData.java
@@ -28,8 +28,11 @@ import org.apache.accumulo.core.data.Range;
 import org.apache.accumulo.core.data.Value;
 import org.apache.accumulo.core.iterators.SortedKeyValueIterator;
 import org.apache.fluo.accumulo.format.FluoFormatter;
+import org.apache.fluo.accumulo.iterators.CountingIterator.Counter;
 import org.apache.fluo.accumulo.util.ColumnConstants;
+import org.apache.fluo.accumulo.util.ReadLockUtil;
 import org.apache.fluo.accumulo.values.DelLockValue;
+import org.apache.fluo.accumulo.values.DelReadLockValue;
 import org.apache.fluo.accumulo.values.LockValue;
 import org.apache.fluo.accumulo.values.WriteValue;
 import org.apache.fluo.api.data.Bytes;
@@ -37,6 +40,7 @@ import org.apache.fluo.api.data.Column;
 
 public class TestData {
   TreeMap<Key, Value> data = new TreeMap<>();
+  Counter counter = new Counter();
 
   TestData() {}
 
@@ -44,19 +48,29 @@ public class TestData {
     data.putAll(td.data);
   }
 
-  TestData(SortedKeyValueIterator<Key, Value> iter, Range range) {
+  TestData(SortedKeyValueIterator<Key, Value> iter, Range range, boolean reseek) {
     try {
       iter.seek(range, new HashSet<ByteSequence>(), false);
 
       while (iter.hasTop()) {
         data.put(iter.getTopKey(), iter.getTopValue());
-        iter.next();
+        if (reseek) {
+          iter.seek(
+              new Range(iter.getTopKey(), false, range.getEndKey(), range.isEndKeyInclusive()),
+              new HashSet<ByteSequence>(), false);
+        } else {
+          iter.next();
+        }
       }
     } catch (IOException e) {
       throw new RuntimeException(e);
     }
   }
 
+  TestData(SortedKeyValueIterator<Key, Value> iter, Range range) {
+    this(iter, range, false);
+  }
+
   TestData(SortedKeyValueIterator<Key, Value> iter) {
     this(iter, new Range());
   }
@@ -116,6 +130,21 @@ public class TestData {
           val = DelLockValue.encodeCommit(commitTs, value.contains("PRIMARY"));
         }
         break;
+      case "RLOCK":
+        ts = ReadLockUtil.encodeTs(ts, false);
+        ts |= ColumnConstants.RLOCK_PREFIX;
+        break;
+      case "DEL_RLOCK":
+        ts = ReadLockUtil.encodeTs(ts, true);
+        ts |= ColumnConstants.RLOCK_PREFIX;
+
+        if (value.contains("ROLLBACK") || value.contains("ABORT")) {
+          val = DelReadLockValue.encodeRollback();
+        } else {
+          long commitTs = Long.parseLong(value.split("\\s+")[0]);
+          val = DelReadLockValue.encodeCommit(commitTs);
+        }
+        break;
       case "ntfy":
         break;
       default:
@@ -163,4 +192,8 @@ public class TestData {
   public int hashCode() {
     return Objects.hashCode(data);
   }
+
+  public SortedKeyValueIterator<Key, Value> getIterator() {
+    return new CountingIterator(counter, data);
+  }
 }
diff --git a/modules/api/src/main/java/org/apache/fluo/api/client/AbstractSnapshotBase.java b/modules/api/src/main/java/org/apache/fluo/api/client/AbstractSnapshotBase.java
index 69780b2..1ba7f9b 100644
--- a/modules/api/src/main/java/org/apache/fluo/api/client/AbstractSnapshotBase.java
+++ b/modules/api/src/main/java/org/apache/fluo/api/client/AbstractSnapshotBase.java
@@ -49,6 +49,15 @@ public abstract class AbstractSnapshotBase implements SnapshotBase {
    */
   private Map<String, Bytes> s2bCache = new WeakHashMap<String, Bytes>();
 
+  public AbstractSnapshotBase() {}
+
+  /**
+   * @since 1.2.0
+   */
+  protected AbstractSnapshotBase(AbstractSnapshotBase other) {
+    this.s2bCache = other.s2bCache;
+  }
+
   Bytes s2bConv(CharSequence cs) {
     Objects.requireNonNull(cs);
     if (cs instanceof String) {
@@ -64,6 +73,7 @@ public abstract class AbstractSnapshotBase implements SnapshotBase {
     }
   }
 
+  @Override
   public Bytes get(Bytes row, Column column, Bytes defaultValue) {
     Bytes ret = get(row, column);
     if (ret == null) {
@@ -73,19 +83,23 @@ public abstract class AbstractSnapshotBase implements SnapshotBase {
     return ret;
   }
 
+  @Override
   public Map<Column, Bytes> get(Bytes row, Column... columns) {
     return get(row, ImmutableSet.copyOf(columns));
   }
 
+  @Override
   public Map<Bytes, Map<Column, Bytes>> get(Collection<Bytes> rows, Column... columns) {
     return get(rows, ImmutableSet.copyOf(columns));
   }
 
+  @Override
   public Map<RowColumn, String> gets(Collection<RowColumn> rowColumns) {
     Map<RowColumn, Bytes> bytesMap = get(rowColumns);
     return Maps.transformValues(bytesMap, b -> b.toString());
   }
 
+  @Override
   public Map<String, Map<Column, String>> gets(Collection<? extends CharSequence> rows,
       Set<Column> columns) {
     Map<Bytes, Map<Column, Bytes>> rcvs = get(Collections2.transform(rows, this::s2bConv), columns);
@@ -97,11 +111,13 @@ public abstract class AbstractSnapshotBase implements SnapshotBase {
     return ret;
   }
 
+  @Override
   public Map<String, Map<Column, String>> gets(Collection<? extends CharSequence> rows,
       Column... columns) {
     return gets(rows, ImmutableSet.copyOf(columns));
   }
 
+  @Override
   public String gets(CharSequence row, Column column) {
     Bytes val = get(s2bConv(row), column);
     if (val == null) {
@@ -110,6 +126,7 @@ public abstract class AbstractSnapshotBase implements SnapshotBase {
     return val.toString();
   }
 
+  @Override
   public String gets(CharSequence row, Column column, String defaultValue) {
     Bytes val = get(s2bConv(row), column);
     if (val == null) {
@@ -119,11 +136,13 @@ public abstract class AbstractSnapshotBase implements SnapshotBase {
     return val.toString();
   }
 
+  @Override
   public Map<Column, String> gets(CharSequence row, Set<Column> columns) {
     Map<Column, Bytes> values = get(s2bConv(row), columns);
     return Maps.transformValues(values, b -> b.toString());
   }
 
+  @Override
   public Map<Column, String> gets(CharSequence row, Column... columns) {
     return gets(row, ImmutableSet.copyOf(columns));
   }
diff --git a/modules/api/src/main/java/org/apache/fluo/api/client/AbstractTransactionBase.java b/modules/api/src/main/java/org/apache/fluo/api/client/AbstractTransactionBase.java
index 8e44590..731aaeb 100644
--- a/modules/api/src/main/java/org/apache/fluo/api/client/AbstractTransactionBase.java
+++ b/modules/api/src/main/java/org/apache/fluo/api/client/AbstractTransactionBase.java
@@ -27,14 +27,17 @@ import org.apache.fluo.api.exceptions.AlreadySetException;
 public abstract class AbstractTransactionBase extends AbstractSnapshotBase
     implements TransactionBase {
 
+  @Override
   public void delete(CharSequence row, Column col) {
     delete(s2bConv(row), col);
   }
 
+  @Override
   public void set(CharSequence row, Column col, CharSequence value) throws AlreadySetException {
     set(s2bConv(row), col, Bytes.of(value));
   }
 
+  @Override
   public void setWeakNotification(CharSequence row, Column col) {
     setWeakNotification(s2bConv(row), col);
   }
diff --git a/modules/api/src/main/java/org/apache/fluo/api/client/SnapshotBase.java b/modules/api/src/main/java/org/apache/fluo/api/client/SnapshotBase.java
index 51c29b9..6f4d1ff 100644
--- a/modules/api/src/main/java/org/apache/fluo/api/client/SnapshotBase.java
+++ b/modules/api/src/main/java/org/apache/fluo/api/client/SnapshotBase.java
@@ -114,6 +114,7 @@ public interface SnapshotBase {
    * </pre>
    *
    * @return A scanner builder.
+   * @see TransactionBase#withReadLock()
    */
 
   ScannerBuilder scanner();
diff --git a/modules/api/src/main/java/org/apache/fluo/api/client/TransactionBase.java b/modules/api/src/main/java/org/apache/fluo/api/client/TransactionBase.java
index 06303b6..bb53da5 100644
--- a/modules/api/src/main/java/org/apache/fluo/api/client/TransactionBase.java
+++ b/modules/api/src/main/java/org/apache/fluo/api/client/TransactionBase.java
@@ -4,9 +4,9 @@
  * 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
@@ -60,4 +60,26 @@ public interface TransactionBase extends SnapshotBase {
    * encoded using UTF-8.
    */
   void setWeakNotification(CharSequence row, Column col);
+
+  /**
+   * Normally when a Fluo transaction reads data and does not write to it, it will not collide with
+   * other transactions making concurrent writes.  When this method is called, all reads will
+   * acquire a read lock. These read locks cause collisions with transactions doing concurrent
+   * writes. However, multiple transactions can get concurrent read locks on the same row+col
+   * without colliding.
+   *
+   * <p>
+   * Scanning with read locks is not supported. Attempting to call {@code withReadLock().scanner()}
+   * will throw an {@link UnsupportedOperationException}. This is because there are an infinite
+   * amount of keys within a range and read locks can not be obtained on them all.
+   *
+   * <p>
+   * A transaction that only acquires read locks will do nothing at commit time. In this case no
+   * read locks are actually written and no collisions will ever occur.
+   *
+   * @since 1.2.0
+   */
+  default SnapshotBase withReadLock() {
+    throw new UnsupportedOperationException("Read locks not supported by this implementation");
+  }
 }
diff --git a/modules/command/src/test/java/org/apache/fluo/command/ScanTest.java b/modules/command/src/test/java/org/apache/fluo/command/ScanTest.java
index 5998e89..20377a9 100644
--- a/modules/command/src/test/java/org/apache/fluo/command/ScanTest.java
+++ b/modules/command/src/test/java/org/apache/fluo/command/ScanTest.java
@@ -34,7 +34,7 @@ public class ScanTest {
     JCommander jcommand = new JCommander(options);
     jcommand.parse(args.split(" "));
     ScanUtil.ScanOpts opts = options.getScanOpts();
-    return new SnapshotScanner.Opts(ScanUtil.getSpan(opts), ScanUtil.getColumns(opts));
+    return new SnapshotScanner.Opts(ScanUtil.getSpan(opts), ScanUtil.getColumns(opts), false);
   }
 
   @Test
diff --git a/modules/core/src/main/java/org/apache/fluo/core/impl/LockResolver.java b/modules/core/src/main/java/org/apache/fluo/core/impl/LockResolver.java
index f3068e9..4172386 100644
--- a/modules/core/src/main/java/org/apache/fluo/core/impl/LockResolver.java
+++ b/modules/core/src/main/java/org/apache/fluo/core/impl/LockResolver.java
@@ -24,24 +24,33 @@ import java.util.Set;
 
 import org.apache.accumulo.core.client.AccumuloException;
 import org.apache.accumulo.core.client.AccumuloSecurityException;
+import org.apache.accumulo.core.client.BatchScanner;
 import org.apache.accumulo.core.client.ConditionalWriter;
 import org.apache.accumulo.core.client.ConditionalWriter.Status;
 import org.apache.accumulo.core.client.IteratorSetting;
 import org.apache.accumulo.core.data.ByteSequence;
 import org.apache.accumulo.core.data.Key;
 import org.apache.accumulo.core.data.Mutation;
+import org.apache.accumulo.core.data.Range;
 import org.apache.accumulo.core.data.Value;
+import org.apache.fluo.accumulo.iterators.OpenReadLockIterator;
 import org.apache.fluo.accumulo.iterators.PrewriteIterator;
 import org.apache.fluo.accumulo.util.ColumnConstants;
+import org.apache.fluo.accumulo.util.ReadLockUtil;
 import org.apache.fluo.accumulo.values.DelLockValue;
+import org.apache.fluo.accumulo.values.DelReadLockValue;
 import org.apache.fluo.accumulo.values.LockValue;
+import org.apache.fluo.accumulo.values.ReadLockValue;
+import org.apache.fluo.api.data.Bytes;
 import org.apache.fluo.api.data.Column;
+import org.apache.fluo.api.data.RowColumn;
 import org.apache.fluo.core.util.ByteUtil;
 import org.apache.fluo.core.util.ColumnUtil;
 import org.apache.fluo.core.util.ConditionalFlutation;
 import org.apache.fluo.core.util.FluoCondition;
 import org.apache.fluo.core.util.SpanUtil;
 
+import static org.apache.fluo.accumulo.util.ColumnConstants.PREFIX_MASK;
 import static org.apache.fluo.api.observer.Observer.NotificationType.STRONG;
 
 /**
@@ -52,38 +61,61 @@ import static org.apache.fluo.api.observer.Observer.NotificationType.STRONG;
 
 public class LockResolver {
 
-  private static Map<PrimaryRowColumn, List<Entry<Key, Value>>> groupLocksByPrimary(
-      List<Entry<Key, Value>> locks) {
-    Map<PrimaryRowColumn, List<Entry<Key, Value>>> groupedLocks = new HashMap<>();
+  private static Map<PrimaryRowColumn, List<LockInfo>> groupLocksByPrimary(List<LockInfo> locks) {
+    Map<PrimaryRowColumn, List<LockInfo>> groupedLocks = new HashMap<>();
     Map<PrimaryRowColumn, Long> transactorIds = new HashMap<>();
 
-    for (Entry<Key, Value> lock : locks) {
-      LockValue lockVal = new LockValue(lock.getValue().get());
-      PrimaryRowColumn prc =
-          new PrimaryRowColumn(lockVal.getPrimaryRow(), lockVal.getPrimaryColumn(),
-              lock.getKey().getTimestamp() & ColumnConstants.TIMESTAMP_MASK);
+    for (LockInfo lockInfo : locks) {
 
-      List<Entry<Key, Value>> lockList = groupedLocks.get(prc);
-      if (lockList == null) {
-        lockList = new ArrayList<>();
-        groupedLocks.put(prc, lockList);
-      }
+      PrimaryRowColumn prc = new PrimaryRowColumn(lockInfo.prow, lockInfo.pcol, lockInfo.lockTs);
+
+      List<LockInfo> lockList = groupedLocks.computeIfAbsent(prc, k -> new ArrayList<>());
 
       Long trid = transactorIds.get(prc);
       if (trid == null) {
-        transactorIds.put(prc, lockVal.getTransactor());
-      } else if (!trid.equals(lockVal.getTransactor())) {
+        transactorIds.put(prc, lockInfo.transactorId);
+      } else if (!trid.equals(lockInfo.transactorId)) {
         // sanity check.. its assumed that all locks w/ the same PrimaryRowColumn should have the
         // same transactor id as well
-        throw new IllegalStateException("transactor ids not equals " + prc + " " + lock.getKey()
-            + " " + trid + " " + lockVal.getTransactor());
+        throw new IllegalStateException("transactor ids not equals " + prc + " "
+            + lockInfo.entry.getKey() + " " + trid + " " + lockInfo.transactorId);
       }
 
-      lockList.add(lock);
+      lockList.add(lockInfo);
     }
 
     return groupedLocks;
+  }
+
 
+  private static class LockInfo {
+
+    final Bytes prow;
+    final Column pcol;
+    final Long transactorId;
+    final long lockTs;
+    final boolean isReadLock;
+    final Entry<Key, Value> entry;
+
+    public LockInfo(Entry<Key, Value> kve) {
+      long rawTs = kve.getKey().getTimestamp();
+      this.entry = kve;
+      if ((rawTs & ColumnConstants.PREFIX_MASK) == ColumnConstants.RLOCK_PREFIX) {
+        this.lockTs = ReadLockUtil.decodeTs(rawTs);
+        ReadLockValue rlv = new ReadLockValue(kve.getValue().get());
+        this.prow = rlv.getPrimaryRow();
+        this.pcol = rlv.getPrimaryColumn();
+        this.transactorId = rlv.getTransactor();
+        this.isReadLock = true;
+      } else {
+        this.lockTs = kve.getKey().getTimestamp() & ColumnConstants.TIMESTAMP_MASK;
+        LockValue lv = new LockValue(kve.getValue().get());
+        this.prow = lv.getPrimaryRow();
+        this.pcol = lv.getPrimaryColumn();
+        this.transactorId = lv.getTransactor();
+        this.isReadLock = false;
+      }
+    }
   }
 
   /**
@@ -99,7 +131,7 @@ public class LockResolver {
    * @return true if all locks passed in were resolved (rolled forward or back)
    */
   static boolean resolveLocks(Environment env, long startTs, TxStats stats,
-      List<Entry<Key, Value>> locks, long startTime) {
+      List<Entry<Key, Value>> locksKVs, long startTime) {
     // check if transactor is still alive
 
     int numResolved = 0;
@@ -110,7 +142,10 @@ public class LockResolver {
 
     TransactorCache transactorCache = env.getSharedResources().getTransactorCache();
 
-    List<Entry<Key, Value>> locksToRecover;
+    List<LockInfo> locks = new ArrayList<>();
+    locksKVs.forEach(e -> locks.add(new LockInfo(e)));
+
+    List<LockInfo> locksToRecover;
     if (System.currentTimeMillis() - startTime > env.getConfiguration()
         .getTransactionRollbackTime()) {
       locksToRecover = locks;
@@ -118,38 +153,32 @@ public class LockResolver {
       timedOut = true;
     } else {
       locksToRecover = new ArrayList<>(locks.size());
-      for (Entry<Key, Value> entry : locks) {
-
-        Long transactorId = new LockValue(entry.getValue().get()).getTransactor();
-        long lockTs = entry.getKey().getTimestamp() & ColumnConstants.TIMESTAMP_MASK;
-
-        if (transactorCache.checkTimedout(transactorId, lockTs)) {
-          locksToRecover.add(entry);
+      for (LockInfo lockInfo : locks) {
+        if (transactorCache.checkTimedout(lockInfo.transactorId, lockInfo.lockTs)) {
+          locksToRecover.add(lockInfo);
           stats.incrementTimedOutLocks();
-        } else if (!transactorCache.checkExists(transactorId)) {
-          locksToRecover.add(entry);
+        } else if (!transactorCache.checkExists(lockInfo.transactorId)) {
+          locksToRecover.add(lockInfo);
           stats.incrementDeadLocks();
         }
       }
     }
 
-    Map<PrimaryRowColumn, List<Entry<Key, Value>>> groupedLocks =
-        groupLocksByPrimary(locksToRecover);
+    Map<PrimaryRowColumn, List<LockInfo>> groupedLocks = groupLocksByPrimary(locksToRecover);
 
     if (timedOut) {
-      Set<Entry<PrimaryRowColumn, List<Entry<Key, Value>>>> es = groupedLocks.entrySet();
+      Set<Entry<PrimaryRowColumn, List<LockInfo>>> es = groupedLocks.entrySet();
 
-      for (Entry<PrimaryRowColumn, List<Entry<Key, Value>>> entry : es) {
+      for (Entry<PrimaryRowColumn, List<LockInfo>> entry : es) {
         long lockTs = entry.getKey().startTs;
-        Long transactorId = new LockValue(entry.getValue().get(0).getValue().get()).getTransactor();
+        Long transactorId = entry.getValue().get(0).transactorId;
         transactorCache.addTimedoutTransactor(transactorId, lockTs, startTime);
       }
     }
 
     TxInfoCache txiCache = env.getSharedResources().getTxInfoCache();
-
-    Set<Entry<PrimaryRowColumn, List<Entry<Key, Value>>>> es = groupedLocks.entrySet();
-    for (Entry<PrimaryRowColumn, List<Entry<Key, Value>>> group : es) {
+    Set<Entry<PrimaryRowColumn, List<LockInfo>>> es = groupedLocks.entrySet();
+    for (Entry<PrimaryRowColumn, List<LockInfo>> group : es) {
       TxInfo txInfo = txiCache.getTransactionInfo(group.getKey());
       switch (txInfo.status) {
         case COMMITTED:
@@ -182,18 +211,24 @@ public class LockResolver {
   }
 
   private static void rollback(Environment env, long startTs, PrimaryRowColumn prc,
-      List<Entry<Key, Value>> value, Map<ByteSequence, Mutation> mutations) {
-    for (Entry<Key, Value> entry : value) {
-      if (isPrimary(prc, entry.getKey())) {
+      List<LockInfo> value, Map<ByteSequence, Mutation> mutations) {
+    for (LockInfo lockInfo : value) {
+      if (isPrimary(prc, lockInfo.entry.getKey())) {
         continue;
       }
 
-      long lockTs = entry.getKey().getTimestamp() & ColumnConstants.TIMESTAMP_MASK;
-      Mutation mut = getMutation(entry.getKey().getRowData(), mutations);
-      Key k = entry.getKey();
-      mut.put(k.getColumnFamilyData().toArray(), k.getColumnQualifierData().toArray(),
-          k.getColumnVisibilityParsed(), ColumnConstants.DEL_LOCK_PREFIX | lockTs,
-          DelLockValue.encodeRollback(false, true));
+      Mutation mut = getMutation(lockInfo.entry.getKey().getRowData(), mutations);
+      Key k = lockInfo.entry.getKey();
+      if (lockInfo.isReadLock) {
+        mut.put(k.getColumnFamilyData().toArray(), k.getColumnQualifierData().toArray(),
+            k.getColumnVisibilityParsed(),
+            ColumnConstants.RLOCK_PREFIX | ReadLockUtil.encodeTs(lockInfo.lockTs, true),
+            DelReadLockValue.encodeRollback());
+      } else {
+        mut.put(k.getColumnFamilyData().toArray(), k.getColumnQualifierData().toArray(),
+            k.getColumnVisibilityParsed(), ColumnConstants.DEL_LOCK_PREFIX | lockInfo.lockTs,
+            DelLockValue.encodeRollback(false, true));
+      }
     }
 
   }
@@ -224,26 +259,31 @@ public class LockResolver {
     }
   }
 
-  private static void commitColumns(Environment env, PrimaryRowColumn prc,
-      List<Entry<Key, Value>> value, long commitTs, Map<ByteSequence, Mutation> mutations) {
-    for (Entry<Key, Value> entry : value) {
-      if (isPrimary(prc, entry.getKey())) {
+  private static void commitColumns(Environment env, PrimaryRowColumn prc, List<LockInfo> value,
+      long commitTs, Map<ByteSequence, Mutation> mutations) {
+    for (LockInfo lockInfo : value) {
+      if (isPrimary(prc, lockInfo.entry.getKey())) {
         continue;
       }
 
-      long lockTs = entry.getKey().getTimestamp() & ColumnConstants.TIMESTAMP_MASK;
+      long lockTs = lockInfo.lockTs;
       // TODO may be that a stronger sanity check that could be done here
       if (commitTs < lockTs) {
         throw new IllegalStateException(
-            "bad commitTs : " + entry.getKey() + " (" + commitTs + "<" + lockTs + ")");
+            "bad commitTs : " + lockInfo.entry.getKey() + " (" + commitTs + "<" + lockTs + ")");
       }
 
-      Mutation mut = getMutation(entry.getKey().getRowData(), mutations);
-      Column col = SpanUtil.toRowColumn(entry.getKey()).getColumn();
+      Mutation mut = getMutation(lockInfo.entry.getKey().getRowData(), mutations);
+      Column col = ColumnUtil.convert(lockInfo.entry.getKey());
 
-      LockValue lv = new LockValue(entry.getValue().get());
-      ColumnUtil.commitColumn(env, lv.isTrigger(), false, col, lv.isWrite(), lv.isDelete(), lockTs,
-          commitTs, env.getConfiguredObservers().getObservedColumns(STRONG), mut);
+      if (lockInfo.isReadLock) {
+        ColumnUtil.commitColumn(env, false, false, col, false, false, true, lockTs, commitTs,
+            env.getConfiguredObservers().getObservedColumns(STRONG), mut);
+      } else {
+        LockValue lv = new LockValue(lockInfo.entry.getValue().get());
+        ColumnUtil.commitColumn(env, lv.isTrigger(), false, col, lv.isWrite(), lv.isDelete(), false,
+            lockTs, commitTs, env.getConfiguredObservers().getObservedColumns(STRONG), mut);
+      }
     }
 
   }
@@ -263,4 +303,43 @@ public class LockResolver {
     return prc.prow.equals(ByteUtil.toBytes(k.getRowData()))
         && prc.pcol.equals(SpanUtil.toRowColumn(k).getColumn());
   }
+
+  static List<Entry<Key, Value>> getOpenReadLocks(Environment env,
+      Map<Bytes, Set<Column>> rowColsToCheck) throws Exception {
+
+    List<Range> ranges = new ArrayList<>();
+
+    for (Entry<Bytes, Set<Column>> e1 : rowColsToCheck.entrySet()) {
+      for (Column col : e1.getValue()) {
+        Key start = SpanUtil.toKey(new RowColumn(e1.getKey(), col));
+        Key end = new Key(start);
+        end.setTimestamp(ColumnConstants.LOCK_PREFIX | ColumnConstants.TIMESTAMP_MASK);
+        ranges.add(new Range(start, true, end, false));
+      }
+    }
+
+    BatchScanner bscanner = null;
+    try {
+      bscanner = env.getConnector().createBatchScanner(env.getTable(), env.getAuthorizations(), 1);
+
+      bscanner.setRanges(ranges);
+      IteratorSetting iterCfg = new IteratorSetting(10, OpenReadLockIterator.class);
+
+      bscanner.addScanIterator(iterCfg);
+
+      List<Entry<Key, Value>> ret = new ArrayList<>();
+      for (Entry<Key, Value> entry : bscanner) {
+        if ((entry.getKey().getTimestamp() & PREFIX_MASK) == ColumnConstants.RLOCK_PREFIX) {
+          ret.add(entry);
+        }
+      }
+
+      return ret;
+
+    } finally {
+      if (bscanner != null) {
+        bscanner.close();
+      }
+    }
+  }
 }
diff --git a/modules/core/src/main/java/org/apache/fluo/core/impl/ParallelSnapshotScanner.java b/modules/core/src/main/java/org/apache/fluo/core/impl/ParallelSnapshotScanner.java
index 37e0df5..48250c0 100644
--- a/modules/core/src/main/java/org/apache/fluo/core/impl/ParallelSnapshotScanner.java
+++ b/modules/core/src/main/java/org/apache/fluo/core/impl/ParallelSnapshotScanner.java
@@ -19,6 +19,7 @@ import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
@@ -51,9 +52,10 @@ public class ParallelSnapshotScanner {
   private List<Range> rangesToScan = new ArrayList<>();
   private Function<ByteSequence, Bytes> rowConverter;
   private Function<Key, Column> columnConverter;
+  private Map<Bytes, Set<Column>> readLocksSeen;
 
   ParallelSnapshotScanner(Collection<Bytes> rows, Set<Column> columns, Environment env,
-      long startTs, TxStats stats) {
+      long startTs, TxStats stats, Map<Bytes, Set<Column>> readLocksSeen) {
     this.rows = rows;
     this.columns = columns;
     this.env = env;
@@ -61,10 +63,11 @@ public class ParallelSnapshotScanner {
     this.stats = stats;
     this.rowConverter = new CachedBytesConverter(rows);
     this.columnConverter = new CachedColumnConverter(columns);
+    this.readLocksSeen = readLocksSeen;
   }
 
-  ParallelSnapshotScanner(Collection<RowColumn> cells, Environment env, long startTs,
-      TxStats stats) {
+  ParallelSnapshotScanner(Collection<RowColumn> cells, Environment env, long startTs, TxStats stats,
+      Map<Bytes, Set<Column>> readLocksSeen) {
     for (RowColumn rc : cells) {
       byte[] r = rc.getRow().toArray();
       byte[] cf = rc.getColumn().getFamily().toArray();
@@ -83,6 +86,7 @@ public class ParallelSnapshotScanner {
     this.stats = stats;
     this.rowConverter = ByteUtil::toBytes;
     this.columnConverter = ColumnUtil::convert;
+    this.readLocksSeen = readLocksSeen;
   }
 
   private BatchScanner setupBatchScanner() {
@@ -101,7 +105,7 @@ public class ParallelSnapshotScanner {
 
     if (rangesToScan.size() > 0) {
       scanner.setRanges(rangesToScan);
-      SnapshotScanner.setupScanner(scanner, Collections.<Column>emptySet(), startTs);
+      SnapshotScanner.setupScanner(scanner, Collections.<Column>emptySet(), startTs, true);
     } else if (rows != null) {
       List<Range> ranges = new ArrayList<>(rows.size());
 
@@ -111,7 +115,7 @@ public class ParallelSnapshotScanner {
 
       scanner.setRanges(ranges);
 
-      SnapshotScanner.setupScanner(scanner, columns, startTs);
+      SnapshotScanner.setupScanner(scanner, columns, startTs, true);
     } else {
       return null;
     }
@@ -176,13 +180,9 @@ public class ParallelSnapshotScanner {
         if (colType == ColumnConstants.LOCK_PREFIX) {
           locks.add(entry);
         } else if (colType == ColumnConstants.DATA_PREFIX) {
-          Map<Column, Bytes> cols = ret.get(row);
-          if (cols == null) {
-            cols = new HashMap<>();
-            ret.put(row, cols);
-          }
-
-          cols.put(col, Bytes.of(entry.getValue().get()));
+          ret.computeIfAbsent(row, k -> new HashMap<>()).put(col, Bytes.of(entry.getValue().get()));
+        } else if (colType == ColumnConstants.RLOCK_PREFIX) {
+          readLocksSeen.computeIfAbsent(row, k -> new HashSet<>()).add(col);
         } else {
           throw new IllegalArgumentException("Unexpected column type " + colType);
         }
@@ -191,5 +191,4 @@ public class ParallelSnapshotScanner {
       bs.close();
     }
   }
-
 }
diff --git a/modules/core/src/main/java/org/apache/fluo/core/impl/ReadLockSnapshot.java b/modules/core/src/main/java/org/apache/fluo/core/impl/ReadLockSnapshot.java
new file mode 100644
index 0000000..eb1ec01
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/fluo/core/impl/ReadLockSnapshot.java
@@ -0,0 +1,79 @@
+/*
+ * 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.fluo.core.impl;
+
+import java.util.Collection;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.fluo.api.client.AbstractSnapshotBase;
+import org.apache.fluo.api.client.SnapshotBase;
+import org.apache.fluo.api.client.scanner.ScannerBuilder;
+import org.apache.fluo.api.data.Bytes;
+import org.apache.fluo.api.data.Column;
+import org.apache.fluo.api.data.RowColumn;
+
+public class ReadLockSnapshot extends AbstractSnapshotBase implements SnapshotBase {
+
+  private TransactionImpl txi;
+
+  ReadLockSnapshot(TransactionImpl txi) {
+    super(txi);
+    this.txi = txi;
+  }
+
+  @Override
+  public Bytes get(Bytes row, Column column) {
+    txi.setReadLock(row, column);
+    return txi.get(row, column);
+  }
+
+  @Override
+  public Map<Column, Bytes> get(Bytes row, Set<Column> columns) {
+    for (Column column : columns) {
+      txi.setReadLock(row, column);
+    }
+    return txi.get(row, columns);
+  }
+
+  @Override
+  public Map<Bytes, Map<Column, Bytes>> get(Collection<Bytes> rows, Set<Column> columns) {
+    for (Bytes row : rows) {
+      for (Column column : columns) {
+        txi.setReadLock(row, column);
+      }
+    }
+    return txi.get(rows, columns);
+  }
+
+  @Override
+  public Map<RowColumn, Bytes> get(Collection<RowColumn> rowColumns) {
+    for (RowColumn rowColumn : rowColumns) {
+      txi.setReadLock(rowColumn.getRow(), rowColumn.getColumn());
+    }
+    return txi.get(rowColumns);
+  }
+
+  @Override
+  public ScannerBuilder scanner() {
+    throw new UnsupportedOperationException();
+  }
+
+  @Override
+  public long getStartTimestamp() {
+    return txi.getStartTimestamp();
+  }
+}
diff --git a/modules/core/src/main/java/org/apache/fluo/core/impl/SnapshotScanner.java b/modules/core/src/main/java/org/apache/fluo/core/impl/SnapshotScanner.java
index cd2e008..a1db35b 100644
--- a/modules/core/src/main/java/org/apache/fluo/core/impl/SnapshotScanner.java
+++ b/modules/core/src/main/java/org/apache/fluo/core/impl/SnapshotScanner.java
@@ -21,6 +21,7 @@ import java.util.Iterator;
 import java.util.List;
 import java.util.Map.Entry;
 import java.util.NoSuchElementException;
+import java.util.function.Consumer;
 
 import com.google.common.collect.ImmutableSet;
 import org.apache.accumulo.core.client.IteratorSetting;
@@ -49,10 +50,12 @@ public class SnapshotScanner implements Iterable<Entry<Key, Value>> {
   public static final class Opts {
     private final Span span;
     private final Collection<Column> columns;
+    private final boolean showReadLocks;
 
-    public Opts(Span span, Collection<Column> columns) {
+    public Opts(Span span, Collection<Column> columns, boolean showReadLocks) {
       this.span = span;
       this.columns = ImmutableSet.copyOf(columns);
+      this.showReadLocks = showReadLocks;
     }
 
     public Span getSpan() {
@@ -62,12 +65,17 @@ public class SnapshotScanner implements Iterable<Entry<Key, Value>> {
     public Collection<Column> getColumns() {
       return columns;
     }
+
+    public boolean getShowReadLocks() {
+      return showReadLocks;
+    }
   }
 
   private final long startTs;
   private final Environment env;
   private final TxStats stats;
   private final Opts config;
+  private Consumer<Entry<Key, Value>> locksSeen;
 
   static final long INITIAL_WAIT_TIME = 50;
   // TODO make configurable
@@ -75,7 +83,8 @@ public class SnapshotScanner implements Iterable<Entry<Key, Value>> {
 
 
 
-  static void setupScanner(ScannerBase scanner, Collection<Column> columns, long startTs) {
+  static void setupScanner(ScannerBase scanner, Collection<Column> columns, long startTs,
+      boolean showReadLocks) {
     for (Column col : columns) {
       if (col.isQualifierSet()) {
         scanner.fetchColumn(ByteUtil.toText(col.getFamily()), ByteUtil.toText(col.getQualifier()));
@@ -86,6 +95,7 @@ public class SnapshotScanner implements Iterable<Entry<Key, Value>> {
 
     IteratorSetting iterConf = new IteratorSetting(10, SnapshotIterator.class);
     SnapshotIterator.setSnaptime(iterConf, startTs);
+    SnapshotIterator.setReturnReadLockPresent(iterConf, showReadLocks);
     scanner.addScanIterator(iterConf);
   }
 
@@ -111,7 +121,7 @@ public class SnapshotScanner implements Iterable<Entry<Key, Value>> {
       scanner.clearScanIterators();
       scanner.setRange(SpanUtil.toRange(snapIterConfig.getSpan()));
 
-      setupScanner(scanner, snapIterConfig.getColumns(), startTs);
+      setupScanner(scanner, snapIterConfig.getColumns(), startTs, snapIterConfig.showReadLocks);
 
       this.iterator = scanner.iterator();
     }
@@ -137,7 +147,7 @@ public class SnapshotScanner implements Iterable<Entry<Key, Value>> {
     }
 
     private void resetScanner(Span span) {
-      snapIterConfig = new Opts(span, snapIterConfig.columns);
+      snapIterConfig = new Opts(span, snapIterConfig.columns, snapIterConfig.showReadLocks);
       setUpIterator();
     }
 
@@ -145,6 +155,8 @@ public class SnapshotScanner implements Iterable<Entry<Key, Value>> {
 
       // read ahead a little bit looking for other locks to resolve
 
+      locksSeen.accept(lockEntry);
+
       long startTime = System.currentTimeMillis();
       long waitTime = INITIAL_WAIT_TIME;
 
@@ -164,6 +176,7 @@ public class SnapshotScanner implements Iterable<Entry<Key, Value>> {
 
           if (colType == ColumnConstants.LOCK_PREFIX) {
             locks.add(entry);
+            locksSeen.accept(lockEntry);
           }
 
           amountRead += entry.getKey().getSize() + entry.getValue().getSize();
@@ -215,6 +228,8 @@ public class SnapshotScanner implements Iterable<Entry<Key, Value>> {
         } else if (colType == ColumnConstants.DATA_PREFIX) {
           stats.incrementEntriesReturned(1);
           return entry;
+        } else if (colType == ColumnConstants.RLOCK_PREFIX) {
+          return entry;
         } else {
           throw new IllegalArgumentException("Unexpected column type " + colType);
         }
@@ -227,11 +242,13 @@ public class SnapshotScanner implements Iterable<Entry<Key, Value>> {
     }
   }
 
-  SnapshotScanner(Environment env, Opts config, long startTs, TxStats stats) {
+  SnapshotScanner(Environment env, Opts config, long startTs, TxStats stats,
+      Consumer<Entry<Key, Value>> locksSeen) {
     this.env = env;
     this.config = config;
     this.startTs = startTs;
     this.stats = stats;
+    this.locksSeen = locksSeen;
   }
 
   @Override
diff --git a/modules/core/src/main/java/org/apache/fluo/core/impl/TransactionImpl.java b/modules/core/src/main/java/org/apache/fluo/core/impl/TransactionImpl.java
index aa5e7da..0abdafe 100644
--- a/modules/core/src/main/java/org/apache/fluo/core/impl/TransactionImpl.java
+++ b/modules/core/src/main/java/org/apache/fluo/core/impl/TransactionImpl.java
@@ -26,10 +26,12 @@ import java.util.Map;
 import java.util.Map.Entry;
 import java.util.Objects;
 import java.util.Set;
+import java.util.function.Consumer;
 
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
 import com.google.common.collect.Iterators;
+import com.google.common.collect.Sets;
 import com.google.common.util.concurrent.FutureCallback;
 import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
@@ -51,10 +53,14 @@ import org.apache.accumulo.core.data.Range;
 import org.apache.accumulo.core.data.Value;
 import org.apache.fluo.accumulo.iterators.PrewriteIterator;
 import org.apache.fluo.accumulo.util.ColumnConstants;
+import org.apache.fluo.accumulo.util.ReadLockUtil;
 import org.apache.fluo.accumulo.values.DelLockValue;
+import org.apache.fluo.accumulo.values.DelReadLockValue;
 import org.apache.fluo.accumulo.values.LockValue;
+import org.apache.fluo.accumulo.values.ReadLockValue;
 import org.apache.fluo.api.client.AbstractTransactionBase;
 import org.apache.fluo.api.client.Snapshot;
+import org.apache.fluo.api.client.SnapshotBase;
 import org.apache.fluo.api.client.scanner.ScannerBuilder;
 import org.apache.fluo.api.data.Bytes;
 import org.apache.fluo.api.data.Column;
@@ -71,13 +77,17 @@ import org.apache.fluo.core.exceptions.AlreadyAcknowledgedException;
 import org.apache.fluo.core.exceptions.StaleScanException;
 import org.apache.fluo.core.impl.scanner.ScannerBuilderImpl;
 import org.apache.fluo.core.oracle.Stamp;
+import org.apache.fluo.core.util.ByteUtil;
 import org.apache.fluo.core.util.ColumnUtil;
 import org.apache.fluo.core.util.ConditionalFlutation;
 import org.apache.fluo.core.util.FluoCondition;
 import org.apache.fluo.core.util.Flutation;
 import org.apache.fluo.core.util.Hex;
 import org.apache.fluo.core.util.SpanUtil;
+import org.apache.fluo.core.util.UtilWaitThread;
 
+import static org.apache.fluo.accumulo.util.ColumnConstants.PREFIX_MASK;
+import static org.apache.fluo.accumulo.util.ColumnConstants.RLOCK_PREFIX;
 import static org.apache.fluo.api.observer.Observer.NotificationType.STRONG;
 import static org.apache.fluo.api.observer.Observer.NotificationType.WEAK;
 
@@ -92,15 +102,21 @@ public class TransactionImpl extends AbstractTransactionBase implements AsyncTra
       Bytes.of("special delete object f804266bf94935edd45ae3e6c287b93c1814295c");
   private static final Bytes NTFY_VAL =
       Bytes.of("special ntfy value ce0c523e6e4dc093be8a2736b82eca1b95f97ed4");
+  private static final Bytes RLOCK_VAL =
+      Bytes.of("special rlock value 94da84e7796ff3b23b779805d820a33f1997cb8b");
 
   private static boolean isWrite(Bytes val) {
-    return val != NTFY_VAL;
+    return val != NTFY_VAL && val != RLOCK_VAL;
   }
 
   private static boolean isDelete(Bytes val) {
     return val == DELETE;
   }
 
+  private static boolean isReadLock(Bytes val) {
+    return val == RLOCK_VAL;
+  }
+
   private static enum TxStatus {
     OPEN, COMMIT_STARTED, COMMITTED, CLOSED
   }
@@ -110,7 +126,9 @@ public class TransactionImpl extends AbstractTransactionBase implements AsyncTra
   private final Map<Bytes, Set<Column>> weakNotifications = new HashMap<>();
   private final Set<Column> observedColumns;
   private final Environment env;
-  final Map<Bytes, Set<Column>> columnsRead = new HashMap<>();
+  private final Map<Bytes, Set<Column>> columnsRead = new HashMap<>();
+  // Tracks row columns that were observed to have had a read lock in the past.
+  private final Map<Bytes, Set<Column>> readLocksSeen = new HashMap<>();
   private final TxStats stats;
   private Notification notification;
   private Notification weakNotification;
@@ -170,7 +188,8 @@ public class TransactionImpl extends AbstractTransactionBase implements AsyncTra
   @Override
   public Map<Column, Bytes> get(Bytes row, Set<Column> columns) {
     checkIfOpen();
-    return getImpl(row, columns);
+    return getImpl(row, columns, kve -> {
+    });
   }
 
   @Override
@@ -183,7 +202,8 @@ public class TransactionImpl extends AbstractTransactionBase implements AsyncTra
 
     env.getSharedResources().getVisCache().validate(columns);
 
-    ParallelSnapshotScanner pss = new ParallelSnapshotScanner(rows, columns, env, startTs, stats);
+    ParallelSnapshotScanner pss =
+        new ParallelSnapshotScanner(rows, columns, env, startTs, stats, readLocksSeen);
 
     Map<Bytes, Map<Column, Bytes>> ret = pss.scan();
 
@@ -202,7 +222,8 @@ public class TransactionImpl extends AbstractTransactionBase implements AsyncTra
       return Collections.emptyMap();
     }
 
-    ParallelSnapshotScanner pss = new ParallelSnapshotScanner(rowColumns, env, startTs, stats);
+    ParallelSnapshotScanner pss =
+        new ParallelSnapshotScanner(rowColumns, env, startTs, stats, readLocksSeen);
 
     Map<Bytes, Map<Column, Bytes>> scan = pss.scan();
     Map<RowColumn, Bytes> ret = new HashMap<>();
@@ -217,7 +238,8 @@ public class TransactionImpl extends AbstractTransactionBase implements AsyncTra
     return ret;
   }
 
-  private Map<Column, Bytes> getImpl(Bytes row, Set<Column> columns) {
+  private Map<Column, Bytes> getImpl(Bytes row, Set<Column> columns,
+      Consumer<Entry<Key, Value>> locksSeen) {
 
     // TODO push visibility filtering to server side?
 
@@ -241,19 +263,26 @@ public class TransactionImpl extends AbstractTransactionBase implements AsyncTra
           cols.add(column);
         }
       }
-      opts = new SnapshotScanner.Opts(Span.exact(row), columns);
+      opts = new SnapshotScanner.Opts(Span.exact(row), columns, true);
     } else {
-      opts = new SnapshotScanner.Opts(Span.exact(row), columns);
+      opts = new SnapshotScanner.Opts(Span.exact(row), columns, true);
     }
 
     Map<Column, Bytes> ret = new HashMap<>();
+    Set<Column> readLockCols = null;
+
+    for (Entry<Key, Value> kve : new SnapshotScanner(env, opts, startTs, stats, locksSeen)) {
 
-    for (Entry<Key, Value> kve : new SnapshotScanner(env, opts, startTs, stats)) {
       Column col = ColumnUtil.convert(kve.getKey());
-      if (shouldCopy) {
-        if (columns.contains(col)) {
-          ret.put(col, Bytes.of(kve.getValue().get()));
+      if (shouldCopy && !columns.contains(col)) {
+        continue;
+      }
+
+      if ((kve.getKey().getTimestamp() & PREFIX_MASK) == RLOCK_PREFIX) {
+        if (readLockCols == null) {
+          readLockCols = readLocksSeen.computeIfAbsent(row, k -> new HashSet<>());
         }
+        readLockCols.add(col);
       } else {
         ret.put(col, Bytes.of(kve.getValue().get()));
       }
@@ -280,6 +309,31 @@ public class TransactionImpl extends AbstractTransactionBase implements AsyncTra
     colsRead.addAll(columns);
   }
 
+  void setReadLock(Bytes row, Column col) {
+    checkIfOpen();
+    Objects.requireNonNull(row);
+    Objects.requireNonNull(col);
+
+    if (col.getFamily().equals(ColumnConstants.NOTIFY_CF)) {
+      throw new IllegalArgumentException(ColumnConstants.NOTIFY_CF + " is a reserved family");
+    }
+
+    env.getSharedResources().getVisCache().validate(col);
+
+    Map<Column, Bytes> colUpdates = updates.computeIfAbsent(row, k -> new HashMap<>());
+    Bytes curVal = colUpdates.get(col);
+    if (curVal != null && (isWrite(curVal) || isDelete(curVal))) {
+      throw new AlreadySetException("Attemped read lock after write lock " + row + " " + col);
+    }
+
+    colUpdates.put(col, RLOCK_VAL);
+  }
+
+  @Override
+  public SnapshotBase withReadLock() {
+    return new ReadLockSnapshot(this);
+  }
+
   @Override
   public void set(Bytes row, Column col, Bytes value) throws AlreadySetException {
     checkIfOpen();
@@ -293,13 +347,7 @@ public class TransactionImpl extends AbstractTransactionBase implements AsyncTra
 
     env.getSharedResources().getVisCache().validate(col);
 
-    // TODO copy?
-
-    Map<Column, Bytes> colUpdates = updates.get(row);
-    if (colUpdates == null) {
-      colUpdates = new HashMap<>();
-      updates.put(row, colUpdates);
-    }
+    Map<Column, Bytes> colUpdates = updates.computeIfAbsent(row, k -> new HashMap<>());
 
     Bytes curVal = colUpdates.get(col);
     if (curVal != null && isWrite(curVal)) {
@@ -346,6 +394,10 @@ public class TransactionImpl extends AbstractTransactionBase implements AsyncTra
       PrewriteIterator.enableAckCheck(iterConf, notification.getTimestamp());
     }
 
+    if (isReadLock(val)) {
+      PrewriteIterator.setReadlock(iterConf);
+    }
+
     Condition cond = new FluoCondition(env, col).setIterators(iterConf);
 
     if (cm == null) {
@@ -358,8 +410,13 @@ public class TransactionImpl extends AbstractTransactionBase implements AsyncTra
       cm.put(col, ColumnConstants.DATA_PREFIX | startTs, val.toArray());
     }
 
-    cm.put(col, ColumnConstants.LOCK_PREFIX | startTs, LockValue.encode(primaryRow, primaryColumn,
-        isWrite(val), isDelete(val), isTriggerRow, getTransactorID()));
+    if (isReadLock(val)) {
+      cm.put(col, ColumnConstants.RLOCK_PREFIX | ReadLockUtil.encodeTs(startTs, false),
+          ReadLockValue.encode(primaryRow, primaryColumn, getTransactorID()));
+    } else {
+      cm.put(col, ColumnConstants.LOCK_PREFIX | startTs, LockValue.encode(primaryRow, primaryColumn,
+          isWrite(val), isDelete(val), isTriggerRow, getTransactorID()));
+    }
 
     return cm;
   }
@@ -497,7 +554,7 @@ public class TransactionImpl extends AbstractTransactionBase implements AsyncTra
    *
    * @param cd Commit data
    */
-  private void readUnread(CommitData cd) throws Exception {
+  private void readUnread(CommitData cd, Consumer<Entry<Key, Value>> locksSeen) throws Exception {
     // TODO make async
     // TODO need to keep track of ranges read (not ranges passed in, but actual data read... user
     // may not iterate over entire range
@@ -517,10 +574,86 @@ public class TransactionImpl extends AbstractTransactionBase implements AsyncTra
     }
 
     for (Entry<Bytes, Set<Column>> entry : columnsToRead.entrySet()) {
-      getImpl(entry.getKey(), entry.getValue());
+      getImpl(entry.getKey(), entry.getValue(), locksSeen);
     }
   }
 
+  private void checkForOrphanedReadLocks(CommitData cd, Map<Bytes, Set<Column>> locksResolved)
+      throws Exception {
+
+    if (readLocksSeen.size() == 0) {
+      return;
+    }
+
+    Map<Bytes, Set<Column>> rowColsToCheck = new HashMap<>();
+
+    for (Entry<Bytes, Set<Column>> entry : cd.getRejected().entrySet()) {
+
+      Set<Column> resolvedColumns =
+          locksResolved.getOrDefault(entry.getKey(), Collections.emptySet());
+
+      Set<Column> colsToCheck = null;
+      Set<Column> readLockCols = readLocksSeen.get(entry.getKey());
+      if (readLockCols != null) {
+        for (Column candidate : Sets.intersection(readLockCols, entry.getValue())) {
+          if (resolvedColumns.contains(candidate)) {
+            // A write lock was seen and this is probably what caused the collision, no need to
+            // check this column for read locks.
+            continue;
+          }
+
+          if (!isReadLock(updates.getOrDefault(entry.getKey(), Collections.emptyMap())
+              .getOrDefault(candidate, EMPTY_BS))) {
+            if (colsToCheck == null) {
+              colsToCheck = new HashSet<>();
+            }
+            colsToCheck.add(candidate);
+          }
+        }
+
+        if (colsToCheck != null) {
+          rowColsToCheck.put(entry.getKey(), colsToCheck);
+        }
+      }
+    }
+
+    if (rowColsToCheck.size() > 0) {
+
+      long startTime = System.currentTimeMillis();
+      long waitTime = SnapshotScanner.INITIAL_WAIT_TIME;
+
+      boolean resolved = false;
+
+      List<Entry<Key, Value>> openReadLocks = LockResolver.getOpenReadLocks(env, rowColsToCheck);
+
+      startTime = System.currentTimeMillis();
+
+      while (!resolved) {
+        resolved = LockResolver.resolveLocks(env, startTs, stats, openReadLocks, startTime);
+        if (!resolved) {
+          UtilWaitThread.sleep(waitTime);
+          stats.incrementLockWaitTime(waitTime);
+          waitTime = Math.min(SnapshotScanner.MAX_WAIT_TIME, waitTime * 2);
+
+          openReadLocks = LockResolver.getOpenReadLocks(env, rowColsToCheck);
+        }
+      }
+    }
+  }
+
+  private void checkForOrphanedLocks(CommitData cd) throws Exception {
+
+    Map<Bytes, Set<Column>> locksSeen = new HashMap<>();
+
+    readUnread(cd, kve -> {
+      Bytes row = ByteUtil.toBytes(kve.getKey().getRowData());
+      Column col = ColumnUtil.convert(kve.getKey());
+      locksSeen.computeIfAbsent(row, k -> new HashSet<>()).add(col);
+    });
+
+    checkForOrphanedReadLocks(cd, locksSeen);
+  }
+
   private boolean checkForAckCollision(ConditionalMutation cm) {
     Bytes row = Bytes.of(cm.getRow());
 
@@ -792,8 +925,8 @@ public class TransactionImpl extends AbstractTransactionBase implements AsyncTra
       stats.incrementEntriesSet(cols.size());
     }
 
-    Bytes primRow;
-    Column primCol;
+    Bytes primRow = null;
+    Column primCol = null;
 
     if (primary != null) {
       primRow = primary.getRow();
@@ -805,9 +938,23 @@ public class TransactionImpl extends AbstractTransactionBase implements AsyncTra
       primRow = notification.getRow();
       primCol = notification.getColumn();
     } else {
-      primRow = updates.keySet().iterator().next();
-      Map<Column, Bytes> colSet = updates.get(primRow);
-      primCol = colSet.keySet().iterator().next();
+
+      outer: for (Entry<Bytes, Map<Column, Bytes>> entry : updates.entrySet()) {
+        for (Entry<Column, Bytes> entry2 : entry.getValue().entrySet()) {
+          if (!isReadLock(entry2.getValue())) {
+            primRow = entry.getKey();
+            primCol = entry2.getKey();
+            break outer;
+          }
+        }
+      }
+
+      if (primRow == null) {
+        // there are only read locks, so nothing to write
+        deleteWeakRow();
+        commitCallback.committed();
+        return;
+      }
     }
 
     // get a primary column
@@ -825,7 +972,6 @@ public class TransactionImpl extends AbstractTransactionBase implements AsyncTra
     final ConditionalMutation pcm =
         prewrite(cd.prow, cd.pcol, cd.pval, cd.prow, cd.pcol, isTriggerRow(cd.prow));
 
-
     ListenableFuture<Iterator<Result>> future = cd.acw.apply(Collections.singletonList(pcm));
     Futures.addCallback(future, new CommitCallback<Iterator<Result>>(cd) {
       @Override
@@ -881,7 +1027,7 @@ public class TransactionImpl extends AbstractTransactionBase implements AsyncTra
       cd.addPrimaryToRejected();
       getStats().setRejected(cd.getRejected());
       // TODO do async
-      readUnread(cd);
+      checkForOrphanedLocks(cd);
       if (checkForAckCollision(pcm)) {
         cd.commitObserver.alreadyAcknowledged();
       } else {
@@ -913,7 +1059,6 @@ public class TransactionImpl extends AbstractTransactionBase implements AsyncTra
 
     cd.acceptedRows = new HashSet<>();
 
-
     ListenableFuture<Iterator<Result>> future = cd.bacw.apply(mutations);
     Futures.addCallback(future, new CommitCallback<Iterator<Result>>(cd) {
       @Override
@@ -940,7 +1085,7 @@ public class TransactionImpl extends AbstractTransactionBase implements AsyncTra
       env.getSharedResources().getSyncCommitExecutor().execute(new SynchronousCommitTask(cd) {
         @Override
         protected void runCommitStep(CommitData cd) throws Exception {
-          readUnread(cd);
+          checkForOrphanedLocks(cd);
           rollbackOtherLocks(cd);
         }
       });
@@ -957,7 +1102,6 @@ public class TransactionImpl extends AbstractTransactionBase implements AsyncTra
     }
   }
 
-
   private void rollbackOtherLocks(CommitData cd) throws Exception {
     // roll back locks
 
@@ -968,9 +1112,14 @@ public class TransactionImpl extends AbstractTransactionBase implements AsyncTra
     ArrayList<Mutation> mutations = new ArrayList<>(cd.acceptedRows.size());
     for (Bytes row : cd.acceptedRows) {
       m = new Flutation(env, row);
-      for (Column col : updates.get(row).keySet()) {
-        m.put(col, ColumnConstants.DEL_LOCK_PREFIX | startTs,
-            DelLockValue.encodeRollback(false, true));
+      for (Entry<Column, Bytes> entry : updates.get(row).entrySet()) {
+        if (isReadLock(entry.getValue())) {
+          m.put(entry.getKey(), ColumnConstants.RLOCK_PREFIX | ReadLockUtil.encodeTs(startTs, true),
+              DelReadLockValue.encodeRollback());
+        } else {
+          m.put(entry.getKey(), ColumnConstants.DEL_LOCK_PREFIX | startTs,
+              DelLockValue.encodeRollback(false, true));
+        }
       }
       mutations.add(m);
     }
@@ -1004,7 +1153,6 @@ public class TransactionImpl extends AbstractTransactionBase implements AsyncTra
     }, env.getSharedResources().getAsyncCommitExecutor());
   }
 
-
   private void beginSecondCommitPhase(CommitData cd, Stamp commitStamp) throws Exception {
     if (startTs < commitStamp.getGcTimestamp()) {
       rollbackOtherLocks(cd);
@@ -1065,7 +1213,6 @@ public class TransactionImpl extends AbstractTransactionBase implements AsyncTra
       }
     }
 
-
     ListenableFuture<Void> future =
         env.getSharedResources().getBatchWriter().writeMutationsAsyncFuture(mutations.values());
     Futures.addCallback(future, new CommitCallback<Void>(cd) {
@@ -1088,8 +1235,7 @@ public class TransactionImpl extends AbstractTransactionBase implements AsyncTra
     final ConditionalMutation delLockMutation = new ConditionalFlutation(env, cd.prow, lockCheck);
 
     ColumnUtil.commitColumn(env, isTrigger, true, cd.pcol, isWrite(cd.pval), isDelete(cd.pval),
-        startTs, commitTs, observedColumns, delLockMutation);
-
+        isReadLock(cd.pval), startTs, commitTs, observedColumns, delLockMutation);
 
     ListenableFuture<Iterator<Result>> future =
         cd.acw.apply(Collections.singletonList(delLockMutation));
@@ -1146,7 +1292,6 @@ public class TransactionImpl extends AbstractTransactionBase implements AsyncTra
     }
   }
 
-
   private void postCommitPrimary(CommitData cd, long commitTs, Status mutationStatus)
       throws Exception {
     if (mutationStatus != Status.ACCEPTED) {
@@ -1170,7 +1315,7 @@ public class TransactionImpl extends AbstractTransactionBase implements AsyncTra
         ColumnUtil.commitColumn(env,
             isTriggerRow && colUpdates.getKey().equals(notification.getColumn()), false,
             colUpdates.getKey(), isWrite(colUpdates.getValue()), isDelete(colUpdates.getValue()),
-            startTs, commitTs, observedColumns, m);
+            isReadLock(colUpdates.getValue()), startTs, commitTs, observedColumns, m);
       }
 
       mutations.add(m);
@@ -1217,6 +1362,8 @@ public class TransactionImpl extends AbstractTransactionBase implements AsyncTra
   }
 
   public SnapshotScanner newSnapshotScanner(Span span, Collection<Column> columns) {
-    return new SnapshotScanner(env, new SnapshotScanner.Opts(span, columns), startTs, stats);
+    return new SnapshotScanner(env, new SnapshotScanner.Opts(span, columns, false), startTs, stats,
+        kve -> {
+        });
   }
 }
diff --git a/modules/core/src/main/java/org/apache/fluo/core/log/TracingTransaction.java b/modules/core/src/main/java/org/apache/fluo/core/log/TracingTransaction.java
index a4c35c3..21406dc 100644
--- a/modules/core/src/main/java/org/apache/fluo/core/log/TracingTransaction.java
+++ b/modules/core/src/main/java/org/apache/fluo/core/log/TracingTransaction.java
@@ -20,8 +20,10 @@ import java.util.Map;
 import java.util.Set;
 
 import com.google.common.collect.Iterators;
+import org.apache.fluo.api.client.AbstractSnapshotBase;
 import org.apache.fluo.api.client.AbstractTransactionBase;
 import org.apache.fluo.api.client.Snapshot;
+import org.apache.fluo.api.client.SnapshotBase;
 import org.apache.fluo.api.client.scanner.ScannerBuilder;
 import org.apache.fluo.api.config.FluoConfiguration;
 import org.apache.fluo.api.data.Bytes;
@@ -84,7 +86,6 @@ public class TracingTransaction extends AbstractTransactionBase
         e -> encNonAscii(e.getKey()) + "=" + encNonAscii(e.getValue())));
   }
 
-
   private String toStringEncNonAsciiCC(Collection<Column> columns) {
     return Iterators.toString(Iterators.transform(columns.iterator(), Hex::encNonAscii));
   }
@@ -126,47 +127,66 @@ public class TracingTransaction extends AbstractTransactionBase
 
   }
 
-  @Override
-  public Bytes get(Bytes row, Column column) {
-    Bytes ret = tx.get(row, column);
+  private Bytes get(SnapshotBase snap, Bytes row, Column column, String prefix) {
+    Bytes ret = snap.get(row, column);
     if (log.isTraceEnabled()) {
-      log.trace("txid: {} get({}, {}) -> {}", txid, encNonAscii(row), toStringEncNonAscii(column),
-          encNonAscii(ret));
+      log.trace("txid: {} {}get({}, {}) -> {}", txid, prefix, encNonAscii(row),
+          toStringEncNonAscii(column), encNonAscii(ret));
     }
     return ret;
   }
 
-  @Override
-  public Map<Column, Bytes> get(Bytes row, Set<Column> columns) {
-    Map<Column, Bytes> ret = tx.get(row, columns);
+  private Map<Column, Bytes> get(SnapshotBase snap, Bytes row, Set<Column> columns, String prefix) {
+    Map<Column, Bytes> ret = snap.get(row, columns);
     if (log.isTraceEnabled()) {
-      log.trace("txid: {} get({}, {}) -> {}", txid, encNonAscii(row),
+      log.trace("txid: {} {}get({}, {}) -> {}", txid, prefix, encNonAscii(row),
           toStringEncNonAsciiCC(columns), toStringEncNonAsciiMCB(ret));
     }
     return ret;
   }
 
-  @Override
-  public Map<Bytes, Map<Column, Bytes>> get(Collection<Bytes> rows, Set<Column> columns) {
-    Map<Bytes, Map<Column, Bytes>> ret = tx.get(rows, columns);
+  private Map<Bytes, Map<Column, Bytes>> get(SnapshotBase snap, Collection<Bytes> rows,
+      Set<Column> columns, String prefix) {
+    Map<Bytes, Map<Column, Bytes>> ret = snap.get(rows, columns);
     if (log.isTraceEnabled()) {
-      log.trace("txid: {} get({}, {}) -> {}", txid, toStringEncNonAsciiCB(rows),
+      log.trace("txid: {} {}get({}, {}) -> {}", txid, prefix, toStringEncNonAsciiCB(rows),
           toStringEncNonAsciiCC(columns), toStringEncNonAsciiMBMCB(ret));
     }
     return ret;
   }
 
-  @Override
-  public Map<RowColumn, Bytes> get(Collection<RowColumn> rowColumns) {
-    Map<RowColumn, Bytes> ret = tx.get(rowColumns);
+
+  private Map<RowColumn, Bytes> get(SnapshotBase snap, Collection<RowColumn> rowColumns,
+      String prefix) {
+    Map<RowColumn, Bytes> ret = snap.get(rowColumns);
     if (log.isTraceEnabled()) {
-      log.trace("txid: {} get({}) -> {}", txid, toStringEncNonAsciiCRC(rowColumns),
+      log.trace("txid: {} {}get({}) -> {}", txid, prefix, toStringEncNonAsciiCRC(rowColumns),
           toStringEncNonAsciiMRCB(ret));
     }
     return ret;
   }
 
   @Override
+  public Bytes get(Bytes row, Column column) {
+    return get(tx, row, column, "");
+  }
+
+  @Override
+  public Map<Column, Bytes> get(Bytes row, Set<Column> columns) {
+    return get(tx, row, columns, "");
+  }
+
+  @Override
+  public Map<Bytes, Map<Column, Bytes>> get(Collection<Bytes> rows, Set<Column> columns) {
+    return get(tx, rows, columns, "");
+  }
+
+  @Override
+  public Map<RowColumn, Bytes> get(Collection<RowColumn> rowColumns) {
+    return get(tx, rowColumns, "");
+  }
+
+  @Override
   public ScannerBuilder scanner() {
     return new TracingScannerBuilder(tx.scanner(), txid);
   }
@@ -303,4 +323,43 @@ public class TracingTransaction extends AbstractTransactionBase
   public int getSize() {
     return tx.getSize();
   }
+
+  @Override
+  public SnapshotBase withReadLock() {
+    SnapshotBase rltx = tx.withReadLock();
+
+    return new AbstractSnapshotBase() {
+
+      @Override
+      public ScannerBuilder scanner() {
+        // this is an unsupported op and will throw an exception so don't bother w/ trace logging
+        return rltx.scanner();
+      }
+
+      @Override
+      public long getStartTimestamp() {
+        return rltx.getStartTimestamp();
+      }
+
+      @Override
+      public Map<RowColumn, Bytes> get(Collection<RowColumn> rowColumns) {
+        return TracingTransaction.this.get(rltx, rowColumns, "withReadLock().");
+      }
+
+      @Override
+      public Map<Bytes, Map<Column, Bytes>> get(Collection<Bytes> rows, Set<Column> columns) {
+        return TracingTransaction.this.get(rltx, rows, columns, "withReadLock().");
+      }
+
+      @Override
+      public Map<Column, Bytes> get(Bytes row, Set<Column> columns) {
+        return TracingTransaction.this.get(rltx, row, columns, "withReadLock().");
+      }
+
+      @Override
+      public Bytes get(Bytes row, Column column) {
+        return TracingTransaction.this.get(rltx, row, column, "withReadLock().");
+      }
+    };
+  }
 }
diff --git a/modules/core/src/main/java/org/apache/fluo/core/util/ColumnUtil.java b/modules/core/src/main/java/org/apache/fluo/core/util/ColumnUtil.java
index bbc2d03..7d488ca 100644
--- a/modules/core/src/main/java/org/apache/fluo/core/util/ColumnUtil.java
+++ b/modules/core/src/main/java/org/apache/fluo/core/util/ColumnUtil.java
@@ -30,7 +30,9 @@ import org.apache.accumulo.core.data.Mutation;
 import org.apache.accumulo.core.data.Value;
 import org.apache.accumulo.core.security.ColumnVisibility;
 import org.apache.fluo.accumulo.util.ColumnConstants;
+import org.apache.fluo.accumulo.util.ReadLockUtil;
 import org.apache.fluo.accumulo.values.DelLockValue;
+import org.apache.fluo.accumulo.values.DelReadLockValue;
 import org.apache.fluo.accumulo.values.WriteValue;
 import org.apache.fluo.api.data.Bytes;
 import org.apache.fluo.api.data.Bytes.BytesBuilder;
@@ -51,9 +53,13 @@ public class ColumnUtil {
   }
 
   public static void commitColumn(Environment env, boolean isTrigger, boolean isPrimary, Column col,
-      boolean isWrite, boolean isDelete, long startTs, long commitTs, Set<Column> observedColumns,
-      Mutation m) {
-    if (isWrite) {
+      boolean isWrite, boolean isDelete, boolean isReadlock, long startTs, long commitTs,
+      Set<Column> observedColumns, Mutation m) {
+    if (isReadlock) {
+      Flutation.put(env, m, col,
+          ColumnConstants.RLOCK_PREFIX | ReadLockUtil.encodeTs(startTs, true),
+          DelReadLockValue.encodeCommit(commitTs));
+    } else if (isWrite) {
       Flutation.put(env, m, col, ColumnConstants.WRITE_PREFIX | commitTs,
           WriteValue.encode(startTs, isPrimary, isDelete));
     } else {
diff --git a/modules/integration/src/test/java/org/apache/fluo/integration/ITBase.java b/modules/integration/src/test/java/org/apache/fluo/integration/ITBase.java
index 57299e9..4e481e4 100644
--- a/modules/integration/src/test/java/org/apache/fluo/integration/ITBase.java
+++ b/modules/integration/src/test/java/org/apache/fluo/integration/ITBase.java
@@ -17,6 +17,7 @@ package org.apache.fluo.integration;
 
 import java.io.File;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Consumer;
 
 import org.apache.accumulo.core.client.Connector;
 import org.apache.accumulo.core.client.Instance;
@@ -110,18 +111,22 @@ public class ITBase {
     return TABLE_BASE + tableCounter.incrementAndGet();
   }
 
-  protected void printSnapshot() throws Exception {
+  protected void printSnapshot(Consumer<String> out) throws Exception {
     try (Snapshot s = client.newSnapshot()) {
-      System.out.println("== snapshot start ==");
+      out.accept("== snapshot start ==");
 
       for (RowColumnValue rcv : s.scanner().build()) {
-        System.out.println(rcv.getRow() + " " + rcv.getColumn() + "\t" + rcv.getValue());
+        out.accept(rcv.getRow() + " " + rcv.getColumn() + "\t" + rcv.getValue());
       }
 
-      System.out.println("=== snapshot end ===");
+      out.accept("=== snapshot end ===");
     }
   }
 
+  protected void printSnapshot() throws Exception {
+    printSnapshot(System.out::println);
+  }
+
   @AfterClass
   public static void tearDownAccumulo() throws Exception {
     if (startedCluster) {
diff --git a/modules/integration/src/test/java/org/apache/fluo/integration/TestTransaction.java b/modules/integration/src/test/java/org/apache/fluo/integration/TestTransaction.java
index 57b3923..a175500 100644
--- a/modules/integration/src/test/java/org/apache/fluo/integration/TestTransaction.java
+++ b/modules/integration/src/test/java/org/apache/fluo/integration/TestTransaction.java
@@ -31,6 +31,7 @@ import org.apache.fluo.accumulo.iterators.NotificationIterator;
 import org.apache.fluo.accumulo.util.ColumnConstants;
 import org.apache.fluo.accumulo.util.NotificationUtil;
 import org.apache.fluo.api.client.AbstractTransactionBase;
+import org.apache.fluo.api.client.SnapshotBase;
 import org.apache.fluo.api.client.TransactionBase;
 import org.apache.fluo.api.client.scanner.ScannerBuilder;
 import org.apache.fluo.api.data.Bytes;
@@ -199,4 +200,9 @@ public class TestTransaction extends AbstractTransactionBase implements Transact
   public long getStartTimestamp() {
     return tx.getStartTimestamp();
   }
+
+  @Override
+  public SnapshotBase withReadLock() {
+    return tx.withReadLock();
+  }
 }
diff --git a/modules/integration/src/test/java/org/apache/fluo/integration/impl/GarbageCollectionIteratorIT.java b/modules/integration/src/test/java/org/apache/fluo/integration/impl/GarbageCollectionIteratorIT.java
index bdb4b51..fc553f5 100644
--- a/modules/integration/src/test/java/org/apache/fluo/integration/impl/GarbageCollectionIteratorIT.java
+++ b/modules/integration/src/test/java/org/apache/fluo/integration/impl/GarbageCollectionIteratorIT.java
@@ -29,6 +29,7 @@ import org.apache.fluo.accumulo.format.FluoFormatter;
 import org.apache.fluo.accumulo.util.ColumnConstants;
 import org.apache.fluo.accumulo.util.ZookeeperPath;
 import org.apache.fluo.accumulo.util.ZookeeperUtil;
+import org.apache.fluo.api.client.TransactionBase;
 import org.apache.fluo.api.data.Column;
 import org.apache.fluo.core.impl.TransactionImpl.CommitData;
 import org.apache.fluo.core.impl.TransactorNode;
@@ -176,6 +177,88 @@ public class GarbageCollectionIteratorIT extends ITBaseImpl {
     Assert.assertEquals(0, countInTable("-DATA"));
   }
 
+  private void increment(TransactionBase tx, String row, Column col) {
+    int count = Integer.parseInt(tx.gets(row, col, "0"));
+    tx.set(row, col, count + 1 + "");
+  }
+
+  @Test(timeout = 60000)
+  public void testReadLocks() throws Exception {
+
+    final Column altIdCol = new Column("info", "altId");
+
+    TestTransaction tx1 = new TestTransaction(env);
+    for (int i = 0; i < 10; i++) {
+      tx1.set(String.format("n:%03d", i), altIdCol, "" + (19 * (1 + i)));
+    }
+
+    tx1.done();
+
+    for (int i = 0; i < 50; i++) {
+      String row = String.format("n:%03d", i % 10);
+
+      TestTransaction tx = new TestTransaction(env);
+      String altId = tx.withReadLock().gets(row, altIdCol);
+
+      increment(tx, "a:" + altId, new Column("count", row));
+
+      tx.done();
+    }
+
+    Assert.assertEquals(50, countInTable("-DEL_RLOCK"));
+    Assert.assertEquals(50, countInTable("-RLOCK"));
+
+    TestTransaction tx2 = new TestTransaction(env);
+    for (int i = 0; i < 10; i++) {
+      String row = String.format("n:%03d", i);
+      String newAltId = (13 * (i + 1)) + "";
+      String currAltId = tx2.gets(row, altIdCol);
+
+
+      tx2.set(row, altIdCol, newAltId);
+
+      String count = tx2.gets("a:" + currAltId, new Column("count", row));
+      tx2.set("a:" + newAltId, new Column("count", row), count);
+      tx2.delete("a:" + currAltId, new Column("count", row));
+    }
+
+    tx2.done();
+
+    // all read locks should be garbage collected because of the writes after the read locks
+    conn.tableOperations().compact(table, null, null, true, true);
+
+    Assert.assertEquals(0, countInTable("-DEL_RLOCK"));
+    Assert.assertEquals(0, countInTable("-RLOCK"));
+
+    for (int i = 0; i < 50; i++) {
+      String row = String.format("n:%03d", i % 10);
+
+      TestTransaction tx = new TestTransaction(env);
+      String altId = tx.withReadLock().gets(row, altIdCol);
+
+      increment(tx, "a:" + altId, new Column("count", row));
+
+      tx.done();
+    }
+
+    TestTransaction tx3 = new TestTransaction(env);
+    for (int i = 0; i < 10; i++) {
+      String row = String.format("n:%03d", i);
+      String currAltId = tx3.gets(row, altIdCol);
+      Assert.assertEquals("10", tx3.gets("a:" + currAltId, new Column("count", row)));
+    }
+
+    tx3.done();
+
+    waitForGcTime(tx3.getStartTimestamp());
+    conn.tableOperations().compact(table, null, null, true, true);
+
+
+    // all read locks older than GC time should be dropped
+    Assert.assertEquals(0, countInTable("-DEL_RLOCK"));
+    Assert.assertEquals(0, countInTable("-RLOCK"));
+  }
+
   private int countInTable(String str) throws TableNotFoundException {
     int count = 0;
     Scanner scanner = conn.createScanner(table, Authorizations.EMPTY);
diff --git a/modules/integration/src/test/java/org/apache/fluo/integration/impl/ReadLockFailureIT.java b/modules/integration/src/test/java/org/apache/fluo/integration/impl/ReadLockFailureIT.java
new file mode 100644
index 0000000..489b9e0
--- /dev/null
+++ b/modules/integration/src/test/java/org/apache/fluo/integration/impl/ReadLockFailureIT.java
@@ -0,0 +1,427 @@
+/*
+ * 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.fluo.integration.impl;
+
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Set;
+import java.util.function.Consumer;
+
+import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.Iterables;
+import org.apache.accumulo.core.client.Scanner;
+import org.apache.accumulo.core.client.TableNotFoundException;
+import org.apache.accumulo.core.data.Key;
+import org.apache.accumulo.core.data.Value;
+import org.apache.accumulo.core.security.Authorizations;
+import org.apache.fluo.accumulo.format.FluoFormatter;
+import org.apache.fluo.api.client.Snapshot;
+import org.apache.fluo.api.client.Transaction;
+import org.apache.fluo.api.client.TransactionBase;
+import org.apache.fluo.api.data.Column;
+import org.apache.fluo.api.data.RowColumn;
+import org.apache.fluo.api.data.RowColumnValue;
+import org.apache.fluo.api.data.Span;
+import org.apache.fluo.api.exceptions.CommitException;
+import org.apache.fluo.core.impl.TransactionImpl.CommitData;
+import org.apache.fluo.core.impl.TransactorNode;
+import org.apache.fluo.core.oracle.Stamp;
+import org.apache.fluo.integration.ITBaseImpl;
+import org.apache.fluo.integration.TestTransaction;
+import org.junit.Assert;
+import org.junit.Test;
+
+import static org.apache.fluo.integration.impl.ReadLockIT.addEdge;
+import static org.apache.fluo.integration.impl.ReadLockIT.setAlias;
+
+public class ReadLockFailureIT extends ITBaseImpl {
+
+  private void dumpTable(Consumer<String> out) throws TableNotFoundException {
+    Scanner scanner = conn.createScanner(getCurTableName(), Authorizations.EMPTY);
+    for (Entry<Key, Value> entry : scanner) {
+      out.accept(FluoFormatter.toString(entry));
+    }
+  }
+
+  private Set<String> getDerivedEdges() {
+    Set<String> derivedEdges = new HashSet<>();
+    try (Snapshot snap = client.newSnapshot()) {
+      snap.scanner().over(Span.prefix("d:")).build().stream().map(RowColumnValue::getsRow)
+          .map(r -> r.substring(2)).forEach(derivedEdges::add);
+    }
+    return derivedEdges;
+  }
+
+  private void expectCommitException(Consumer<Transaction> retryAction) {
+    try (Transaction tx = client.newTransaction()) {
+      retryAction.accept(tx);
+      tx.commit();
+      Assert.fail();
+    } catch (CommitException ce) {
+
+    }
+  }
+
+  private void retryOnce(Consumer<Transaction> retryAction) {
+
+    expectCommitException(retryAction);
+
+    try (Transaction tx = client.newTransaction()) {
+      retryAction.accept(tx);
+      tx.commit();
+    }
+  }
+
+  private void retryTwice(Consumer<Transaction> retryAction) {
+
+    expectCommitException(retryAction);
+    expectCommitException(retryAction);
+
+    try (Transaction tx = client.newTransaction()) {
+      retryAction.accept(tx);
+      tx.commit();
+    }
+  }
+
+
+  private TransactorNode partiallyCommit(Consumer<TransactionBase> action, boolean commitPrimary,
+      boolean closeTransactor) throws Exception {
+    TransactorNode t2 = new TransactorNode(env);
+    TestTransaction tx2 = new TestTransaction(env, t2);
+
+    action.accept(tx2);
+
+    CommitData cd = tx2.createCommitData();
+    Assert.assertTrue(tx2.preCommit(cd));
+
+    if (commitPrimary) {
+      Stamp commitTs = env.getSharedResources().getOracleClient().getStamp();
+      Assert.assertTrue(tx2.commitPrimaryColumn(cd, commitTs));
+    }
+
+    if (closeTransactor) {
+      t2.close();
+    }
+
+    return t2;
+  }
+
+  private void testBasicRollback(boolean closeTransactor) throws Exception {
+    try (Transaction tx = client.newTransaction()) {
+      setAlias(tx, "node1", "bob");
+      setAlias(tx, "node2", "joe");
+      setAlias(tx, "node3", "alice");
+      tx.commit();
+    }
+
+    try (Transaction tx = client.newTransaction()) {
+      addEdge(tx, "node1", "node2");
+      tx.commit();
+    }
+
+    TransactorNode tn =
+        partiallyCommit(tx -> addEdge(tx, "node1", "node3"), false, closeTransactor);
+
+    Assert.assertEquals(ImmutableSet.of("bob:joe", "joe:bob"), getDerivedEdges());
+
+    retryOnce(tx -> setAlias(tx, "node1", "bobby"));
+
+    Assert.assertEquals(ImmutableSet.of("bobby:joe", "joe:bobby"), getDerivedEdges());
+
+    retryOnce(tx -> setAlias(tx, "node3", "alex"));
+
+    Assert.assertEquals(ImmutableSet.of("bobby:joe", "joe:bobby"), getDerivedEdges());
+
+    if (!closeTransactor) {
+      tn.close();
+    }
+  }
+
+  @Test
+  public void testBasicRollback1() throws Exception {
+    testBasicRollback(true);
+  }
+
+  @Test
+  public void testBasicRollback2() throws Exception {
+    testBasicRollback(false);
+  }
+
+  private void testBasicRollforward(boolean closeTransactor) throws Exception {
+    try (Transaction tx = client.newTransaction()) {
+      setAlias(tx, "node1", "bob");
+      setAlias(tx, "node2", "joe");
+      setAlias(tx, "node3", "alice");
+      tx.commit();
+    }
+
+    try (Transaction tx = client.newTransaction()) {
+      addEdge(tx, "node1", "node2");
+      tx.commit();
+    }
+
+    TransactorNode tn = partiallyCommit(tx -> addEdge(tx, "node1", "node3"), true, closeTransactor);
+
+    retryOnce(tx -> setAlias(tx, "node1", "bobby"));
+
+    Assert.assertEquals(ImmutableSet.of("bobby:joe", "joe:bobby", "bobby:alice", "alice:bobby"),
+        getDerivedEdges());
+
+    retryOnce(tx -> setAlias(tx, "node3", "alex"));
+
+    Assert.assertEquals(ImmutableSet.of("bobby:joe", "joe:bobby", "bobby:alex", "alex:bobby"),
+        getDerivedEdges());
+
+    if (!closeTransactor) {
+      tn.close();
+    }
+  }
+
+  @Test
+  public void testBasicRollforward1() throws Exception {
+    testBasicRollforward(false);
+  }
+
+  @Test
+  public void testBasicRollforward2() throws Exception {
+    testBasicRollforward(true);
+  }
+
+  private void testParallelScan(boolean closeTransactor) throws Exception {
+    Column crCol = new Column("stat", "completionRatio");
+
+    try (Transaction tx = client.newTransaction()) {
+      tx.set("user5", crCol, "0.5");
+      tx.set("user6", crCol, "0.75");
+      tx.commit();
+    }
+
+    TransactorNode tn = partiallyCommit(tx -> {
+      // get multiple read locks with a parallel scan
+      Map<String, Map<Column, String>> ratios =
+          tx.withReadLock().gets(Arrays.asList("user5", "user6"), crCol);
+
+      double cr1 = Double.parseDouble(ratios.get("user5").get(crCol));
+      double cr2 = Double.parseDouble(ratios.get("user5").get(crCol));
+
+      tx.set("org1", crCol, (cr1 + cr2) / 2 + "");
+    }, false, closeTransactor);
+
+    retryTwice(tx -> {
+      Map<String, Map<Column, String>> ratios = tx.gets(Arrays.asList("user5", "user6"), crCol);
+
+      tx.set("user5", crCol, "0.51");
+      tx.set("user6", crCol, "0.76");
+    });
+
+    try (Snapshot snap = client.newSnapshot()) {
+      Assert.assertNull(snap.gets("org1", crCol));
+      Assert.assertEquals("0.51", snap.gets("user5", crCol));
+      Assert.assertEquals("0.76", snap.gets("user6", crCol));
+    }
+
+    if (!closeTransactor) {
+      tn.close();
+    }
+  }
+
+  @Test
+  public void testParallelScan1() throws Exception {
+    testParallelScan(true);
+  }
+
+  @Test
+  public void testParallelScan2() throws Exception {
+    testParallelScan(false);
+  }
+
+  private void testParallelScanRC(boolean closeTransactor) throws Exception {
+    // currently get w/ RowColumn has a different code path than other gets that take multiple rows
+    // and columns
+
+    Column crCol = new Column("stat", "completionRatio");
+
+    try (Transaction tx = client.newTransaction()) {
+      tx.set("user5", crCol, "0.5");
+      tx.set("user6", crCol, "0.75");
+      tx.commit();
+    }
+
+    TransactorNode tn = partiallyCommit(tx -> {
+      // get multiple read locks with a parallel scan
+      Map<RowColumn, String> ratios = tx.withReadLock()
+          .gets(Arrays.asList(new RowColumn("user5", crCol), new RowColumn("user6", crCol)));
+
+
+      double cr1 = Double.parseDouble(ratios.get(new RowColumn("user5", crCol)));
+      double cr2 = Double.parseDouble(ratios.get(new RowColumn("user6", crCol)));
+
+      tx.set("org1", crCol, (cr1 + cr2) / 2 + "");
+    }, false, true);
+
+    retryTwice(tx -> {
+      Map<RowColumn, String> ratios =
+          tx.gets(Arrays.asList(new RowColumn("user5", crCol), new RowColumn("user6", crCol)));
+
+      tx.set("user5", crCol, "0.51");
+      tx.set("user6", crCol, "0.76");
+    });
+
+    try (Snapshot snap = client.newSnapshot()) {
+      Assert.assertNull(snap.gets("org1", crCol));
+      Assert.assertEquals("0.51", snap.gets("user5", crCol));
+      Assert.assertEquals("0.76", snap.gets("user6", crCol));
+    }
+
+    if (!closeTransactor) {
+      tn.close();
+    }
+  }
+
+  @Test
+  public void testParallelScanRC1() throws Exception {
+    testParallelScanRC(true);
+  }
+
+  @Test
+  public void testParallelScanRC2() throws Exception {
+    testParallelScanRC(false);
+  }
+
+  private void testWriteWoRead(boolean commitPrimary, boolean closeTransactor) throws Exception {
+    // Reads can cause locks to be recovered. This test the case of a transactions that only does a
+    // write to a field that has an open read lock.
+
+    try (Transaction tx = client.newTransaction()) {
+      tx.set("r1", new Column("f1", "q1"), "v1");
+      tx.set("r2", new Column("f1", "q1"), "v2");
+      tx.commit();
+    }
+
+    TransactorNode transactor = partiallyCommit(tx -> {
+      String v1 = tx.withReadLock().gets("r1", new Column("f1", "q1"));
+      String v2 = tx.withReadLock().gets("r2", new Column("f1", "q1"));
+
+      tx.set("r3", new Column("f1", "qa"), v1 + ":" + v2);
+    }, commitPrimary, closeTransactor);
+
+    // TODO open an issue... does not really need to retry in this case
+    retryOnce(tx -> {
+      tx.set("r1", new Column("f1", "q1"), "v3");
+    });
+
+    try (Transaction tx = client.newTransaction()) {
+      if (commitPrimary) {
+        Assert.assertEquals("v1:v2", tx.gets("r3", new Column("f1", "qa")));
+      } else {
+        Assert.assertNull(tx.gets("r3", new Column("f1", "qa")));
+      }
+      Assert.assertEquals("v3", tx.gets("r1", new Column("f1", "q1")));
+    }
+
+    if (!closeTransactor) {
+      transactor.close();
+    }
+  }
+
+  @Test
+  public void testWriteWoRead1() throws Exception {
+    testWriteWoRead(false, false);
+  }
+
+  @Test
+  public void testWriteWoRead2() throws Exception {
+    testWriteWoRead(false, true);
+  }
+
+  @Test
+  public void testWriteWoRead3() throws Exception {
+    testWriteWoRead(true, false);
+  }
+
+  @Test
+  public void testWriteWoRead4() throws Exception {
+    testWriteWoRead(true, true);
+  }
+
+  private int countInTable(String str) throws TableNotFoundException {
+    int count = 0;
+    Scanner scanner = conn.createScanner(table, Authorizations.EMPTY);
+    for (String e : Iterables.transform(scanner, FluoFormatter::toString)) {
+      if (e.contains(str)) {
+        count++;
+      }
+    }
+
+    return count;
+  }
+
+  @Test
+  public void testFailDeletesReadLocks() throws Exception {
+    try (Transaction tx = client.newTransaction()) {
+      for (int i = 0; i < 20; i++) {
+        tx.set("r-" + i, new Column("f1", "q1"), "" + i);
+      }
+
+      tx.commit();
+    }
+
+    long startTs = 0;
+
+    try (Transaction tx1 = client.newTransaction()) {
+      tx1.set("r-5", new Column("f1", "q1"), "9");
+      try (Transaction tx2 = client.newTransaction()) {
+        tx1.commit();
+
+        int sum = 0;
+        for (int i = 0; i < 20; i++) {
+          sum += Integer.parseInt(tx2.withReadLock().gets("r-" + i, new Column("f1", "q1")));
+        }
+
+        tx2.set("sum1", new Column("f", "s"), "" + sum);
+        startTs = tx2.getStartTimestamp();
+        tx2.commit();
+        Assert.fail();
+      } catch (CommitException e) {
+
+      }
+    }
+
+    try (Snapshot snapshot = client.newSnapshot()) {
+      Assert.assertNull(snapshot.gets("sum1", new Column("f", "s")));
+    }
+
+    // ensure the failed tx deleted its read locks....
+    Assert.assertEquals(19, countInTable(startTs + "-RLOCK"));
+    Assert.assertEquals(19, countInTable(startTs + "-DEL_RLOCK"));
+
+    try (Transaction tx = client.newTransaction()) {
+      int sum = 0;
+      for (int i = 0; i < 20; i++) {
+        sum += Integer.parseInt(tx.withReadLock().gets("r-" + i, new Column("f1", "q1")));
+      }
+
+      tx.set("sum1", new Column("f", "s"), "" + sum);
+      tx.commit();
+    }
+
+    try (Snapshot snapshot = client.newSnapshot()) {
+      Assert.assertEquals("194", snapshot.gets("sum1", new Column("f", "s")));
+    }
+  }
+}
diff --git a/modules/integration/src/test/java/org/apache/fluo/integration/impl/ReadLockIT.java b/modules/integration/src/test/java/org/apache/fluo/integration/impl/ReadLockIT.java
new file mode 100644
index 0000000..b88d66e
--- /dev/null
+++ b/modules/integration/src/test/java/org/apache/fluo/integration/impl/ReadLockIT.java
@@ -0,0 +1,548 @@
+/*
+ * 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.fluo.integration.impl;
+
+import java.io.BufferedWriter;
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Random;
+import java.util.Set;
+import java.util.function.Consumer;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.Maps;
+import com.google.common.collect.Sets;
+import org.apache.accumulo.core.client.Scanner;
+import org.apache.accumulo.core.client.TableNotFoundException;
+import org.apache.accumulo.core.data.Key;
+import org.apache.accumulo.core.data.Range;
+import org.apache.accumulo.core.data.Value;
+import org.apache.accumulo.core.security.Authorizations;
+import org.apache.fluo.accumulo.format.FluoFormatter;
+import org.apache.fluo.api.client.FluoClient;
+import org.apache.fluo.api.client.FluoFactory;
+import org.apache.fluo.api.client.Loader;
+import org.apache.fluo.api.client.LoaderExecutor;
+import org.apache.fluo.api.client.Snapshot;
+import org.apache.fluo.api.client.SnapshotBase;
+import org.apache.fluo.api.client.Transaction;
+import org.apache.fluo.api.client.TransactionBase;
+import org.apache.fluo.api.client.scanner.CellScanner;
+import org.apache.fluo.api.config.FluoConfiguration;
+import org.apache.fluo.api.data.Column;
+import org.apache.fluo.api.data.RowColumn;
+import org.apache.fluo.api.data.RowColumnValue;
+import org.apache.fluo.api.data.Span;
+import org.apache.fluo.api.exceptions.AlreadySetException;
+import org.apache.fluo.api.exceptions.CommitException;
+import org.apache.fluo.integration.ITBaseImpl;
+import org.apache.fluo.integration.TestTransaction;
+import org.junit.Assert;
+import org.junit.Test;
+
+import static java.util.Arrays.asList;
+
+public class ReadLockIT extends ITBaseImpl {
+
+  private static final Column ALIAS_COL = new Column("node", "alias");
+
+  private void addEdge(String node1, String node2) {
+    try (Transaction tx = client.newTransaction()) {
+      addEdge(tx, node1, node2);
+      tx.commit();
+    }
+  }
+
+  static void addEdge(TransactionBase tx, String node1, String node2) {
+    Map<String, Map<Column, String>> aliases =
+        tx.withReadLock().gets(asList("r:" + node1, "r:" + node2), ALIAS_COL);
+    String alias1 = aliases.get("r:" + node1).get(ALIAS_COL);
+    String alias2 = aliases.get("r:" + node2).get(ALIAS_COL);
+
+    addEdge(tx, node1, node2, alias1, alias2);
+  }
+
+  static void addEdge(TransactionBase tx, String node1, String node2, String alias1,
+      String alias2) {
+    tx.set("d:" + alias1 + ":" + alias2, new Column("edge", node1 + ":" + node2), "");
+    tx.set("d:" + alias2 + ":" + alias1, new Column("edge", node2 + ":" + node1), "");
+
+    tx.set("r:" + node1 + ":" + node2, new Column("edge", "aliases"), alias1 + ":" + alias2);
+    tx.set("r:" + node2 + ":" + node1, new Column("edge", "aliases"), alias2 + ":" + alias1);
+  }
+
+  static void setAlias(TransactionBase tx, String node, String alias) {
+    tx.set("r:" + node, new Column("node", "alias"), alias);
+
+    CellScanner scanner = tx.scanner().over(Span.prefix("r:" + node + ":")).build();
+
+    for (RowColumnValue rcv : scanner) {
+      String otherNode = rcv.getsRow().split(":")[2];
+      String[] aliases = rcv.getsValue().split(":");
+
+      if (aliases.length != 2) {
+        throw new RuntimeException("bad alias " + rcv);
+      }
+
+      if (!alias.equals(aliases[0])) {
+        tx.delete("d:" + aliases[0] + ":" + aliases[1], new Column("edge", node + ":" + otherNode));
+        tx.delete("d:" + aliases[1] + ":" + aliases[0], new Column("edge", otherNode + ":" + node));
+
+        addEdge(tx, node, otherNode, alias, aliases[1]);
+      }
+    }
+  }
+
+  @Test
+  public void testConcurrentReadlocks() throws Exception {
+
+    try (Transaction tx = client.newTransaction()) {
+      setAlias(tx, "node1", "bob");
+      setAlias(tx, "node2", "joe");
+      setAlias(tx, "node3", "alice");
+      tx.commit();
+    }
+
+
+    TestTransaction tx1 = new TestTransaction(env);
+    setAlias(tx1, "node2", "jojo");
+
+    TestTransaction tx2 = new TestTransaction(env);
+    TestTransaction tx3 = new TestTransaction(env);
+
+    addEdge(tx2, "node1", "node2");
+    addEdge(tx3, "node1", "node3");
+
+    tx2.commit();
+    tx2.close();
+
+    tx3.commit();
+    tx3.close();
+
+    Assert.assertEquals(ImmutableSet.of("bob:joe", "joe:bob", "bob:alice", "alice:bob"),
+        getDerivedEdges());
+
+    try {
+      tx1.commit();
+      Assert.fail("Expected exception");
+    } catch (CommitException ce) {
+    }
+
+    Assert.assertEquals(ImmutableSet.of("bob:joe", "joe:bob", "bob:alice", "alice:bob"),
+        getDerivedEdges());
+  }
+
+  @Test
+  public void testWriteCausesReadLockToFail() throws Exception {
+    try (Transaction tx = client.newTransaction()) {
+      setAlias(tx, "node1", "bob");
+      setAlias(tx, "node2", "joe");
+      tx.commit();
+    }
+
+    TestTransaction tx1 = new TestTransaction(env);
+    setAlias(tx1, "node2", "jojo");
+
+    TestTransaction tx2 = new TestTransaction(env);
+
+    addEdge(tx2, "node1", "node2");
+
+    tx1.commit();
+    tx1.close();
+
+    Assert.assertEquals(0, getDerivedEdges().size());
+
+    try {
+      tx2.commit();
+      Assert.fail("Expected exception");
+    } catch (CommitException ce) {
+    }
+
+    // ensure the failed read lock on node1 is cleaned up
+    try (Transaction tx = client.newTransaction()) {
+      setAlias(tx, "node1", "fred");
+      tx.commit();
+    }
+
+    try (Transaction tx = client.newTransaction()) {
+      addEdge(tx, "node1", "node2");
+      tx.commit();
+    }
+
+    Assert.assertEquals(ImmutableSet.of("fred:jojo", "jojo:fred"), getDerivedEdges());
+  }
+
+  private void dumpRow(String row, Consumer<String> out) throws TableNotFoundException {
+    Scanner scanner = conn.createScanner(getCurTableName(), Authorizations.EMPTY);
+    scanner.setRange(Range.exact(row));
+    for (Entry<Key, Value> entry : scanner) {
+      out.accept(FluoFormatter.toString(entry));
+    }
+  }
+
+  private void dumpTable(Consumer<String> out) throws TableNotFoundException {
+    Scanner scanner = conn.createScanner(getCurTableName(), Authorizations.EMPTY);
+    for (Entry<Key, Value> entry : scanner) {
+      out.accept(FluoFormatter.toString(entry));
+    }
+  }
+
+  @Test
+  public void testWriteAfterReadLock() throws Exception {
+    try (Transaction tx = client.newTransaction()) {
+      setAlias(tx, "node1", "bob");
+      setAlias(tx, "node2", "joe");
+      setAlias(tx, "node3", "alice");
+
+      tx.commit();
+    }
+
+    addEdge("node1", "node2");
+    Assert.assertEquals(ImmutableSet.of("bob:joe", "joe:bob"), getDerivedEdges());
+
+    try (Transaction tx = client.newTransaction()) {
+      setAlias(tx, "node1", "bobby");
+      tx.commit();
+    }
+
+    addEdge("node1", "node3");
+    Assert.assertEquals(ImmutableSet.of("bobby:joe", "joe:bobby", "bobby:alice", "alice:bobby"),
+        getDerivedEdges());
+  }
+
+  @Test
+  public void testRandom() throws Exception {
+    int numNodes = 100;
+    int numEdges = 1000;
+    int numAliasChanges = 25;
+
+    Random rand = new Random();
+
+    Map<String, String> nodes = new HashMap<>();
+    while (nodes.size() < numNodes) {
+      nodes.put(String.format("n-%09d", rand.nextInt(1000000000)),
+          String.format("a-%09d", rand.nextInt(1000000000)));
+    }
+
+    List<String> nodesList = new ArrayList<>(nodes.keySet());
+    Set<String> edges = new HashSet<>();
+    while (edges.size() < numEdges) {
+      String n1 = nodesList.get(rand.nextInt(nodesList.size()));
+      String n2 = nodesList.get(rand.nextInt(nodesList.size()));
+      if (n1.equals(n2) || edges.contains(n2 + ":" + n1)) {
+        continue;
+      }
+
+      edges.add(n1 + ":" + n2);
+    }
+
+    try (LoaderExecutor le = client.newLoaderExecutor()) {
+      for (Entry<String, String> entry : nodes.entrySet()) {
+        le.execute((tx, ctx) -> setAlias(tx, entry.getKey(), entry.getValue()));
+      }
+    }
+
+    List<Loader> loadOps = new ArrayList<>();
+    for (String edge : edges) {
+      String[] enodes = edge.split(":");
+      loadOps.add((tx, ctx) -> {
+        try {
+          addEdge(tx, enodes[0], enodes[1]);
+        } catch (NullPointerException e) {
+          // TOOD remove after finding bug
+          System.out.println(
+              " en0 " + enodes[0] + " en1 " + enodes[1] + " start ts " + tx.getStartTimestamp());
+          dumpRow("r:" + enodes[0], System.out::println);
+          dumpRow("r:" + enodes[1], System.out::println);
+          throw e;
+        }
+      });
+    }
+
+    Map<String, String> changes = new HashMap<>();
+    while (changes.size() < numAliasChanges) {
+      String node = nodesList.get(rand.nextInt(nodesList.size()));
+      String alias = String.format("a-%09d", rand.nextInt(1000000000));
+      changes.put(node, alias);
+    }
+
+    Map<String, String> nodes2 = new HashMap<>(nodes);
+    nodes2.putAll(changes);
+
+    changes.forEach((node, alias) -> {
+      loadOps.add((tx, ctx) -> setAlias(tx, node, alias));
+    });
+
+    Collections.shuffle(loadOps, rand);
+
+    FluoConfiguration conf = new FluoConfiguration(config);
+    conf.setLoaderThreads(20);
+    try (FluoClient client = FluoFactory.newClient(conf);
+        LoaderExecutor le = client.newLoaderExecutor()) {
+      loadOps.forEach(loader -> le.execute(loader));
+    }
+
+    Set<String> expectedEdges = new HashSet<>();
+    for (String edge : edges) {
+      String[] enodes = edge.split(":");
+      String alias1 = nodes2.get(enodes[0]);
+      String alias2 = nodes2.get(enodes[1]);
+
+      expectedEdges.add(alias1 + ":" + alias2);
+      expectedEdges.add(alias2 + ":" + alias1);
+    }
+
+    Set<String> actualEdges = getDerivedEdges();
+
+    if (!expectedEdges.equals(actualEdges)) {
+      Path dumpFile = Paths.get("target/ReadLockIT.txt");
+
+      try (BufferedWriter writer = Files.newBufferedWriter(dumpFile)) {
+
+        Consumer<String> out = s -> {
+          try {
+            writer.append(s);
+            writer.append("\n");
+          } catch (IOException e) {
+            throw new UncheckedIOException(e);
+          }
+        };
+
+
+        writer.append("Alias changes : \n");
+        Maps.difference(nodes, nodes2).entriesDiffering()
+            .forEach((k, v) -> out.accept(k + " " + v));
+
+        writer.append("expected - actual : \n");
+        Sets.difference(expectedEdges, actualEdges).forEach(out);
+        writer.append("\n");
+
+        writer.append("actual - expected : \n");
+        Sets.difference(actualEdges, expectedEdges).forEach(out);
+        writer.append("\n");
+
+        printSnapshot(out);
+        dumpTable(out);
+      }
+      Assert.fail("Did not produce expected graph, dumped info to " + dumpFile);
+    }
+  }
+
+  private Set<String> getDerivedEdges() {
+    Set<String> derivedEdges = new HashSet<>();
+    try (Snapshot snap = client.newSnapshot()) {
+      snap.scanner().over(Span.prefix("d:")).build().stream().map(RowColumnValue::getsRow)
+          .map(r -> r.substring(2)).forEach(derivedEdges::add);
+    }
+    return derivedEdges;
+  }
+
+  @Test(expected = AlreadySetException.class)
+  public void testReadAndWriteLockInSameTx() {
+    try (Transaction tx = client.newTransaction()) {
+      setAlias(tx, "node1", "bob");
+      setAlias(tx, "node2", "joe");
+
+
+      tx.commit();
+    }
+
+    try (Transaction tx = client.newTransaction()) {
+      setAlias(tx, "node1", "bobby");
+      // tries to get a read lock on node1 in same tx
+      addEdge(tx, "node1", "node2");
+    }
+  }
+
+  @Test(expected = AlreadySetException.class)
+  public void testReadAndDeleteInSameTx() {
+    try (Transaction tx = client.newTransaction()) {
+      tx.set("123456", new Column("f", "q"), "abc");
+      tx.commit();
+    }
+
+    try (Transaction tx = client.newTransaction()) {
+      tx.delete("123456", new Column("f", "q"));
+      // should fail here because already have write lock
+      String val = tx.withReadLock().gets("123456", new Column("f", "q"));
+      tx.set("123457", new Column("f", "q"), val + "7");
+      tx.commit();
+    }
+  }
+
+  private static final Column c1 = new Column("f1", "q1");
+  private static final Column c2 = new Column("f1", "q2");
+  private static final Column invCol = new Column("f1", "inv");
+
+  final private void ensureReadLocksSet(Consumer<TransactionBase> readLockOperation) {
+
+    try (Transaction txi = client.newTransaction()) {
+      txi.set("test1", c1, "45");
+      txi.set("test1", c2, "90");
+      txi.set("test2", c1, "30");
+      txi.set("test2", c2, "60");
+      txi.commit();
+    }
+
+
+    List<Consumer<TransactionBase>> writeLockOperations = ImmutableList.of(txw -> {
+      txw.set("test1", c1, "47");
+    }, txw -> {
+      txw.set("test1", c2, "94");
+    }, txw -> {
+      txw.set("test2", c1, "37");
+    }, txw -> {
+      txw.set("test2", c2, "74");
+    });
+
+    List<Transaction> writeTxs = new ArrayList<>();
+    for (Consumer<TransactionBase> wop : writeLockOperations) {
+      Transaction wtx = client.newTransaction();
+      wop.accept(wtx);
+      writeTxs.add(wtx);
+    }
+
+    try (Transaction txr = client.newTransaction()) {
+      readLockOperation.accept(txr);
+      txr.commit();
+    }
+
+    for (Transaction wtx : writeTxs) {
+      try {
+        wtx.commit();
+        Assert.fail();
+      } catch (CommitException ce) {
+      }
+    }
+
+    try (Snapshot snap = client.newSnapshot()) {
+      Assert.assertEquals("45", snap.gets("test1", c1));
+      Assert.assertEquals("90", snap.gets("test1", c2));
+      Assert.assertEquals("test1", snap.gets("45", invCol));
+      Assert.assertEquals("test1", snap.gets("90", invCol));
+      Assert.assertEquals("30", snap.gets("test2", c1));
+      Assert.assertEquals("60", snap.gets("test2", c2));
+      Assert.assertEquals("test2", snap.gets("30", invCol));
+      Assert.assertEquals("test2", snap.gets("60", invCol));
+    }
+  }
+
+  @Test
+  public void testGet() {
+    ensureReadLocksSet(txr -> {
+      // ensure this operation sets two read locks
+      SnapshotBase rlSnap = txr.withReadLock();
+
+      txr.set(rlSnap.gets("test1", c1), invCol, "test1");
+      txr.set(rlSnap.gets("test1", c2), invCol, "test1");
+      txr.set(rlSnap.gets("test2", c1), invCol, "test2");
+      txr.set(rlSnap.gets("test2", c2), invCol, "test2");
+    });
+  }
+
+  @Test
+  public void testGetColumns() {
+    ensureReadLocksSet(txr -> {
+      // ensure this operation sets two read locks
+      Map<Column, String> vals = txr.withReadLock().gets("test1", c1, c2);
+      txr.set(vals.get(c1), invCol, "test1");
+      txr.set(vals.get(c2), invCol, "test1");
+      vals = txr.withReadLock().gets("test2", c1, c2);
+      txr.set(vals.get(c1), invCol, "test2");
+      txr.set(vals.get(c2), invCol, "test2");
+    });
+  }
+
+  @Test
+  public void testGetRowsColumns() {
+    ensureReadLocksSet(txr -> {
+      // ensure this operation sets two read locks
+      Map<String, Map<Column, String>> vals =
+          txr.withReadLock().gets(ImmutableList.of("test1", "test2"), c1, c2);
+      txr.set(vals.get("test1").get(c1), invCol, "test1");
+      txr.set(vals.get("test1").get(c2), invCol, "test1");
+      txr.set(vals.get("test2").get(c1), invCol, "test2");
+      txr.set(vals.get("test2").get(c2), invCol, "test2");
+    });
+  }
+
+  @Test
+  public void testGetRowColumns() {
+    ensureReadLocksSet(txr -> {
+      // ensure this operation sets two read locks
+      Map<RowColumn, String> vals =
+          txr.withReadLock().gets(ImmutableList.of(new RowColumn("test1", c1),
+              new RowColumn("test1", c2), new RowColumn("test2", c1), new RowColumn("test2", c2)));
+      txr.set(vals.get(new RowColumn("test1", c1)), invCol, "test1");
+      txr.set(vals.get(new RowColumn("test1", c2)), invCol, "test1");
+      txr.set(vals.get(new RowColumn("test2", c1)), invCol, "test2");
+      txr.set(vals.get(new RowColumn("test2", c2)), invCol, "test2");
+    });
+  }
+
+  @Test(expected = UnsupportedOperationException.class)
+  public void testScan() {
+    try (Transaction tx = client.newTransaction()) {
+      tx.withReadLock().scanner().build();
+    }
+  }
+
+  @Test
+  public void testOnlyReadLocks() {
+    try (Transaction tx = client.newTransaction()) {
+      tx.set("r1", new Column("q1", "f1"), "v1");
+      tx.set("r2", new Column("q1", "f1"), "v2");
+      tx.commit();
+    }
+
+    try (Transaction tx1 = client.newTransaction()) {
+      try (Transaction tx2 = client.newTransaction()) {
+        String v1 = tx2.withReadLock().gets("r1", new Column("q1", "f1"));
+        String v2 = tx2.withReadLock().gets("r2", new Column("q1", "f1"));
+
+        Assert.assertEquals("v1", v1);
+        Assert.assertEquals("v2", v2);
+
+        // commit should be a no-op because only read locks
+        tx2.commit();
+      }
+
+      tx1.set("r1", new Column("q1", "f1"), "v3");
+      tx1.set("r2", new Column("q1", "f1"), "v4");
+
+      // should not collide with read locks
+      tx1.commit();
+    }
+    try (Snapshot snap = client.newSnapshot()) {
+      String v1 = snap.gets("r1", new Column("q1", "f1"));
+      String v2 = snap.gets("r2", new Column("q1", "f1"));
+
+      Assert.assertEquals("v3", v1);
+      Assert.assertEquals("v4", v2);
+    }
+  }
+}
diff --git a/modules/integration/src/test/java/org/apache/fluo/integration/impl/ScannerIT.java b/modules/integration/src/test/java/org/apache/fluo/integration/impl/ScannerIT.java
index e5a8d02..0a6bc99 100644
--- a/modules/integration/src/test/java/org/apache/fluo/integration/impl/ScannerIT.java
+++ b/modules/integration/src/test/java/org/apache/fluo/integration/impl/ScannerIT.java
@@ -70,7 +70,7 @@ public class ScannerIT extends ITBaseImpl {
 
     try (Snapshot snap = client.newSnapshot()) {
       HashSet<RowColumnValue> actual = new HashSet<>();
-      Iterables.addAll(actual, snap.scanner().over("r2").build());
+      Iterables.addAll(actual, snap.scanner().over(Span.exact("r2")).build());
       Assert.assertEquals(expectedR2, actual);
 
       actual.clear();
diff --git a/modules/integration/src/test/java/org/apache/fluo/integration/log/LogIT.java b/modules/integration/src/test/java/org/apache/fluo/integration/log/LogIT.java
index 1736248..0835310 100644
--- a/modules/integration/src/test/java/org/apache/fluo/integration/log/LogIT.java
+++ b/modules/integration/src/test/java/org/apache/fluo/integration/log/LogIT.java
@@ -18,6 +18,7 @@ package org.apache.fluo.integration.log;
 import java.io.StringWriter;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.HashMap;
 import java.util.Map;
 
 import com.google.common.collect.ImmutableMap;
@@ -592,4 +593,72 @@ public class LogIT extends ITBaseMini {
     Assert.assertTrue(origLogMsgs, logMsgs.matches(pattern));
 
   }
+
+  @Test
+  public void testReadLocks() {
+    Column c1 = new Column("f1", "q1");
+    Column c2 = new Column("f1", "q2");
+
+    try (Transaction tx = client.newTransaction()) {
+      tx.set("r1", c1, "v1");
+      tx.set("r1", c2, "v2");
+      tx.set("r2", c1, "v3");
+      tx.set("r2", c2, "v4");
+      tx.commit();
+    }
+
+    Logger logger = Logger.getLogger("fluo.tx");
+
+    StringWriter writer = new StringWriter();
+    WriterAppender appender =
+        new WriterAppender(new PatternLayout("%d{ISO8601} [%-8c{2}] %-5p: %m%n"), writer);
+
+    Level level = logger.getLevel();
+    boolean additivity = logger.getAdditivity();
+
+    try {
+      logger.setLevel(Level.TRACE);
+      logger.setAdditivity(false);
+      logger.addAppender(appender);
+
+      try (Transaction tx = client.newTransaction()) {
+        Assert.assertEquals("v1", tx.withReadLock().gets("r1", c1));
+        Assert.assertEquals(ImmutableMap.of(c1, "v3", c2, "v4"),
+            tx.withReadLock().gets("r2", c1, c2));
+        Assert.assertEquals(ImmutableMap.of(new RowColumn("r1", c2), "v2"),
+            tx.withReadLock().gets(Arrays.asList(new RowColumn("r1", c2))));
+        Map<String, Map<Column, String>> expected = new HashMap<>();
+        expected.computeIfAbsent("r1", k -> new HashMap<>()).put(c1, "v1");
+        expected.computeIfAbsent("r2", k -> new HashMap<>()).put(c1, "v3");
+        Map<String, Map<Column, String>> actual =
+            tx.withReadLock().gets(Arrays.asList("r1", "r2"), ImmutableSet.of(c1));
+        Assert.assertEquals(expected, actual);
+        tx.set("r3", c1, "345");
+        tx.commit();
+      }
+
+    } finally {
+      logger.removeAppender(appender);
+      logger.setAdditivity(additivity);
+      logger.setLevel(level);
+    }
+
+    String origLogMsgs = writer.toString();
+    String logMsgs = origLogMsgs.replace('\n', ' ');
+
+    String pattern = "";
+
+    pattern += ".*txid: (\\d+) begin\\(\\) thread: \\d+";
+    pattern += ".*txid: \\1 \\QwithReadLock().get(r1, f1 q1 ) -> v1\\E";
+    pattern +=
+        ".*txid: \\1 \\QwithReadLock().get(r2, [f1 q1 , f1 q2 ]) -> [f1 q1 =v3, f1 q2 =v4]\\E";
+    pattern += ".*txid: \\1 \\QwithReadLock().get([r1 f1 q2 ]) -> [r1 f1 q2 =v2]\\E";
+    pattern +=
+        ".*txid: \\1 \\QwithReadLock().get([r1, r2], [f1 q1 ]) -> [r1=[f1 q1 =v1], r2=[f1 q1 =v3]]\\E";
+    pattern += ".*txid: \\1 \\Qset(r3, f1 q1 , 345)\\E";
+    pattern += ".*txid: \\1 \\Qcommit()\\E -> SUCCESSFUL commitTs: \\d+";
+    pattern += ".*txid: \\1 \\Qclose()\\E.*";
+
+    Assert.assertTrue(origLogMsgs, logMsgs.matches(pattern));
+  }
 }

-- 
To stop receiving notification emails like this one, please contact
['"commits@fluo.apache.org" <commits@fluo.apache.org>'].

Mime
View raw message