/**
 * Copyright 2012 Twitter, Inc.
 *
 * Licensed 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 redelm.column.primitive;

import java.io.ByteArrayInputStream;
import java.io.DataInputStream;
import java.io.IOException;

import static redelm.column.primitive.SimplePrimitiveColumnWriter.CHARSET;

public class SimplePrimitiveColumnReader extends PrimitiveColumnReader {

  private DataInputStream in;

  public SimplePrimitiveColumnReader(byte[] data, int offset, int length) {
    this.in = new DataInputStream(new ByteArrayInputStream(data, offset, length));
  }

  @Override
  public float readFloat() {
    try {
      return in.readFloat();
    } catch (IOException e) {
      throw new RuntimeException("never happens", e);
    }
  }

  @Override
  public byte[] readBytes() {
    try {
      byte[] value = new byte[in.readInt()];
      in.readFully(value);
      return value;
    } catch (IOException e) {
      throw new RuntimeException("never happens", e);
    }
  }

  @Override
  public boolean readBoolean() {
    try {
      return in.readBoolean();
    } catch (IOException e) {
      throw new RuntimeException("never happens", e);
    }
  }

  @Override
  public String readString() {
    try {
      int size = in.readInt();
      byte[] bytes = new byte[size];
      int i = 0;
      do {
        int n = in.read(bytes, i, bytes.length - i);
        if (n == -1) {
          throw new RuntimeException("Reached end of stream");
        }
        i = i + n;
      } while (i < bytes.length);
      return new String(bytes, CHARSET);
    } catch (IOException e) {
      throw new RuntimeException("never happens", e);
    }
  }

  @Override
  public double readDouble() {
    try {
      return in.readDouble();
    } catch (IOException e) {
      throw new RuntimeException("never happens", e);
    }
  }

  @Override
  public int readInt() {
    try {
      return in.readInt();
    } catch (IOException e) {
      throw new RuntimeException("never happens", e);
    }
  }

  @Override
  public long readLong() {
    try {
      return in.readLong();
    } catch (IOException e) {
      throw new RuntimeException("never happens", e);
    }
  }

  @Override
  public int readByte() {
    try {
      return in.read();
    } catch (IOException e) {
      throw new RuntimeException("never happens", e);
    }
  }

}

@@ -1739,24 +1739,83 @@
 .readInt();%0A
+      if (size == 0) %7B%0A        return %22%22;%0A      %7D else %7B%0A  
       byte%5B%5D
@@ -1841,24 +1841,26 @@
 ize%5D;%0A      
+  
 int i = 0;%0A 
@@ -1864,16 +1864,18 @@
 ;%0A      
+  
 do %7B%0A   
@@ -1875,24 +1875,26 @@
 o %7B%0A        
+ 
+ 
 int n = in.r
@@ -1926,16 +1926,18 @@
 h - i);%0A
+  
         
@@ -1957,24 +1957,26 @@
 %7B%0A          
+  
 throw new Ru
@@ -2024,18 +2024,22 @@
         
+ 
+ 
 %7D%0A
+  
         
@@ -2049,16 +2049,18 @@
  i + n;%0A
+  
       %7D 
@@ -2087,24 +2087,26 @@
 gth);%0A      
+  
 return new S
@@ -2128,16 +2128,24 @@
 ARSET);%0A
+      %7D%0A
     %7D ca

fix empty string bug