Implement some encodings.

Author: Nong Li

src/CMakeLists.txt

@@ -12,7 +12,6 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-
 add_library(Parquet STATIC
   parquet.cc
 )

src/encodings.h

@@ -22,27 +22,38 @@ class Decoder {
   virtual float GetFloat() { return 0; }
   virtual String GetString() { return String(); }
 
+  int value_left() const { return num_values_; }
+
  protected:
-  Decoder(parquet::SchemaElement* schema) : schema_(schema) {}
+  Decoder(parquet::SchemaElement* schema, int num_values)
+    : schema_(schema), num_values_(num_values) {}
   parquet::SchemaElement* schema_;
+  int num_values_;
 };
 
 class BoolDecoder : public Decoder {
  public:
-  BoolDecoder(parquet::SchemaElement* schema) : Decoder(schema) { }
+  BoolDecoder(parquet::SchemaElement* schema, int num_values) : Decoder(schema, num_values) { }
 
   virtual void SetData(const uint8_t* data, int len) {
+    decoder_ = impala::RleDecoder(data, len, 1);
+  }
+
+  virtual bool GetBool() {
+    bool result;
+    if (!decoder_.Get(&result)) throw "EOF";
+    --num_values_;
+    return result;
   }
 
- virtual bool GetBool() {
-   return false;
- }
+ private:
+  impala::RleDecoder decoder_;
 };
 
 class PlainDecoder : public Decoder {
  public:
-  PlainDecoder(parquet::SchemaElement* schema)
-    : Decoder(schema), data_(NULL), len_(0) {
+  PlainDecoder(parquet::SchemaElement* schema, int num_values)
+    : Decoder(schema, num_values), data_(NULL), len_(0) {
   }
 
   virtual void SetData(const uint8_t* data, int len) {
@@ -51,39 +62,43 @@ class PlainDecoder : public Decoder {
   }
 
   virtual int32_t GetInt32() {
-    if (len_ < sizeof(int32_t)) throw "Bad";
+    if (len_ < sizeof(int32_t)) throw "EOF";
     int32_t val = *reinterpret_cast<const int32_t*>(data_);
     data_ += sizeof(int32_t);
     len_ -= sizeof(int32_t);
+    --num_values_;
     return val;
   }
 
   virtual int64_t GetInt64() {
-    if (len_ < sizeof(int64_t)) throw "Bad";
+    if (len_ < sizeof(int64_t)) throw "EOF";
     int64_t val = *reinterpret_cast<const int64_t*>(data_);
     data_ += sizeof(int64_t);
     len_ -= sizeof(int64_t);
+    --num_values_;
     return val;
   }
 
   virtual float GetFloat() {
-    if (len_ < sizeof(float)) throw "Bad";
+    if (len_ < sizeof(float)) throw "EOF";
     float val = *reinterpret_cast<const float*>(data_);
     data_ += sizeof(float);
     len_ -= sizeof(float);
+    --num_values_;
     return val;
   }
 
   virtual String GetString() {
     String result;
-    if (len_ < sizeof(uint32_t)) throw "Bad";
+    if (len_ < sizeof(uint32_t)) throw "EOF";
     result.len = *reinterpret_cast<const uint32_t*>(data_);
     data_ += sizeof(uint32_t);
     len_ -= sizeof(uint32_t);
-    if (len_ < result.len) throw "Bad";
+    if (len_ < result.len) throw "EOF";
     result.ptr = data_;
     data_ += result.len;
     len_ -= result.len;
+    --num_values_;
     return result;
   }
 
@@ -94,25 +109,68 @@ class PlainDecoder : public Decoder {
 
 class DictionaryDecoder : public Decoder {
  public:
-  DictionaryDecoder(parquet::SchemaElement* schema, Decoder* dictionary)
-    : Decoder(schema) {
+  DictionaryDecoder(parquet::SchemaElement* schema, int num_values, Decoder* dictionary)
+    : Decoder(schema, num_values) {
+    int num_dictionary_values = dictionary->value_left();
+    switch (schema->type) {
+      case parquet::Type::BOOLEAN: throw "Boolean cols should not be dictionary encoded.";
+      case parquet::Type::INT32:
+        int32_dictionary_.resize(num_dictionary_values);
+        for (int i = 0; i < num_dictionary_values; ++i) {
+          int32_dictionary_[i] = dictionary->GetInt32();
+        }
+        break;
+      case parquet::Type::INT64:
+        int64_dictionary_.resize(num_dictionary_values);
+        for (int i = 0; i < num_dictionary_values; ++i) {
+          int64_dictionary_[i] = dictionary->GetInt64();
+        }
+        break;
+      case parquet::Type::FLOAT:
+        float_dictionary_.resize(num_dictionary_values);
+        for (int i = 0; i < num_dictionary_values; ++i) {
+          float_dictionary_[i] = dictionary->GetFloat();
+        }
+        break;
+      case parquet::Type::BYTE_ARRAY:
+        string_dictionary_.resize(num_dictionary_values);
+        for (int i = 0; i < num_dictionary_values; ++i) {
+          string_dictionary_[i] = dictionary->GetString();
+        }
+        break;
+      default:
+        throw "NYI";
+    }
   }
 
   virtual void SetData(const uint8_t* data, int len) {
+    if (len == 0) return;
+    uint8_t bit_width = *data;
+    ++data;
+    --len;
+    idx_decoder_ = impala::RleDecoder(data, len, bit_width);
   }
 
-  virtual int32_t GetInt32() {
-    return 0;
-  }
-  virtual int64_t GetInt64() {
-    return 0;
-  }
-  virtual float GetFloat() {
-    return 0;
-  }
-  virtual String GetString() {
-    return String();
+  virtual int32_t GetInt32() { return int32_dictionary_[index()]; }
+  virtual int64_t GetInt64() { return int64_dictionary_[index()]; }
+  virtual float GetFloat() { return float_dictionary_[index()]; }
+  virtual String GetString() { return string_dictionary_[index()]; }
+
+ private:
+  int index() {
+    int idx;
+    if (!idx_decoder_.Get(&idx)) throw "EOF";
+    --num_values_;
+    return idx;
   }
+
+  // Only one is set.
+  std::vector<int32_t> int32_dictionary_;
+  std::vector<int64_t> int64_dictionary_;
+  std::vector<float> float_dictionary_;
+  std::vector<String> string_dictionary_;
+
+  impala::RleDecoder idx_decoder_;
 };
 
 }

src/impala/bit-stream-utils.h

@@ -92,7 +92,7 @@ class BitWriter {
 class BitReader {
  public:
   // 'buffer' is the buffer to read from.  The buffer's length is 'buffer_len'.
-  BitReader(uint8_t* buffer, int buffer_len) :
+  BitReader(const uint8_t* buffer, int buffer_len) :
       buffer_(buffer),
       max_bytes_(buffer_len),
       byte_offset_(0),
@@ -127,7 +127,7 @@ class BitReader {
   static const int MAX_VLQ_BYTE_LEN = 5;
 
  private:
-  uint8_t* buffer_;
+  const uint8_t* buffer_;
   int max_bytes_;
 
   // Bytes are memcpy'd from buffer_ and values are read from this variable. This is

src/impala/rle-encoding.h

@@ -79,7 +79,7 @@ class RleDecoder {
  public:
   // Create a decoder object. buffer/buffer_len is the decoded data.
   // bit_width is the width of each value (before encoding).
-  RleDecoder(uint8_t* buffer, int buffer_len, int bit_width)
+  RleDecoder(const uint8_t* buffer, int buffer_len, int bit_width)
     : bit_reader_(buffer, buffer_len),
       bit_width_(bit_width),
       current_value_(0),