package parquet.column.values.rle;

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

import parquet.Preconditions;
import parquet.column.values.ValuesReader;

/**
 * This ValuesReader does all the reading in {@link #initFromPage}
 * and stores the values in an in memory buffer, which is less than ideal.
 *
 * @author Alex Levenson
 */
public class RunLengthBitPackingHybridValuesReader extends ValuesReader {
  private final int bitWidth;

  private int[][] values;
  private int currentSlab;
  private int currentValueInSlab;

  public RunLengthBitPackingHybridValuesReader(int bitWidth) {
    this.bitWidth = bitWidth;
  }

  @Override
  public int initFromPage(long valueCount, byte[] page, int offset) throws IOException {
    PosReportingByteArrayInputStream in =
        new PosReportingByteArrayInputStream(page, offset, page.length);

    RunLengthBitPackingHybridDecoder decoder = new RunLengthBitPackingHybridDecoder(bitWidth, in);

    if (valueCount <= 0) {
      // readInteger() will never be called,
      // there is no data to read
      return offset;
    }

    // ceil(valueCount / Integer.MAX_VALUE)
    long numSlabsL = 1 + ((valueCount - 1L) / Integer.MAX_VALUE);

    // safe cast to to int
    // TODO: is this necessary?
    Preconditions.checkArgument(numSlabsL <= Integer.MAX_VALUE,
        String.format("valueCount (%d) is too large", valueCount));
    int numSlabs = (int) numSlabsL;

    // make the slabs array
    values = new int[numSlabs][];

    // all the slabs except the last one are of size Integer.MAX_VALUE
    for (int i = 0; i < numSlabs - 1; i++) {
      values[i] = new int[Integer.MAX_VALUE];
    }

    // create the last slab
    if (valueCount % Integer.MAX_VALUE != 0) {
      // this slab can be smaller
      values[numSlabs - 1] = new int[(int) (valueCount % Integer.MAX_VALUE)];
    } else {
      // it just so happens that the last slab also needs to be
      // Integer.MAX_VALUE
      values[numSlabs - 1] = new int[Integer.MAX_VALUE];
    }

    currentSlab = 0;
    currentValueInSlab = 0;

    // decode all the values
    for (long i = 0; i < valueCount; i++) {
      if (currentValueInSlab > values[currentSlab].length) {
        ++currentSlab;
        currentValueInSlab = 0;
      }
      values[currentSlab][currentValueInSlab] = decoder.readInt();
      ++currentValueInSlab;
    }

    currentSlab = 0;
    currentValueInSlab = 0;

    // in has been keeping track of its position in page
    return in.getPos();
  }

  @Override
  public int readInteger() {
    if (currentValueInSlab > values[currentSlab].length) {
      ++currentSlab;
      currentValueInSlab = 0;
    }
    int ret = values[currentSlab][currentValueInSlab];
    ++currentValueInSlab;
    return ret;
  }

  /**
   * A regular {@link ByteArrayInputStream}
   * that exposes its position in buf
   */
  static class PosReportingByteArrayInputStream extends ByteArrayInputStream {
    public PosReportingByteArrayInputStream(byte[] buf) {
      super(buf);
    }

    public PosReportingByteArrayInputStream(byte[] buf, int offset, int length) {
      super(buf, offset, length);
    }

    public int getPos() {
      return pos;
    }
  }
}

@@ -467,18 +467,16 @@
 te int%5B%5D
-%5B%5D
  values;
@@ -494,53 +494,18 @@
 int 
-currentSlab;%0A  private int currentV
+v
 alueIn
-Slab
+dex
 ;%0A%0A 
@@ -654,16 +654,17 @@
 lueCount
+L
 , byte%5B%5D
@@ -711,425 +711,33 @@
     
-PosReportingByteArrayInputStream in =%0A        new PosReportingByteArrayInputStream(page, offset, page.length);%0A%0A    RunLengthBitPackingHybridDecoder decoder = new RunLengthBitPackingHybridDecoder(bitWidth, in);%0A%0A    if (valueCount %3C= 0) %7B%0A      // readInteger() will never be called,%0A      // there is no data to read%0A      return offset;%0A    %7D%0A%0A    // ceil(valueCount / Integer.MAX_VALUE)%0A    long numSlabsL = 1 + ((
+// TODO: we are assuming 
 valu
@@ -747,15 +747,9 @@
 unt 
-- 1L) /
+%3C
  Int
@@ -766,19 +766,16 @@
 ALUE
-);%0A
 %0A    // 
 safe
@@ -774,59 +774,55 @@
  // 
-safe cast to to int%0A    // TODO: is this necessary?
+      we should address this here and elsewhere
 %0A   
@@ -854,20 +854,21 @@
 ent(
-numSlabs
+valueCount
 L %3C
-=
  Int
@@ -889,18 +889,16 @@
 ,%0A      
- 
- 
 String.f
@@ -924,20 +924,48 @@
 %25d) 
-is too large
+cannot be greater than Integer.MAX_VALUE
 %22, v
@@ -977,20 +977,22 @@
 ount
+L
 ));%0A
+%0A
     int 
 numS
@@ -991,671 +991,423 @@
 int 
-numSlabs = (int) numSlabsL;%0A%0A    // make the slabs array%0A    values = new int%5BnumSlabs%5D%5B%5D;%0A%0A    // all the slabs except the last one are of size Integer.MAX_VALUE%0A    for (int i = 0; i %3C numSlabs - 1; i++) %7B%0A      values%5Bi%5D = new int%5BInteger.MAX_VALUE%5D;%0A    %7D%0A%0A    // create the last slab%0A    if (valueCount %25 Integer.MAX_VALUE != 0) %7B%0A      // this slab can be smaller%0A      values%5BnumSlabs - 1%5D = new int%5B(int) (valueCount %25 Integer.MAX_VALUE)%5D;%0A    %7D else %7B%0A      // it just so happens that the last slab also needs to be%0A      // Integer.MAX_VALUE%0A      values%5BnumSlabs - 1%5D = new int%5BInteger.MAX_VALUE%5D;%0A    %7D%0A%0A    currentSlab = 0;%0A    currentValueInSlab = 0
+valueCount = (int) valueCountL;%0A%0A    PosReportingByteArrayInputStream in =%0A        new PosReportingByteArrayInputStream(page, offset, page.length);%0A%0A    RunLengthBitPackingHybridDecoder decoder = new RunLengthBitPackingHybridDecoder(bitWidth, in);%0A%0A    if (valueCount %3C= 0) %7B%0A      // readInteger() will never be called,%0A      // there is no data to read%0A      return offset;%0A    %7D%0A%0A    values = new int%5BvalueCount%5D
 ;%0A%0A 
@@ -1443,20 +1443,19 @@
    for (
-long
+int
  i = 0; 
@@ -1487,278 +1487,66 @@
     
-if (currentValueInSlab %3E values%5BcurrentSlab%5D.length) %7B%0A        ++currentSlab;%0A        currentValueInSlab = 0;%0A      %7D%0A      values%5BcurrentSlab%5D%5BcurrentValueInSlab%5D = decoder.readInt();%0A      ++currentValueInSlab;%0A    %7D%0A%0A    currentSlab = 0;%0A    currentV
+values%5Bi%5D = decoder.readInt();%0A    %7D%0A%0A    v
 alueIn
-Slab
+dex
  = 0;%0A
-%0A
     
@@ -1677,198 +1677,53 @@
    i
-f (currentValueInSlab %3E values%5BcurrentSlab%5D.length) %7B%0A      ++currentSlab;%0A      currentValueInSlab = 0;%0A    %7D%0A    int ret = values%5BcurrentSlab%5D%5BcurrentV
+nt ret = values%5Bv
 alueIn
-Slab
+dex
 %5D;%0A    
-++currentV
+v
 alueIn
-Slab
+dex++
 ;%0A  

Remove logic for valueCount > Integer.MAX_VALUE