[Issue #1363] implement ByteColumnReader and RunLenByteDecoder

cpp/pixels-common/include/utils/Constants.h

@@ -40,9 +40,9 @@ class Constants
     static int REDIS_BUFFER_SIZE;
     static int GCS_BUFFER_SIZE;
 
-    static int MIN_REPEAT;
-    static int MAX_SCOPE;
-    static int MAX_SHORT_REPEAT_LENGTH;
+    static int RLE_MIN_REPEAT;
+    static int INT_RLE_MAX_SCOPE;
+    static int INT_RLE_MAX_SHORT_REPEAT;
     static float DICT_KEY_SIZE_THRESHOLD;
     static int INIT_DICT_SIZE;
 

cpp/pixels-common/lib/utils/Constants.cpp

@@ -34,9 +34,9 @@ int Constants::S3_BUFFER_SIZE = 8 * 1024 * 1024;
 int Constants::REDIS_BUFFER_SIZE = 8 * 1024 * 1024;
 int Constants::GCS_BUFFER_SIZE = 8 * 1024 * 1024;
 
-int Constants::MIN_REPEAT = 3;
-int Constants::MAX_SCOPE = 512;
-int Constants::MAX_SHORT_REPEAT_LENGTH = 10;
+int Constants::RLE_MIN_REPEAT = 3;
+int Constants::INT_RLE_MAX_SCOPE = 512;
+int Constants::INT_RLE_MAX_SHORT_REPEAT = 10;
 float Constants::DICT_KEY_SIZE_THRESHOLD = 0.1F;
 int Constants::INIT_DICT_SIZE = 4096;
 

cpp/pixels-core/lib/encoding/RunLenIntDecoder.cpp

@@ -26,7 +26,7 @@
 
 RunLenIntDecoder::RunLenIntDecoder(const std::shared_ptr <ByteBuffer> &bb, bool isSigned)
 {
-    literals = new long[Constants::MAX_SCOPE];
+    literals = new long[Constants::INT_RLE_MAX_SCOPE];
     inputStream = bb;
     this->isSigned = isSigned;
     numLiterals = 0;
@@ -308,8 +308,8 @@ void RunLenIntDecoder::readShortRepeatValues(int firstByte)
 
     // read the run length
     int len = firstByte & 0x07;
-    // run length values are stored only after MIN_REPEAT value is met
-    len += Constants::MIN_REPEAT;
+    // run length values are stored only after RLE_MIN_REPEAT value is met
+    len += Constants::RLE_MIN_REPEAT;
 
     // read the repeated value which is stored using fixed bytes
     long val = bytesToLongBE(inputStream, size);

cpp/pixels-core/lib/encoding/RunLenIntEncoder.cpp

@@ -48,11 +48,11 @@ RunLenIntEncoder::RunLenIntEncoder(bool isSigned, bool isAlignedBitPacking) :
     // PENDING: will the byte buffer be used in a buffer pool
     //          so that we do not need to create it here
     outputStream = std::make_shared<ByteBuffer>();
-    literals = new long[Constants::MAX_SCOPE];
-    zigzagLiterals = new long[Constants::MAX_SCOPE];
-    baseRedLiterals = new long[Constants::MAX_SCOPE];
-    adjDeltas = new long[Constants::MAX_SCOPE];
-    gapVsPatchList = new long[Constants::MAX_SCOPE];
+    literals = new long[Constants::INT_RLE_MAX_SCOPE];
+    zigzagLiterals = new long[Constants::INT_RLE_MAX_SCOPE];
+    baseRedLiterals = new long[Constants::INT_RLE_MAX_SCOPE];
+    adjDeltas = new long[Constants::INT_RLE_MAX_SCOPE];
+    gapVsPatchList = new long[Constants::INT_RLE_MAX_SCOPE];
     clear();
 }
 
@@ -137,9 +137,9 @@ void RunLenIntEncoder::determineEncoding()
     zzBits100p = percentileBits(zigzagLiterals, 0, numLiterals, 1.0);
 
     // less than min repeat num so direct encoding
-    if (numLiterals <= Constants::MIN_REPEAT)
+    if (numLiterals <= Constants::RLE_MIN_REPEAT)
     {
-        // std::cout << "numLiterals <= Constants::MIN_REPEAT" << std::endl;
+        // std::cout << "numLiterals <= Constants::RLE_MIN_REPEAT" << std::endl;
         encodingType = EncodingType::DIRECT;
         return;
     }
@@ -467,7 +467,7 @@ void RunLenIntEncoder::writeShortRepeatValues()
     header |= ((numBytesRepeatVal - 1) << 3);
 
     // repeat count (3 bits, 3~10 values)
-    fixedRunLength -= Constants::MIN_REPEAT;
+    fixedRunLength -= Constants::RLE_MIN_REPEAT;
     header |= fixedRunLength;
 
     // write header
@@ -602,7 +602,7 @@ void RunLenIntEncoder::writeDeltaValues()
         // if fixed run length is greater than threshold then it will be fixed
         // delta sequence with delta value 0 else fixed delta sequence with
         // non-zero delta value
-        if (fixedRunLength > Constants::MIN_REPEAT)
+        if (fixedRunLength > Constants::RLE_MIN_REPEAT)
         {
             // ex. sequence: 2 2 2 2 2 2 2 2
             len = fixedRunLength - 1;
@@ -815,26 +815,26 @@ void RunLenIntEncoder::write(long value)
                 fixedRunLength += 1;
 
                 // if fixed run len meets the minimum repeat threshold, and variable len is non-zero
-                if (fixedRunLength >= Constants::MIN_REPEAT && variableRunLength > 0)
+                if (fixedRunLength >= Constants::RLE_MIN_REPEAT && variableRunLength > 0)
                 {
-                    numLiterals -= Constants::MIN_REPEAT;
+                    numLiterals -= Constants::RLE_MIN_REPEAT;
                     // before entering this branch, last (min_repeat - 1) same values are counted into variable run
-                    variableRunLength -= (Constants::MIN_REPEAT - 1);
-                    long *tailVals = new long[Constants::MIN_REPEAT];
+                    variableRunLength -= (Constants::RLE_MIN_REPEAT - 1);
+                    long *tailVals = new long[Constants::RLE_MIN_REPEAT];
                     // copy out the current fixed run part
                     // PENDING: can we use memcpy here?
-                    std::memcpy(tailVals, literals + numLiterals, Constants::MIN_REPEAT * sizeof(long));
+                    std::memcpy(tailVals, literals + numLiterals, Constants::RLE_MIN_REPEAT * sizeof(long));
                     // flush the variable run  
                     determineEncoding();
                     writeValues();
                     // shift the tail fixed runs to the start of the buffer
-                    memcpy(literals + numLiterals, tailVals, Constants::MIN_REPEAT * sizeof(long));
-                    numLiterals += Constants::MIN_REPEAT;
+                    memcpy(literals + numLiterals, tailVals, Constants::RLE_MIN_REPEAT * sizeof(long));
+                    numLiterals += Constants::RLE_MIN_REPEAT;
                     delete[] tailVals;
 
                 }
 
-                if (fixedRunLength == Constants::MAX_SCOPE)
+                if (fixedRunLength == Constants::INT_RLE_MAX_SCOPE)
                 {
                     determineEncoding();
                     writeValues();
@@ -844,10 +844,10 @@ void RunLenIntEncoder::write(long value)
             else
             {
                 // if fixed run length meets the minimum repeat threshold
-                if (fixedRunLength >= Constants::MIN_REPEAT)
+                if (fixedRunLength >= Constants::RLE_MIN_REPEAT)
                 {
                     // if meets the short repeat condition, write values as short repeats
-                    if (fixedRunLength <= Constants::MAX_SHORT_REPEAT_LENGTH)
+                    if (fixedRunLength <= Constants::INT_RLE_MAX_SHORT_REPEAT)
                     {
                         encodingType = EncodingType::SHORT_REPEAT;
                         writeValues();
@@ -864,7 +864,7 @@ void RunLenIntEncoder::write(long value)
                 // if fixed run length is smaller than the minimum repeat threshold
                 // and current value is different from previous one
                 // it is a variable run
-                if (fixedRunLength > 0 && fixedRunLength < Constants::MIN_REPEAT)
+                if (fixedRunLength > 0 && fixedRunLength < Constants::RLE_MIN_REPEAT)
                 {
                     if (value != literals[numLiterals - 1])
                     {
@@ -886,7 +886,7 @@ void RunLenIntEncoder::write(long value)
                     variableRunLength += 1;
 
                     // flush variable run if it reaches the max scope
-                    if (variableRunLength == Constants::MAX_SCOPE)
+                    if (variableRunLength == Constants::INT_RLE_MAX_SCOPE)
                     {
                         determineEncoding();
                         writeValues();
@@ -908,15 +908,15 @@ void RunLenIntEncoder::flush()
         }
         else if (fixedRunLength != 0)
         {
-            if (fixedRunLength < Constants::MIN_REPEAT)
+            if (fixedRunLength < Constants::RLE_MIN_REPEAT)
             {
                 variableRunLength = fixedRunLength;
                 fixedRunLength = 0;
                 determineEncoding();
                 writeValues();
             }
-            else if (fixedRunLength >= Constants::MIN_REPEAT
-                     && fixedRunLength <= Constants::MAX_SHORT_REPEAT_LENGTH)
+            else if (fixedRunLength >= Constants::RLE_MIN_REPEAT
+                     && fixedRunLength <= Constants::INT_RLE_MAX_SHORT_REPEAT)
             {
                 encodingType = EncodingType::SHORT_REPEAT;
                 writeValues();

pixels-common/src/main/java/io/pixelsdb/pixels/common/utils/Constants.java

@@ -39,9 +39,12 @@ public final class Constants
     public static final int STREAM_READER_RG_BUFFER_SIZE = 1024 * 1024;
     public static final int STREAM_READER_RG_FOOTER_BUFFER_SIZE = 1024;
 
-    public static final int MIN_REPEAT = 3;
-    public static final int MAX_SCOPE = 512;
-    public static final int MAX_SHORT_REPEAT_LENGTH = 10;
+    public static final int RLE_MIN_REPEAT = 3;
+    public static final int INT_RLE_MAX_SCOPE = 512;
+    public static final int INT_RLE_MAX_SHORT_REPEAT = 10;
+    public static final int BYTE_RLE_MAX_LITERAL_SIZE = 128;
+    public static final int BYTE_RLE_MAX_REPEAT_SIZE = 127 + RLE_MIN_REPEAT;
+
     public static final float DICT_KEY_SIZE_THRESHOLD = 0.1F;
     public static final int INIT_DICT_SIZE = 4096;
     public static final int MAX_STREAM_RETRY_COUNT = 100;

pixels-core/src/main/java/io/pixelsdb/pixels/core/encoding/RunLenByteDecoder.java

@@ -19,27 +19,105 @@
  */
 package io.pixelsdb.pixels.core.encoding;
 
+import io.pixelsdb.pixels.common.utils.Constants;
+
+import java.io.EOFException;
 import java.io.IOException;
+import java.io.InputStream;
 
 /**
+ * A decoder for a sequence of bytes encoded by {@link RunLenByteEncoder}.
+ * A control byte is read before each run with positive values 0 to 127 meaning 3 to 130
+ * repetitions. If the byte is -1 to -128, 1 to 128 literal byte values follow.
+ *
  * @author guodong
  * @author hank
  */
 public class RunLenByteDecoder extends Decoder
 {
+    private final InputStream inputStream;
+    private final byte[] literals = new byte[Constants.BYTE_RLE_MAX_LITERAL_SIZE];
+    private int numLiterals = 0;
+    private int used = 0;
+    private boolean repeat = false;
+
+    public RunLenByteDecoder(InputStream inputStream)
+    {
+        this.inputStream = inputStream;
+    }
+
+    public byte next() throws IOException
+    {
+        if (used == numLiterals)
+        {
+            readValues();
+        }
+        if (repeat)
+        {
+            used += 1;
+            return literals[0];
+        }
+        else
+        {
+            return literals[used++];
+        }
+    }
+
     @Override
     public boolean hasNext() throws IOException
     {
-        return false;
+        return used != numLiterals || inputStream.available() > 0;
     }
 
     @Override
-    public void close()
+    public void close() throws IOException
     {
+        if (inputStream != null)
+        {
+            inputStream.close();
+        }
     }
 
-    public byte next()
+    private void readValues() throws IOException
     {
-        return (byte) 1;
+        int nextByte = inputStream.read();
+        if (nextByte == -1)
+        {
+            throw new EOFException("Read past end of buffer RLE byte");
+        }
+
+        int control = (byte) nextByte;
+        int runLength;
+        if (control >= 0)
+        {
+            // repeat: control 0..127 means 3..130 repetitions
+            int val = inputStream.read();
+            if (val == -1)
+            {
+                throw new EOFException("Reading RLE byte got EOF");
+            }
+            literals[0] = (byte) val;
+            runLength = control + Constants.RLE_MIN_REPEAT;
+        }
+        else
+        {
+            // literal: control -1..-128 means 1..128 literal bytes
+            runLength = -control;
+            int bytes = 0;
+            while (bytes < runLength)
+            {
+                int result = inputStream.read(literals, bytes, runLength - bytes);
+                if (result <= 0)
+                {
+                    throw new EOFException("Reading RLE byte literal got EOF");
+                }
+                bytes += result;
+            }
+        }
+
+        // Publish a run only after its complete payload has been read.
+        repeat = control >= 0;
+        used = 0;
+        numLiterals = runLength;
     }
 }

pixels-core/src/main/java/io/pixelsdb/pixels/core/encoding/RunLenByteEncoder.java

@@ -19,24 +19,22 @@
  */
 package io.pixelsdb.pixels.core.encoding;
 
+import io.pixelsdb.pixels.common.utils.Constants;
+
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 
 /**
  * A encoder for a sequence of bytes.
- * A control byte is written before each run with positive values 0 to 127 meaning 2 to 129 repetitions.
+ * A control byte is written before each run with positive values 0 to 127 meaning 3 to 130 repetitions.
  * If the bytes is -1 to -128, 1 to 128 literal byte values follow.
  *
  * @author guodong
  */
 public class RunLenByteEncoder extends Encoder
 {
-    private static final int MIN_REPEAT_SIZE = 3;
-    private static final int MAX_LITERAL_SIZE = 128;
-    private static final int MAX_REPEAT_SIZE = 127 + MIN_REPEAT_SIZE;
-
     private final ByteArrayOutputStream output;
-    private final byte[] literals = new byte[MAX_LITERAL_SIZE];
+    private final byte[] literals = new byte[Constants.BYTE_RLE_MAX_LITERAL_SIZE];
     private int numLiterals = 0;
     private boolean repeat = false;
     private int tailRunLength = 0;
@@ -87,7 +85,7 @@ private void writeValues()
         {
             if (repeat)
             {
-                output.write(numLiterals - MIN_REPEAT_SIZE);
+                output.write(numLiterals - Constants.RLE_MIN_REPEAT);
                 output.write(literals, 0, 1);
             }
             else
@@ -120,7 +118,7 @@ else if (repeat)
             if (value == literals[0])
             {
                 numLiterals += 1;
-                if (numLiterals == MAX_REPEAT_SIZE)
+                if (numLiterals == Constants.BYTE_RLE_MAX_REPEAT_SIZE)
                 {
                     writeValues();
                 }
@@ -142,26 +140,26 @@ else if (repeat)
             {
                 tailRunLength = 1;
             }
-            if (tailRunLength == MIN_REPEAT_SIZE)
+            if (tailRunLength == Constants.RLE_MIN_REPEAT)
             {
-                if (numLiterals + 1 == MIN_REPEAT_SIZE)
+                if (numLiterals + 1 == Constants.RLE_MIN_REPEAT)
                 {
                     repeat = true;
                     numLiterals += 1;
                 }
                 else
                 {
-                    numLiterals -= MIN_REPEAT_SIZE - 1;
+                    numLiterals -= Constants.RLE_MIN_REPEAT - 1;
                     writeValues();
                     literals[0] = value;
                     repeat = true;
-                    numLiterals = MIN_REPEAT_SIZE;
+                    numLiterals = Constants.RLE_MIN_REPEAT;
                 }
             }
             else
             {
                 literals[numLiterals++] = value;
-                if (numLiterals == MAX_LITERAL_SIZE)
+                if (numLiterals == Constants.BYTE_RLE_MAX_LITERAL_SIZE)
                 {
                     writeValues();
                 }