[core] Optimize toArray of all ByteIterators except RandomByteIterator

Method nextBuf is a clever hack that outperforms Random.nextBytes
but performs poorly for all other ByteIterator implementations

This commit moves it to RandomByteIterator and adds efficient
toArray implementations for other ByteIterator classes.

Also InputStreamByteIterator.reset method that unconditionally
throws UnsupportedOperationException is fixed

core/src/main/java/com/yahoo/ycsb/ByteArrayByteIterator.java

@@ -21,7 +21,7 @@
  */
 public class ByteArrayByteIterator extends ByteIterator {
   private final int originalOffset;
-  private byte[] str;
+  private final byte[] str;
   private int off;
   private final int len;
 
@@ -60,5 +60,14 @@ public long bytesLeft() {
   public void reset() {
     off = originalOffset;
   }
-
+
+  @Override
+  public byte[] toArray() {
+    int size = (int) bytesLeft();
+    byte[] bytes = new byte[size];
+    System.arraycopy(str, off, bytes, 0, size);
+    off = len;
+    return bytes;
+  }
+
 }

core/src/main/java/com/yahoo/ycsb/ByteIterator.java

@@ -81,7 +81,7 @@ public void remove() {
   public void reset() {
     throw new UnsupportedOperationException();
   }
-  
+
   /** Consumes remaining contents of this object, and returns them as a string. */
   public String toString() {
     Charset cset = Charset.forName("UTF-8");
@@ -95,10 +95,10 @@ public byte[] toArray() {
     if (left != (int) left) {
       throw new ArrayIndexOutOfBoundsException("Too much data to fit in one array!");
     }
+
     byte[] ret = new byte[(int) left];
-    int off = 0;
-    while (off < ret.length) {
-      off = nextBuf(ret, off);
+    for (int i = 0; i < ret.length; i++) {
+      ret[i] = nextByte();
     }
     return ret;
   }

core/src/main/java/com/yahoo/ycsb/InputStreamByteIterator.java

@@ -23,8 +23,8 @@
  *  A ByteIterator that iterates through an inputstream of bytes.
  */
 public class InputStreamByteIterator extends ByteIterator {
-  private long len;
-  private InputStream ins;
+  private final long len;
+  private final InputStream ins;
   private long off;
   private final boolean resetable;
 
@@ -63,17 +63,34 @@ public long bytesLeft() {
     return len - off;
   }
 
+  @Override
+  public byte[] toArray() {
+    int size = (int) bytesLeft();
+    byte[] bytes = new byte[size];
+    try {
+      if (ins.read(bytes) < size) {
+        throw new IllegalStateException("Past EOF!");
+      }
+    } catch (IOException e) {
+      throw new IllegalStateException(e);
+    }
+    off = len;
+    return bytes;
+  }
+
   @Override
   public void reset() {
     if (resetable) {
       try {
         ins.reset();
         ins.mark((int) len);
+        off = 0;
       } catch (IOException e) {
         throw new IllegalStateException("Failed to reset the input stream", e);
       }
+    } else {
+      throw new UnsupportedOperationException();
     }
-    throw new UnsupportedOperationException();
   }
 
 }

core/src/main/java/com/yahoo/ycsb/RandomByteIterator.java

@@ -20,10 +20,10 @@
  *  A ByteIterator that generates a random sequence of bytes.
  */
 public class RandomByteIterator extends ByteIterator {
-  private long len;
+  private final long len;
   private long off;
   private int bufOff;
-  private byte[] buf;
+  private final byte[] buf;
 
   @Override
   public boolean hasNext() {
@@ -98,5 +98,19 @@ public long bytesLeft() {
   public void reset() {
     off = 0;
   }
-
+
+  /** Consumes remaining contents of this object, and returns them as a byte array. */
+  public byte[] toArray() {
+    long left = bytesLeft();
+    if (left != (int) left) {
+      throw new ArrayIndexOutOfBoundsException("Too much data to fit in one array!");
+    }
+    byte[] ret = new byte[(int) left];
+    int bufOffset = 0;
+    while (bufOffset < ret.length) {
+      bufOffset = nextBuf(ret, bufOffset);
+    }
+    return ret;
+  }
+
 }

core/src/main/java/com/yahoo/ycsb/StringByteIterator.java

@@ -100,7 +100,17 @@ public long bytesLeft() {
   public void reset() {
     off = 0;
   }
-
+
+  @Override
+  public byte[] toArray() {
+    byte[] bytes = new byte[(int) bytesLeft()];
+    for (int i = 0; i < bytes.length; i++) {
+      bytes[i] = (byte) str.charAt(off + i);
+    }
+    off = str.length();
+    return bytes;
+  }
+
   /**
    * Specialization of general purpose toString() to avoid unnecessary
    * copies.