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

** Intro
This document details the design of the compression options for data stored in cached tables in shark. The goals are to 
1. Allow shark to store more data in a fixed amount of memory without sacrificing read access rates 
2. Provide direct access to the compressed data so that query execution does not require uncompression.

** How does it get used?
When a cached table is created in shark, all the data gets stored as columns
into memory. Compression can then be used to store these columns more efficiently in memory. The default behavior is to determine whether compression is appropriate and which compression scheme should be used based on the properties of the data being stored in memory.


** Compression 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          |                                    |
| Auto  (Default)               | Shark chooses one of the above compression schemes that it deems ideal based on the properties of the column data     | All                                   |

Null Encoding can be mixed with one of Dict, RLE or LZF Encoding on Int columns

** 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.

** The cost of compression
Disk access times are expected to dominate the time taken to create cache tables. But the cost of compression can be observed as an increase in the time taken to create cache tables and must be traded off against the benefits of storing compressed data in memory.

** Specifying compression in SQL
We expect the "Auto" scheme to work well in most situations. However, this can be overridden in SQL when desired. Currently compression schemes can only be set for all columns of a given data type in a cached table.
=CREATE TABLE test_cached 
TBLPROPERTIES("shark.column.compress.int"="RLE") 
AS SELECT * FROM test;=

=CREATE TABLE test_cached 
TBLPROPERTIES("shark.column.compress.string"="None") 
AS SELECT * FROM test;=