Compressing Columns Stored in Memory

Functionality to compress columns stored in memory (cached tables).
This is a cleaned-up version of https://github.com/amplab/shark/pull/103

To see a more friendly version of this document go to -
https://github.com/junkumar/shark/wiki/Compressing-Columns-Stored-in-Memory

** Encoding schemes available
| *Scheme*                  | *Target column properties*                                     | *Column Types Currently Supported* |
| Run Length Encoding(RLE)  | sorted OR small number of transitions OR Partition key columns | Int or String                      |
| Dictionary Encoding(Dict) | small number of unique values, but no runs                     | Int                                |
| Null Bit Vector Encoding  | sparse data, few non-null values                               | Int, Boolean, Long or Byte         |
| LZF Encoding              | slowest decompression speed  but best compression rate         | Int or String                      |
| NONE                      | raw storage; ideal if data has too many unique values          |                                    |


** Currently possible mixing of Encoding Schemes and column types
Nullable Column Types - {Byte, Short, Int, Long, Float, Double, Timestamp}

Null Encoding - Any Nullable Column Type
(Null Encoding is always optional when available)

RLE + Null - Int
RLE String

Dict + Null - Int

LZF - String


** How does it get used?
When a cached table is created in shark, all the data gets stored as columns
into memory. The compression is then used to store these columns based on their
properties.


** Compression results on production data at Yahoo
Tested on a 14 node cluster with a total of (14GB*14) 196 GB of RAM
Input data is a single table on HDFS/Hive with 4 parition keys and 3 other columns.

| *partition specification in input data* | *number of rows* | *HDFS bytes (RCfile compressed on disk)*  | *shark 0.7 bytes in memory (no compr)* | *after compression bytes*  | *compr rate (% of uncompr)* |
| action=ad_click p=1 dt=20130501         | 289,652          | 2,683,931     | 23,315,012                   | 8,832,576      |                       37.88 |
| p=1 dt=20130501                         | 153,494,679      | 964,635,051   | 13,602,326,763               | 2,565,180,073  |                       18.86 |
| action=search dt=20130501               | 226,388,853      | 7,350,168,969 | 18,066,874,698               | 6,353,214,544  |                       35.16 |
| action=search dt=2013050[1-5]           | 1,146,965,502    |               |                              | 91,496,897,244 |                             |


6 of the columns in this data were strings with no sort order and the seventh column was a integer column.
RLE was chosen automatically whenever compression got used. Especially on the 4
partition columns.

** Uncompression results 
We are building simpler "light-weight" compression schemes instead of just using
established "heavy-weight" schemes like gzip or LZF so that we are not limited
by uncompression speeds. Heavy-weight schemes work well for data on disk because
disk is much slower than the CPU uncompression. Memory access speeds are much
faster and CPU uncompression can become a bottleneck.

This comparison measures uncompression speeds across multiple schemes. Input
data is a single slice of the data from the last test tested on a single node
setup. This is a single string column with data in sorted order, with a small
number of unique values; expect very high compression rates. These synthetic
benchmarks were simply used to check our implementation beliefs about the
bottleneck and do not numerically model how things will work in the production
clusters.

50 trial runs were executed (with the first 20 not counted) in each case to
measure averages.

A comparison of memory throughput versus data throughput is a useful way for us to reason about our implementations. The memory throughput for "None" compression will always be the best. But the data throughput will depend on how compressible the data is. The more that the data compresses the less space it will take and higher the data throughput will be for a fixed memory throughput.

[[https://dl.dropboxusercontent.com/u/7259597/SharkDecompressionSpeeds.jpg]]

| *scheme* | *rows*     |    *ms* |  *data MBps* |  *mem MBps* |   *% compr rate* |
| None     | 2,500,000  |   74.87 |     413.99 |     541.38 |         100.00 |
| LZF      | 2,500,000  |   75.50 |     410.52 |      21.92 |           4.08 |
| RLE      | 2,500,000  |   34.56 |     896.65 |      62.61 |           5.34 |
| None     | 20,500,000 |  706.50 |     359.74 |     470.43 |         100.00 |
| LZF      | 20,500,000 |  765.30 |     332.10 |      17.73 |           4.08 |
| RLE      | 20,500,000 |  326.87 |     777.55 |      54.30 |           5.34 |
| None     | 90,500,000 | 3126.03 |     358.92 |     469.36 |         100.00 |
| LZF      | 90,500,000 | 3071.10 |     365.34 |      19.50 |           4.08 |
| RLE      | 90,500,000 | 1539.23 |     728.93 |      50.90 |           5.34 |


*** Observations
- Zero compression (None) yields the highest raw memory throughput - the CPU is 
  interfering the least in this case.
- In a classic space-time trade-off, the fact that compressed data uses less
  memory allows nearly the same overall data throughput (raw MBps) even
  though the memory throughput (mem MBps) is lower due to uncompression.
- LZF has better compression rates for this data but cannot keep up with either
  no compression or RLE in term of uncompression speeds.
- RLE offers us the best compromise by allowing us to maintain the overall access
  speed while still letting us store more data into memory.
- Ultimately, like other columnar stores, we will be able to work directly with
  compressed data without needing uncompression which will make RLE's advantage
  even bigger.