HBASE-11865 Result implements CellScannable; rather it should BE a CellScanner

hbase-client/src/main/java/org/apache/hadoop/hbase/client/Result.java

@@ -58,20 +58,23 @@ import org.apache.hadoop.hbase.util.Bytes;
  *
  * To get the latest value for a specific family and qualifier use {@link #getValue(byte[], byte[])}.
  *
- * A Result is backed by an array of {@link KeyValue} objects, each representing
+ * A Result is backed by an array of {@link Cell} objects, each representing
  * an HBase cell defined by the row, family, qualifier, timestamp, and value.<p>
  *
- * The underlying {@link KeyValue} objects can be accessed through the method {@link #listCells()}.
- * Each KeyValue can then be accessed through
- * {@link KeyValue#getRow()}, {@link KeyValue#getFamily()}, {@link KeyValue#getQualifier()},
- * {@link KeyValue#getTimestamp()}, and {@link KeyValue#getValue()}.<p>
+ * The underlying {@link Cell} objects can be accessed through the method {@link #listCells()}.
+ * This will create a List from the internal Cell []. Better is to exploit the fact that
+ * a new Result instance is a primed {@link CellScanner}; just call {@link #advance()} and
+ * {@link #current()} to iterate over Cells as you would any {@link CellScanner}.
+ * Call {@link #cellScanner()} to reset should you need to iterate the same Result over again
+ * ({@link CellScanner}s are one-shot).
  *
- * If you need to overwrite a Result with another Result instance -- as in the old 'mapred' RecordReader next
- * invocations -- then create an empty Result with the null constructor and in then use {@link #copyFrom(Result)}
+ * If you need to overwrite a Result with another Result instance -- as in the old 'mapred'
+ * RecordReader next invocations -- then create an empty Result with the null constructor and
+ * in then use {@link #copyFrom(Result)}
  */
 @InterfaceAudience.Public
 @InterfaceStability.Stable
-public class Result implements CellScannable {
+public class Result implements CellScannable, CellScanner {
   private Cell[] cells;
   private Boolean exists; // if the query was just to check existence.
   private boolean stale = false;
@@ -86,6 +89,13 @@ public class Result implements CellScannable {
   private static final int PAD_WIDTH = 128;
   public static final Result EMPTY_RESULT = new Result();
 
+  private final static int INITIAL_CELLSCANNER_INDEX = -1;
+
+  /**
+   * Index for where we are when Result is acting as a {@link CellScanner}.
+   */
+  private int cellScannerIndex = INITIAL_CELLSCANNER_INDEX;
+
   /**
    * Creates an empty Result w/ no KeyValue payload; returns null if you call {@link #rawCells()}.
    * Use this to represent no results if <code>null</code> won't do or in old 'mapred' as oppposed to 'mapreduce' package
@@ -827,7 +837,21 @@ public class Result implements CellScannable {
 
   @Override
   public CellScanner cellScanner() {
-    return CellUtil.createCellScanner(this.cells);
+    // Reset
+    this.cellScannerIndex = INITIAL_CELLSCANNER_INDEX;
+    return this;
+  }
+
+  @Override
+  public Cell current() {
+    if (cells == null) return null;
+    return (cellScannerIndex < 0)? null: this.cells[cellScannerIndex];
+  }
+
+  @Override
+  public boolean advance() {
+    if (cells == null) return false;
+    return ++cellScannerIndex < this.cells.length;
   }
 
   public Boolean getExists() {
@@ -846,5 +870,4 @@ public class Result implements CellScannable {
   public boolean isStale() {
     return stale;
   }
-
 }

hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestResult.java

@@ -21,6 +21,7 @@ package org.apache.hadoop.hbase.client;
 
 import static org.apache.hadoop.hbase.HBaseTestCase.assertByteEquals;
 
+import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.util.Arrays;
 import java.util.List;
@@ -30,6 +31,7 @@ import junit.framework.TestCase;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.hbase.Cell;
+import org.apache.hadoop.hbase.CellScanner;
 import org.apache.hadoop.hbase.CellUtil;
 import org.apache.hadoop.hbase.KeyValue;
 import org.apache.hadoop.hbase.SmallTests;
@@ -60,6 +62,31 @@ public class TestResult extends TestCase {
   static final byte [] family = Bytes.toBytes("family");
   static final byte [] value = Bytes.toBytes("value");
 
+  /**
+   * Run some tests to ensure Result acts like a proper CellScanner.
+   * @throws IOException
+   */
+  public void testResultAsCellScanner() throws IOException {
+    Cell [] cells = genKVs(row, family, value, 1, 10);
+    Arrays.sort(cells, KeyValue.COMPARATOR);
+    Result r = Result.create(cells);
+    assertSame(r, cells);
+    // Assert I run over same result multiple times.
+    assertSame(r.cellScanner(), cells);
+    assertSame(r.cellScanner(), cells);
+    // Assert we are not creating new object when doing cellscanner
+    assertTrue(r == r.cellScanner());
+  }
+
+  private void assertSame(final CellScanner cellScanner, final Cell [] cells) throws IOException {
+    int count = 0;
+    while (cellScanner.advance()) {
+      assertTrue(cells[count].equals(cellScanner.current()));
+      count++;
+    }
+    assertEquals(cells.length, count);
+  }
+
   public void testBasicGetColumn() throws Exception {
     KeyValue [] kvs = genKVs(row, family, value, 1, 100);