accumulo-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ibe...@apache.org
Subject accumulo git commit: ACCUMULO-4643 Allow iterators to yield
Date Thu, 20 Jul 2017 17:43:31 GMT
Repository: accumulo
Updated Branches:
  refs/heads/1.8 a5ed1ba3a -> 1ab2431c9


ACCUMULO-4643 Allow iterators to yield

Implemented a mechanism to allow iterators to yield control allowing other scans to use the
scan thread.
Added requirement for hasTop to return false after yielding.

closes #263


Project: http://git-wip-us.apache.org/repos/asf/accumulo/repo
Commit: http://git-wip-us.apache.org/repos/asf/accumulo/commit/1ab2431c
Tree: http://git-wip-us.apache.org/repos/asf/accumulo/tree/1ab2431c
Diff: http://git-wip-us.apache.org/repos/asf/accumulo/diff/1ab2431c

Branch: refs/heads/1.8
Commit: 1ab2431c96b64e90350ba54045a774a0f1e25404
Parents: a5ed1ba
Author: Ivan Bella <ivan@bella.name>
Authored: Thu May 25 14:22:24 2017 -0400
Committer: Ivan Bella <ivan@bella.name>
Committed: Thu Jul 20 11:19:52 2017 -0400

----------------------------------------------------------------------
 .../core/iterators/SortedKeyValueIterator.java  |   3 +-
 .../accumulo/core/iterators/YieldCallback.java  |  57 +++++++
 .../iterators/YieldingKeyValueIterator.java     |  36 +++++
 .../system/SourceSwitchingIterator.java         |  49 +++++-
 .../system/SourceSwitchingIteratorTest.java     | 112 +++++++++++++
 .../testcases/YieldingTestCase.java             |  87 ++++++++++
 .../apache/accumulo/tserver/TabletServer.java   |   4 +
 .../Metrics2TabletServerScanMetrics.java        |   5 +-
 .../metrics/TabletServerScanMetricsKeys.java    |   1 +
 .../apache/accumulo/tserver/tablet/Scanner.java |   2 +-
 .../apache/accumulo/tserver/tablet/Tablet.java  |  77 +++++++--
 .../apache/accumulo/test/YieldScannersIT.java   | 161 +++++++++++++++++++
 .../test/functional/YieldingIterator.java       | 129 +++++++++++++++
 13 files changed, 702 insertions(+), 21 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/accumulo/blob/1ab2431c/core/src/main/java/org/apache/accumulo/core/iterators/SortedKeyValueIterator.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/iterators/SortedKeyValueIterator.java
b/core/src/main/java/org/apache/accumulo/core/iterators/SortedKeyValueIterator.java
index ce5ef24..38158ff 100644
--- a/core/src/main/java/org/apache/accumulo/core/iterators/SortedKeyValueIterator.java
+++ b/core/src/main/java/org/apache/accumulo/core/iterators/SortedKeyValueIterator.java
@@ -50,7 +50,8 @@ public interface SortedKeyValueIterator<K extends WritableComparable<?>,V
extend
   void init(SortedKeyValueIterator<K,V> source, Map<String,String> options, IteratorEnvironment
env) throws IOException;
 
   /**
-   * Returns true if the iterator has more elements.
+   * Returns true if the iterator has more elements. Note that if this iterator has yielded
(@see enableYielding(YieldCallback)), this this method must return
+   * false.
    *
    * @return <tt>true</tt> if the iterator has more elements.
    * @exception IllegalStateException

http://git-wip-us.apache.org/repos/asf/accumulo/blob/1ab2431c/core/src/main/java/org/apache/accumulo/core/iterators/YieldCallback.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/iterators/YieldCallback.java b/core/src/main/java/org/apache/accumulo/core/iterators/YieldCallback.java
new file mode 100644
index 0000000..3d151cf
--- /dev/null
+++ b/core/src/main/java/org/apache/accumulo/core/iterators/YieldCallback.java
@@ -0,0 +1,57 @@
+/*
+ * 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.accumulo.core.iterators;
+
+/**
+ * This callback handles the state of yielding within an iterator
+ */
+public class YieldCallback<K> {
+  private K key;
+
+  /**
+   * Called by the iterator when a next or seek call yields control.
+   *
+   * @param key
+   *          the key position at which the iterator yielded.
+   */
+  public void yield(K key) {
+    this.key = key;
+  }
+
+  /**
+   * Called by the client to see if the iterator yielded
+   *
+   * @return true if iterator yielded control
+   */
+  public boolean hasYielded() {
+    return (this.key != null);
+  }
+
+  /**
+   * Called by the client to get the yield position used as the start key (non-inclusive)
of the range in a subsequent seek call when the iterator is rebuilt.
+   * This will also reset the state returned by hasYielded.
+   *
+   * @return <tt>K</tt> The key position
+   */
+  public K getPositionAndReset() {
+    try {
+      return this.key;
+    } finally {
+      this.key = null;
+    }
+  }
+}

http://git-wip-us.apache.org/repos/asf/accumulo/blob/1ab2431c/core/src/main/java/org/apache/accumulo/core/iterators/YieldingKeyValueIterator.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/iterators/YieldingKeyValueIterator.java
b/core/src/main/java/org/apache/accumulo/core/iterators/YieldingKeyValueIterator.java
new file mode 100644
index 0000000..76f8f31
--- /dev/null
+++ b/core/src/main/java/org/apache/accumulo/core/iterators/YieldingKeyValueIterator.java
@@ -0,0 +1,36 @@
+/*
+ * 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.accumulo.core.iterators;
+
+import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.io.WritableComparable;
+
+/**
+ * An iterator that supports iterating over key and value pairs, and supports yielding on
a next or seek call.
+ */
+public interface YieldingKeyValueIterator<K extends WritableComparable<?>,V extends
Writable> extends SortedKeyValueIterator<K,V> {
+
+  /**
+   * Allows implementations to preempt further iteration of this iterator in the current
RPC. Implementations can use the yield method on the callback to
+   * instruct the caller to cease collecting more results within this RPC. An implementation
would only need to implement this mechanism if a next or seek call
+   * has been taking so long as to starve out other scans within the same thread pool. Most
iterators do not need to implement this method. The yield method on
+   * the callback accepts a Key which will be used as the start key (non-inclusive) on the
seek call in the next RPC. This feature is not supported for isolated
+   * scans.
+   */
+  void enableYielding(YieldCallback<K> callback);
+
+}

http://git-wip-us.apache.org/repos/asf/accumulo/blob/1ab2431c/core/src/main/java/org/apache/accumulo/core/iterators/system/SourceSwitchingIterator.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/iterators/system/SourceSwitchingIterator.java
b/core/src/main/java/org/apache/accumulo/core/iterators/system/SourceSwitchingIterator.java
index 098aa63..9cb3bd7 100644
--- a/core/src/main/java/org/apache/accumulo/core/iterators/system/SourceSwitchingIterator.java
+++ b/core/src/main/java/org/apache/accumulo/core/iterators/system/SourceSwitchingIterator.java
@@ -23,6 +23,7 @@ import java.util.List;
 import java.util.Map;
 import java.util.concurrent.atomic.AtomicBoolean;
 
+import com.google.common.base.Optional;
 import org.apache.accumulo.core.data.ByteSequence;
 import org.apache.accumulo.core.data.Key;
 import org.apache.accumulo.core.data.PartialKey;
@@ -30,13 +31,15 @@ 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.accumulo.core.iterators.YieldCallback;
+import org.apache.accumulo.core.iterators.YieldingKeyValueIterator;
 
 /**
  * A SortedKeyValueIterator which presents a view over some section of data, regardless of
whether or not it is backed by memory (InMemoryMap) or an RFile
  * (InMemoryMap that was minor compacted to a file). Clients reading from a table that has
data in memory should not see interruption in their scan when that
  * data is minor compacted. This iterator is designed to manage this behind the scene.
  */
-public class SourceSwitchingIterator implements InterruptibleIterator {
+public class SourceSwitchingIterator implements InterruptibleIterator, YieldingKeyValueIterator<Key,Value>
{
 
   public interface DataSource {
     boolean isCurrent();
@@ -53,6 +56,8 @@ public class SourceSwitchingIterator implements InterruptibleIterator {
   private DataSource source;
   private SortedKeyValueIterator<Key,Value> iter;
 
+  private Optional<YieldCallback<Key>> yield = Optional.absent();
+
   private Key key;
   private Value val;
 
@@ -113,6 +118,18 @@ public class SourceSwitchingIterator implements InterruptibleIterator
{
   }
 
   @Override
+  public void enableYielding(YieldCallback<Key> yield) {
+    this.yield = Optional.of(yield);
+
+    // if we require row isolation, then we cannot support yielding in the middle.
+    if (!onlySwitchAfterRow) {
+      if (iter != null && iter instanceof YieldingKeyValueIterator) {
+        ((YieldingKeyValueIterator<Key,Value>) iter).enableYielding(yield);
+      }
+    }
+  }
+
+  @Override
   public void init(SortedKeyValueIterator<Key,Value> source, Map<String,String>
options, IteratorEnvironment env) throws IOException {
     throw new UnsupportedOperationException();
   }
@@ -126,14 +143,23 @@ public class SourceSwitchingIterator implements InterruptibleIterator
{
 
   private void readNext(boolean initialSeek) throws IOException {
 
+    // we need to check here if we were yielded in case the source was switched out and re-seeked
by someone else (minor compaction/InMemoryMap)
+    boolean yielded = (yield.isPresent() && yield.get().hasYielded());
+
     // check of initialSeek second is intentional so that it does not short
     // circuit the call to switchSource
-    boolean seekNeeded = (!onlySwitchAfterRow && switchSource()) || initialSeek;
+    boolean seekNeeded = yielded || (!onlySwitchAfterRow && switchSource()) || initialSeek;
 
     if (seekNeeded)
       if (initialSeek)
         iter.seek(range, columnFamilies, inclusive);
-      else
+      else if (yielded) {
+        Key yieldPosition = yield.get().getPositionAndReset();
+        if (!range.contains(yieldPosition)) {
+          throw new IOException("Underlying iterator yielded to a position outside of its
range: " + yieldPosition + " not in " + range);
+        }
+        iter.seek(new Range(yieldPosition, false, range.getEndKey(), range.isEndKeyInclusive()),
columnFamilies, inclusive);
+      } else
         iter.seek(new Range(key, false, range.getEndKey(), range.isEndKeyInclusive()), columnFamilies,
inclusive);
     else {
       iter.next();
@@ -144,6 +170,10 @@ public class SourceSwitchingIterator implements InterruptibleIterator
{
     }
 
     if (iter.hasTop()) {
+      if (yield.isPresent() && yield.get().hasYielded()) {
+        throw new IOException("Coding error: hasTop returned true but has yielded at " +
yield.get().getPositionAndReset());
+      }
+
       Key nextKey = iter.getTopKey();
       Value nextVal = iter.getTopValue();
 
@@ -163,6 +193,11 @@ public class SourceSwitchingIterator implements InterruptibleIterator
{
     if (!source.isCurrent()) {
       source = source.getNewDataSource();
       iter = source.iterator();
+      if (!onlySwitchAfterRow && yield.isPresent()) {
+        if (iter instanceof YieldingKeyValueIterator) {
+          ((YieldingKeyValueIterator<Key,Value>) iter).enableYielding(yield.get());
+        }
+      }
       return true;
     }
 
@@ -176,8 +211,14 @@ public class SourceSwitchingIterator implements InterruptibleIterator
{
       this.inclusive = inclusive;
       this.columnFamilies = columnFamilies;
 
-      if (iter == null)
+      if (iter == null) {
         iter = source.iterator();
+        if (!onlySwitchAfterRow && yield.isPresent()) {
+          if (iter instanceof YieldingKeyValueIterator) {
+            ((YieldingKeyValueIterator<Key,Value>) iter).enableYielding(yield.get());
+          }
+        }
+      }
 
       readNext(true);
     }

http://git-wip-us.apache.org/repos/asf/accumulo/blob/1ab2431c/core/src/test/java/org/apache/accumulo/core/iterators/system/SourceSwitchingIteratorTest.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/org/apache/accumulo/core/iterators/system/SourceSwitchingIteratorTest.java
b/core/src/test/java/org/apache/accumulo/core/iterators/system/SourceSwitchingIteratorTest.java
index e6ca6d3..037ee7e 100644
--- a/core/src/test/java/org/apache/accumulo/core/iterators/system/SourceSwitchingIteratorTest.java
+++ b/core/src/test/java/org/apache/accumulo/core/iterators/system/SourceSwitchingIteratorTest.java
@@ -16,21 +16,28 @@
  */
 package org.apache.accumulo.core.iterators.system;
 
+import java.io.IOException;
 import java.util.ArrayList;
+import java.util.Collection;
 import java.util.List;
 import java.util.TreeMap;
 import java.util.concurrent.atomic.AtomicBoolean;
 
+import com.google.common.base.Optional;
 import junit.framework.TestCase;
 
 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.IterationInterruptedException;
 import org.apache.accumulo.core.iterators.IteratorEnvironment;
 import org.apache.accumulo.core.iterators.SortedKeyValueIterator;
 import org.apache.accumulo.core.iterators.SortedMapIterator;
+import org.apache.accumulo.core.iterators.WrappingIterator;
+import org.apache.accumulo.core.iterators.YieldCallback;
+import org.apache.accumulo.core.iterators.YieldingKeyValueIterator;
 import org.apache.accumulo.core.iterators.system.SourceSwitchingIterator.DataSource;
 import org.apache.hadoop.io.Text;
 
@@ -273,4 +280,109 @@ public class SourceSwitchingIteratorTest extends TestCase {
     } catch (IterationInterruptedException iie) {}
 
   }
+
+  private Range yield(Range r, SourceSwitchingIterator ssi, YieldCallback<Key> yield)
throws IOException {
+    while (yield.hasYielded()) {
+      Key yieldPosition = yield.getPositionAndReset();
+      if (!r.contains(yieldPosition)) {
+        throw new IOException("Underlying iterator yielded to a position outside of its range:
" + yieldPosition + " not in " + r);
+      }
+      r = new Range(yieldPosition, false, (Key) null, r.isEndKeyInclusive());
+      ssi.seek(r, new ArrayList<ByteSequence>(), false);
+    }
+    return r;
+  }
+
+  public void testYield() throws Exception {
+    TreeMap<Key,Value> tm1 = new TreeMap<>();
+    put(tm1, "r1", "cf1", "cq1", 5, "v1");
+    put(tm1, "r1", "cf1", "cq3", 5, "v2");
+    put(tm1, "r2", "cf1", "cq1", 5, "v3");
+
+    SortedMapIterator smi = new SortedMapIterator(tm1);
+    YieldingIterator ymi = new YieldingIterator(smi);
+    TestDataSource tds = new TestDataSource(ymi);
+    SourceSwitchingIterator ssi = new SourceSwitchingIterator(tds);
+
+    YieldCallback<Key> yield = new YieldCallback<>();
+    ssi.enableYielding(yield);
+
+    Range r = new Range();
+    ssi.seek(r, new ArrayList<ByteSequence>(), false);
+    r = yield(r, ssi, yield);
+    testAndCallNext(ssi, "r1", "cf1", "cq1", 5, "v1", true);
+    r = yield(r, ssi, yield);
+    testAndCallNext(ssi, "r1", "cf1", "cq3", 5, "v2", true);
+    r = yield(r, ssi, yield);
+    testAndCallNext(ssi, "r2", "cf1", "cq1", 5, "v3", true);
+    r = yield(r, ssi, yield);
+    assertFalse(ssi.hasTop());
+  }
+
+  /**
+   * This iterator which implements yielding will yield after every other next and every
other seek call.
+   */
+  private final AtomicBoolean yieldNextKey = new AtomicBoolean(false);
+  private final AtomicBoolean yieldSeekKey = new AtomicBoolean(false);
+
+  public class YieldingIterator extends WrappingIterator implements YieldingKeyValueIterator<Key,Value>
{
+    private Optional<YieldCallback<Key>> yield = Optional.absent();
+
+    public YieldingIterator(SortedKeyValueIterator<Key,Value> source) {
+      setSource(source);
+    }
+
+    @Override
+    public SortedKeyValueIterator<Key,Value> deepCopy(IteratorEnvironment env) {
+      return new YieldingIterator(getSource().deepCopy(env));
+    }
+
+    @Override
+    public boolean hasTop() {
+      return (!(yield.isPresent() && yield.get().hasYielded()) && super.hasTop());
+    }
+
+    @Override
+    public void next() throws IOException {
+      boolean yielded = false;
+
+      // yield on every other next call.
+      yieldNextKey.set(!yieldNextKey.get());
+      if (yield.isPresent() && yieldNextKey.get()) {
+        yielded = true;
+        // since we are not actually skipping keys underneath, simply use the key following
the top key as the yield key
+        yield.get().yield(getTopKey().followingKey(PartialKey.ROW_COLFAM_COLQUAL_COLVIS_TIME));
+      }
+
+      // if not yielding, then simply pass on the next call
+      if (!yielded) {
+        super.next();
+      }
+    }
+
+    @Override
+    public void seek(Range range, Collection<ByteSequence> columnFamilies, boolean
inclusive) throws IOException {
+      boolean yielded = false;
+
+      if (!range.isStartKeyInclusive()) {
+        // yield on every other seek call.
+        yieldSeekKey.set(!yieldSeekKey.get());
+        if (yield.isPresent() && yieldSeekKey.get()) {
+          yielded = true;
+          // since we are not actually skipping keys underneath, simply use the key following
the range start key
+          yield.get().yield(range.getStartKey().followingKey(PartialKey.ROW_COLFAM_COLQUAL_COLVIS_TIME));
+        }
+      }
+
+      // if not yielding, then simply pass on the call to the source
+      if (!yielded) {
+        super.seek(range, columnFamilies, inclusive);
+      }
+    }
+
+    @Override
+    public void enableYielding(YieldCallback<Key> yield) {
+      this.yield = Optional.of(yield);
+    }
+  }
 }

http://git-wip-us.apache.org/repos/asf/accumulo/blob/1ab2431c/iterator-test-harness/src/main/java/org/apache/accumulo/iteratortest/testcases/YieldingTestCase.java
----------------------------------------------------------------------
diff --git a/iterator-test-harness/src/main/java/org/apache/accumulo/iteratortest/testcases/YieldingTestCase.java
b/iterator-test-harness/src/main/java/org/apache/accumulo/iteratortest/testcases/YieldingTestCase.java
new file mode 100644
index 0000000..f9de207
--- /dev/null
+++ b/iterator-test-harness/src/main/java/org/apache/accumulo/iteratortest/testcases/YieldingTestCase.java
@@ -0,0 +1,87 @@
+/*
+ * 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.accumulo.iteratortest.testcases;
+
+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.SortedKeyValueIterator;
+import org.apache.accumulo.core.iterators.YieldCallback;
+import org.apache.accumulo.core.iterators.YieldingKeyValueIterator;
+import org.apache.accumulo.iteratortest.IteratorTestInput;
+import org.apache.accumulo.iteratortest.IteratorTestOutput;
+import org.apache.accumulo.iteratortest.IteratorTestUtil;
+import org.apache.accumulo.iteratortest.environments.SimpleIteratorEnvironment;
+
+import java.io.IOException;
+import java.util.TreeMap;
+
+/**
+ * Test case that verifies that an iterator works correctly with the yielding api. Note that
most iterators do nothing in terms of yielding in which case this
+ * merely tests that the iterator produces the correct output. If however the iterator does
override the yielding api, then this ensures that it works correctly
+ * iff the iterator actually decides to yield. Nothing can force an iterator to yield without
knowing something about the internals of the iterator being
+ * tested.
+ */
+public class YieldingTestCase extends OutputVerifyingTestCase {
+
+  @Override
+  public IteratorTestOutput test(IteratorTestInput testInput) {
+    final SortedKeyValueIterator<Key,Value> skvi = IteratorTestUtil.instantiateIterator(testInput);
+    final SortedKeyValueIterator<Key,Value> source = IteratorTestUtil.createSource(testInput);
+
+    try {
+      skvi.init(source, testInput.getIteratorOptions(), new SimpleIteratorEnvironment());
+
+      YieldCallback<Key> yield = new YieldCallback<>();
+      if (skvi instanceof YieldingKeyValueIterator) {
+        ((YieldingKeyValueIterator<Key,Value>) skvi).enableYielding(yield);
+      }
+
+      skvi.seek(testInput.getRange(), testInput.getFamilies(), testInput.isInclusive());
+      return new IteratorTestOutput(consume(testInput, skvi, yield));
+    } catch (IOException e) {
+      return new IteratorTestOutput(e);
+    }
+  }
+
+  TreeMap<Key,Value> consume(IteratorTestInput testInput, SortedKeyValueIterator<Key,Value>
skvi, YieldCallback<Key> yield) throws IOException {
+    TreeMap<Key,Value> data = new TreeMap<>();
+    Key lastKey = null;
+    while (yield.hasYielded() || skvi.hasTop()) {
+      if (yield.hasYielded()) {
+        Range r = testInput.getRange();
+        Key yieldPosition = yield.getPositionAndReset();
+        if (!r.contains(yieldPosition)) {
+          throw new IOException("Underlying iterator yielded to a position outside of its
range: " + yieldPosition + " not in " + r);
+        }
+        if (skvi.hasTop()) {
+          throw new IOException("Underlying iterator reports having a top, but has yielded:
" + yieldPosition);
+        }
+        if (lastKey != null && yieldPosition.compareTo(lastKey) <= 0) {
+          throw new IOException("Underlying iterator yielded at a position that is not past
the last key returned");
+        }
+        skvi.seek(new Range(yieldPosition, false, r.getEndKey(), r.isEndKeyInclusive()),
testInput.getFamilies(), testInput.isInclusive());
+      } else {
+        // Make sure to copy the K-V
+        data.put(new Key(skvi.getTopKey()), new Value(skvi.getTopValue()));
+        skvi.next();
+      }
+    }
+    return data;
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/accumulo/blob/1ab2431c/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java
----------------------------------------------------------------------
diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java b/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java
index a937235..6ead6b8 100644
--- a/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java
+++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java
@@ -282,6 +282,10 @@ public class TabletServer extends AccumuloServerContext implements Runnable
{
   private final Metrics scanMetrics;
   private final Metrics mincMetrics;
 
+  public Metrics getScanMetrics() {
+    return scanMetrics;
+  }
+
   public Metrics getMinCMetrics() {
     return mincMetrics;
   }

http://git-wip-us.apache.org/repos/asf/accumulo/blob/1ab2431c/server/tserver/src/main/java/org/apache/accumulo/tserver/metrics/Metrics2TabletServerScanMetrics.java
----------------------------------------------------------------------
diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/metrics/Metrics2TabletServerScanMetrics.java
b/server/tserver/src/main/java/org/apache/accumulo/tserver/metrics/Metrics2TabletServerScanMetrics.java
index 0c72cb5..d2fc871 100644
--- a/server/tserver/src/main/java/org/apache/accumulo/tserver/metrics/Metrics2TabletServerScanMetrics.java
+++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/metrics/Metrics2TabletServerScanMetrics.java
@@ -33,7 +33,7 @@ public class Metrics2TabletServerScanMetrics implements Metrics, MetricsSource,
 
   private final MetricsSystem system;
   private final MetricsRegistry registry;
-  private final MutableStat scans, resultsPerScan;
+  private final MutableStat scans, resultsPerScan, yields;
 
   // Use TabletServerMetricsFactory
   Metrics2TabletServerScanMetrics(MetricsSystem system) {
@@ -42,6 +42,7 @@ public class Metrics2TabletServerScanMetrics implements Metrics, MetricsSource,
 
     scans = registry.newStat(SCAN, "Scans", "Ops", "Count", true);
     resultsPerScan = registry.newStat(RESULT_SIZE, "Results per scan", "Ops", "Count", true);
+    yields = registry.newStat(YIELD, "Yields", "Ops", "Count", true);
   }
 
   @Override
@@ -50,6 +51,8 @@ public class Metrics2TabletServerScanMetrics implements Metrics, MetricsSource,
       scans.add(value);
     } else if (RESULT_SIZE.equals(name)) {
       resultsPerScan.add(value);
+    } else if (YIELD.equals(name)) {
+      yields.add(value);
     } else {
       throw new RuntimeException("Could not find metric to update for name " + name);
     }

http://git-wip-us.apache.org/repos/asf/accumulo/blob/1ab2431c/server/tserver/src/main/java/org/apache/accumulo/tserver/metrics/TabletServerScanMetricsKeys.java
----------------------------------------------------------------------
diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/metrics/TabletServerScanMetricsKeys.java
b/server/tserver/src/main/java/org/apache/accumulo/tserver/metrics/TabletServerScanMetricsKeys.java
index 7d33a84..8c9ab34 100644
--- a/server/tserver/src/main/java/org/apache/accumulo/tserver/metrics/TabletServerScanMetricsKeys.java
+++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/metrics/TabletServerScanMetricsKeys.java
@@ -23,5 +23,6 @@ public interface TabletServerScanMetricsKeys {
 
   String SCAN = "scan";
   String RESULT_SIZE = "result";
+  String YIELD = "yield";
 
 }

http://git-wip-us.apache.org/repos/asf/accumulo/blob/1ab2431c/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/Scanner.java
----------------------------------------------------------------------
diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/Scanner.java
b/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/Scanner.java
index 15526d7..bf3ba86 100644
--- a/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/Scanner.java
+++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/Scanner.java
@@ -95,7 +95,7 @@ public class Scanner {
         iter = new SourceSwitchingIterator(dataSource, false);
       }
 
-      results = tablet.nextBatch(iter, range, options.getNum(), options.getColumnSet(), options.getBatchTimeOut());
+      results = tablet.nextBatch(iter, range, options.getNum(), options.getColumnSet(), options.getBatchTimeOut(),
options.isIsolated());
 
       if (results.getResults() == null) {
         range = null;

http://git-wip-us.apache.org/repos/asf/accumulo/blob/1ab2431c/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/Tablet.java
----------------------------------------------------------------------
diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/Tablet.java b/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/Tablet.java
index 6637521..9278a36 100644
--- a/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/Tablet.java
+++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/Tablet.java
@@ -75,6 +75,8 @@ import org.apache.accumulo.core.file.FileSKVIterator;
 import org.apache.accumulo.core.iterators.IterationInterruptedException;
 import org.apache.accumulo.core.iterators.IteratorUtil.IteratorScope;
 import org.apache.accumulo.core.iterators.SortedKeyValueIterator;
+import org.apache.accumulo.core.iterators.YieldCallback;
+import org.apache.accumulo.core.iterators.YieldingKeyValueIterator;
 import org.apache.accumulo.core.iterators.system.SourceSwitchingIterator;
 import org.apache.accumulo.core.master.thrift.BulkImportState;
 import org.apache.accumulo.core.master.thrift.TabletLoadState;
@@ -138,6 +140,7 @@ import org.apache.accumulo.tserver.log.DfsLogger;
 import org.apache.accumulo.tserver.log.MutationReceiver;
 import org.apache.accumulo.tserver.mastermessage.TabletStatusMessage;
 import org.apache.accumulo.tserver.metrics.TabletServerMinCMetrics;
+import org.apache.accumulo.tserver.metrics.TabletServerScanMetrics;
 import org.apache.accumulo.tserver.tablet.Compactor.CompactionCanceledException;
 import org.apache.accumulo.tserver.tablet.Compactor.CompactionEnv;
 import org.apache.commons.codec.DecoderException;
@@ -546,11 +549,17 @@ public class Tablet implements TabletCommitter {
       batchTimeOut = 0;
     }
 
+    // determine if the iterator supported yielding
+    YieldCallback<Key> yield = new YieldCallback<>();
+    if (mmfi instanceof YieldingKeyValueIterator)
+      ((YieldingKeyValueIterator<Key,Value>) mmfi).enableYielding(yield);
+    boolean yielded = false;
+
     for (Range range : ranges) {
 
       boolean timesUp = batchTimeOut > 0 && System.nanoTime() > returnTime;
 
-      if (exceededMemoryUsage || tabletClosed || timesUp) {
+      if (exceededMemoryUsage || tabletClosed || timesUp || yielded) {
         lookupResult.unfinishedRanges.add(range);
         continue;
       }
@@ -564,6 +573,9 @@ public class Tablet implements TabletCommitter {
           mmfi.seek(range, LocalityGroupUtil.EMPTY_CF_SET, false);
 
         while (mmfi.hasTop()) {
+          if (yield.hasYielded()) {
+            throw new IOException("Coding error: hasTop returned true but has yielded at
" + yield.getPositionAndReset());
+          }
           Key key = mmfi.getTopKey();
 
           KVEntry kve = new KVEntry(key, mmfi.getTopValue());
@@ -584,6 +596,23 @@ public class Tablet implements TabletCommitter {
           mmfi.next();
         }
 
+        if (yield.hasYielded()) {
+          yielded = true;
+          Key yieldPosition = yield.getPositionAndReset();
+          if (!range.contains(yieldPosition)) {
+            throw new IOException("Underlying iterator yielded to a position outside of its
range: " + yieldPosition + " not in " + range);
+          }
+          if (!results.isEmpty() && yieldPosition.compareTo(results.get(results.size()
- 1).getKey()) <= 0) {
+            throw new IOException("Underlying iterator yielded to a position that does not
follow the last key returned: " + yieldPosition + " <= "
+                + results.get(results.size() - 1).getKey());
+          }
+          addUnfinishedRange(lookupResult, range, yieldPosition, false);
+
+          log.debug("Scan yield detected at position " + yieldPosition);
+          Metrics scanMetrics = getTabletServer().getScanMetrics();
+          if (scanMetrics.isEnabled())
+            scanMetrics.add(TabletServerScanMetrics.YIELD, 1);
+        }
       } catch (TooManyFilesException tmfe) {
         // treat this as a closed tablet, and let the client retry
         log.warn("Tablet " + getExtent() + " has too many files, batch lookup can not run");
@@ -696,7 +725,7 @@ public class Tablet implements TabletCommitter {
     }
   }
 
-  Batch nextBatch(SortedKeyValueIterator<Key,Value> iter, Range range, int num, Set<Column>
columns, long batchTimeOut) throws IOException {
+  Batch nextBatch(SortedKeyValueIterator<Key,Value> iter, Range range, int num, Set<Column>
columns, long batchTimeOut, boolean isolated) throws IOException {
 
     // log.info("In nextBatch..");
 
@@ -713,18 +742,27 @@ public class Tablet implements TabletCommitter {
 
     long maxResultsSize = tableConfiguration.getMemoryInBytes(Property.TABLE_SCAN_MAXMEM);
 
+    Key continueKey = null;
+    boolean skipContinueKey = false;
+
+    YieldCallback<Key> yield = new YieldCallback<>();
+
+    // we cannot yield if we are in isolation mode
+    if (!isolated) {
+      if (iter instanceof YieldingKeyValueIterator)
+        ((YieldingKeyValueIterator<Key,Value>) iter).enableYielding(yield);
+    }
+
     if (columns.size() == 0) {
       iter.seek(range, LocalityGroupUtil.EMPTY_CF_SET, false);
     } else {
       iter.seek(range, LocalityGroupUtil.families(columns), true);
     }
 
-    Key continueKey = null;
-    boolean skipContinueKey = false;
-
-    boolean endOfTabletReached = false;
     while (iter.hasTop()) {
-
+      if (yield.hasYielded()) {
+        throw new IOException("Coding error: hasTop returned true but has yielded at " +
yield.getPositionAndReset());
+      }
       value = iter.getTopValue();
       key = iter.getTopKey();
 
@@ -744,17 +782,28 @@ public class Tablet implements TabletCommitter {
       iter.next();
     }
 
-    if (iter.hasTop() == false) {
-      endOfTabletReached = true;
-    }
+    if (yield.hasYielded()) {
+      continueKey = new Key(yield.getPositionAndReset());
+      skipContinueKey = true;
+      if (!range.contains(continueKey)) {
+        throw new IOException("Underlying iterator yielded to a position outside of its range:
" + continueKey + " not in " + range);
+      }
+      if (!results.isEmpty() && continueKey.compareTo(results.get(results.size()
- 1).getKey()) <= 0) {
+        throw new IOException("Underlying iterator yielded to a position that does not follow
the last key returned: " + continueKey + " <= "
+            + results.get(results.size() - 1).getKey());
+      }
 
-    if (endOfTabletReached) {
+      log.debug("Scan yield detected at position " + continueKey);
+      Metrics scanMetrics = getTabletServer().getScanMetrics();
+      if (scanMetrics.isEnabled())
+        scanMetrics.add(TabletServerScanMetrics.YIELD, 1);
+    } else if (iter.hasTop() == false) {
+      // end of tablet has been reached
       continueKey = null;
+      if (results.size() == 0)
+        results = null;
     }
 
-    if (endOfTabletReached && results.size() == 0)
-      results = null;
-
     return new Batch(skipContinueKey, results, continueKey, resultBytes);
   }
 

http://git-wip-us.apache.org/repos/asf/accumulo/blob/1ab2431c/test/src/main/java/org/apache/accumulo/test/YieldScannersIT.java
----------------------------------------------------------------------
diff --git a/test/src/main/java/org/apache/accumulo/test/YieldScannersIT.java b/test/src/main/java/org/apache/accumulo/test/YieldScannersIT.java
new file mode 100644
index 0000000..4d86dd3
--- /dev/null
+++ b/test/src/main/java/org/apache/accumulo/test/YieldScannersIT.java
@@ -0,0 +1,161 @@
+/*
+ * 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.accumulo.test;
+
+import org.apache.accumulo.core.client.BatchScanner;
+import org.apache.accumulo.core.client.BatchWriter;
+import org.apache.accumulo.core.client.BatchWriterConfig;
+import org.apache.accumulo.core.client.Connector;
+import org.apache.accumulo.core.client.IteratorSetting;
+import org.apache.accumulo.core.client.Scanner;
+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.accumulo.core.security.Authorizations;
+import org.apache.accumulo.harness.AccumuloClusterHarness;
+import org.apache.accumulo.minicluster.impl.MiniAccumuloConfigImpl;
+import org.apache.accumulo.test.functional.YieldingIterator;
+import org.apache.commons.lang.StringUtils;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.io.Text;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.Map;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+// ACCUMULO-4643
+public class YieldScannersIT extends AccumuloClusterHarness {
+  Logger log = LoggerFactory.getLogger(YieldScannersIT.class);
+  private static final char START_ROW = 'a';
+
+  @Override
+  public int defaultTimeoutSeconds() {
+    return 60;
+  }
+
+  @Override
+  public void configureMiniCluster(MiniAccumuloConfigImpl cfg, Configuration hadoopCoreSite)
{
+    cfg.setNumTservers(1);
+  }
+
+  @Test
+  public void testScan() throws Exception {
+    // make a table
+    final String tableName = getUniqueNames(1)[0];
+    final Connector conn = getConnector();
+    conn.tableOperations().create(tableName);
+    final BatchWriter writer = conn.createBatchWriter(tableName, new BatchWriterConfig());
+    for (int i = 0; i < 10; i++) {
+      byte[] row = new byte[] {(byte) (START_ROW + i)};
+      Mutation m = new Mutation(new Text(row));
+      m.put(new Text(), new Text(), new Value());
+      writer.addMutation(m);
+    }
+    writer.flush();
+    writer.close();
+
+    log.info("Creating scanner");
+    // make a scanner for a table with 10 keys
+    final Scanner scanner = conn.createScanner(tableName, Authorizations.EMPTY);
+    final IteratorSetting cfg = new IteratorSetting(100, YieldingIterator.class);
+    scanner.addScanIterator(cfg);
+
+    log.info("iterating");
+    Iterator<Map.Entry<Key,Value>> it = scanner.iterator();
+    int keyCount = 0;
+    int yieldNextCount = 0;
+    int yieldSeekCount = 0;
+    while (it.hasNext()) {
+      Map.Entry<Key,Value> next = it.next();
+      log.info(Integer.toString(keyCount) + ": Got key " + next.getKey() + " with value "
+ next.getValue());
+
+      // verify we got the expected key
+      char expected = (char) (START_ROW + keyCount);
+      Assert.assertEquals("Unexpected row", Character.toString(expected), next.getKey().getRow().toString());
+
+      // determine whether we yielded on a next and seek
+      if ((keyCount & 1) != 0) {
+        yieldNextCount++;
+        yieldSeekCount++;
+      }
+      String[] value = StringUtils.split(next.getValue().toString(), ',');
+      Assert.assertEquals("Unexpected yield next count", Integer.toString(yieldNextCount),
value[0]);
+      Assert.assertEquals("Unexpected yield seek count", Integer.toString(yieldSeekCount),
value[1]);
+      Assert.assertEquals("Unexpected rebuild count", Integer.toString(yieldNextCount + yieldSeekCount),
value[2]);
+
+      keyCount++;
+    }
+    Assert.assertEquals("Did not get the expected number of results", 10, keyCount);
+  }
+
+  @Test
+  public void testBatchScan() throws Exception {
+    // make a table
+    final String tableName = getUniqueNames(1)[0];
+    final Connector conn = getConnector();
+    conn.tableOperations().create(tableName);
+    final BatchWriter writer = conn.createBatchWriter(tableName, new BatchWriterConfig());
+    for (int i = 0; i < 10; i++) {
+      byte[] row = new byte[] {(byte) (START_ROW + i)};
+      Mutation m = new Mutation(new Text(row));
+      m.put(new Text(), new Text(), new Value());
+      writer.addMutation(m);
+    }
+    writer.flush();
+    writer.close();
+
+    log.info("Creating batch scanner");
+    // make a scanner for a table with 10 keys
+    final BatchScanner scanner = conn.createBatchScanner(tableName, Authorizations.EMPTY,
1);
+    final IteratorSetting cfg = new IteratorSetting(100, YieldingIterator.class);
+    scanner.addScanIterator(cfg);
+    scanner.setRanges(Collections.singleton(new Range()));
+
+    log.info("iterating");
+    Iterator<Map.Entry<Key,Value>> it = scanner.iterator();
+    int keyCount = 0;
+    int yieldNextCount = 0;
+    int yieldSeekCount = 0;
+    while (it.hasNext()) {
+      Map.Entry<Key,Value> next = it.next();
+      log.info(Integer.toString(keyCount) + ": Got key " + next.getKey() + " with value "
+ next.getValue());
+
+      // verify we got the expected key
+      char expected = (char) (START_ROW + keyCount);
+      Assert.assertEquals("Unexpected row", Character.toString(expected), next.getKey().getRow().toString());
+
+      // determine whether we yielded on a next and seek
+      if ((keyCount & 1) != 0) {
+        yieldNextCount++;
+        yieldSeekCount++;
+      }
+      String[] value = StringUtils.split(next.getValue().toString(), ',');
+      Assert.assertEquals("Unexpected yield next count", Integer.toString(yieldNextCount),
value[0]);
+      Assert.assertEquals("Unexpected yield seek count", Integer.toString(yieldSeekCount),
value[1]);
+      Assert.assertEquals("Unexpected rebuild count", Integer.toString(yieldNextCount + yieldSeekCount),
value[2]);
+
+      keyCount++;
+    }
+    Assert.assertEquals("Did not get the expected number of results", 10, keyCount);
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/accumulo/blob/1ab2431c/test/src/main/java/org/apache/accumulo/test/functional/YieldingIterator.java
----------------------------------------------------------------------
diff --git a/test/src/main/java/org/apache/accumulo/test/functional/YieldingIterator.java
b/test/src/main/java/org/apache/accumulo/test/functional/YieldingIterator.java
new file mode 100644
index 0000000..4ac325d
--- /dev/null
+++ b/test/src/main/java/org/apache/accumulo/test/functional/YieldingIterator.java
@@ -0,0 +1,129 @@
+/*
+ * 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.accumulo.test.functional;
+
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import com.google.common.base.Optional;
+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.accumulo.core.iterators.WrappingIterator;
+
+import java.io.IOException;
+import java.util.Collection;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.accumulo.core.iterators.YieldCallback;
+import org.apache.accumulo.core.iterators.YieldingKeyValueIterator;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * This iterator which implements yielding will yield after every other next and every other
seek call.
+ */
+public class YieldingIterator extends WrappingIterator implements YieldingKeyValueIterator<Key,Value>
{
+  private static final Logger log = LoggerFactory.getLogger(YieldingIterator.class);
+  private static final AtomicInteger yieldNexts = new AtomicInteger(0);
+  private static final AtomicInteger yieldSeeks = new AtomicInteger(0);
+  private static final AtomicInteger rebuilds = new AtomicInteger(0);
+
+  private static final AtomicBoolean yieldNextKey = new AtomicBoolean(false);
+  private static final AtomicBoolean yieldSeekKey = new AtomicBoolean(false);
+
+  private Optional<YieldCallback<Key>> yield = Optional.absent();
+
+  @Override
+  public SortedKeyValueIterator<Key,Value> deepCopy(IteratorEnvironment env) {
+    YieldingIterator it = new YieldingIterator();
+    it.setSource(getSource().deepCopy(env));
+    return it;
+  }
+
+  @Override
+  public boolean hasTop() {
+    return (!(yield.isPresent() && yield.get().hasYielded()) && super.hasTop());
+  }
+
+  @Override
+  public void next() throws IOException {
+    log.info("start YieldingIterator.next: " + getTopValue());
+    boolean yielded = false;
+
+    // yield on every other next call.
+    yieldNextKey.set(!yieldNextKey.get());
+    if (yield.isPresent() && yieldNextKey.get()) {
+      yielded = true;
+      yieldNexts.incrementAndGet();
+      // since we are not actually skipping keys underneath, simply use the key following
the top key as the yield key
+      yield.get().yield(getTopKey().followingKey(PartialKey.ROW_COLFAM_COLQUAL_COLVIS_TIME));
+      log.info("end YieldingIterator.next: yielded at " + getTopKey());
+    }
+
+    // if not yielding, then simply pass on the next call
+    if (!yielded) {
+      super.next();
+      log.info("end YieldingIterator.next: " + (hasTop() ? getTopKey() + " " + getTopValue()
: "no top"));
+    }
+  }
+
+  /**
+   * The top value will encode the current state of the yields, seeks, and rebuilds for use
by the YieldScannersIT tests.
+   *
+   * @return a top value of the form {yieldNexts},{yieldSeeks},{rebuilds}
+   */
+  @Override
+  public Value getTopValue() {
+    String value = Integer.toString(yieldNexts.get()) + ',' + Integer.toString(yieldSeeks.get())
+ ',' + Integer.toString(rebuilds.get());
+    return new Value(value);
+  }
+
+  @Override
+  public void seek(Range range, Collection<ByteSequence> columnFamilies, boolean inclusive)
throws IOException {
+    log.info("start YieldingIterator.seek: " + getTopValue() + " with range " + range);
+    boolean yielded = false;
+
+    if (!range.isStartKeyInclusive()) {
+      rebuilds.incrementAndGet();
+
+      // yield on every other seek call.
+      yieldSeekKey.set(!yieldSeekKey.get());
+      if (yield.isPresent() && yieldSeekKey.get()) {
+        yielded = true;
+        yieldSeeks.incrementAndGet();
+        // since we are not actually skipping keys underneath, simply use the key following
the range start key
+        yield.get().yield(range.getStartKey().followingKey(PartialKey.ROW_COLFAM_COLQUAL_COLVIS_TIME));
+        log.info("end YieldingIterator.next: yielded at " + range.getStartKey());
+      }
+    }
+
+    // if not yielding, then simply pass on the call to the source
+    if (!yielded) {
+      super.seek(range, columnFamilies, inclusive);
+      log.info("end YieldingIterator.seek: " + (hasTop() ? getTopKey() + " " + getTopValue()
: "no top"));
+    }
+  }
+
+  @Override
+  public void enableYielding(YieldCallback<Key> yield) {
+    this.yield = Optional.of(yield);
+  }
+}


Mime
View raw message