apache/spark — pull request #829 by liancheng (2014-05-19 21:39:05+00:00)
Title: [SPARK-1669 & SPARK-1379][SQL][WIP] Made SchemaRDD.cache() to leverage InMemoryColumnarTableScan

Corresponding JIRA issues: [SPARK-1669](https://issues.apache.org/jira/browse/SPARK-1669), [SPARK-1379](https://issues.apache.org/jira/browse/SPARK-1379)

(To @rxin @marmbrus: this change seems to bring some subtleties, and makes me doubt whether it's a good idea to make `SchemaRDD.cache()` to leverage in-memory columnar storage. See the "Open issues" section below. Maybe we should simply throw an exception in `SchemaRDD.cache()` to inform people use `cacheTable` instead. Or maybe there are simpler and more intuitive ways that I failed to discover.)

In this PR, we overrode cache related RDD operations in `SchemaRDD` to make `.cache()` to leverage `InMemoryColumnarTableScan`, and made table caching operation idempotent (calling `cacheTable()` multiple times gives the same result).

### Basic ideas

When calling `.cache()` or `.persist()` on a `SchemaRDD`, an `InMemoryColumnarTableScan` is constructed, and the parent RDD of the `SchemaRDD` is replaced with the cached columns RDD returned by `InMemoryColumnarTableScan.execute()`.

Notice that in this way, from the viewpoint of Spark runtime, the RDD being cached is the cached columns RDD rather than the `SchemaRDD` itself.

### Updated APIs

- `SchemaRDD.persist(newLevel: StorageLevel)`

Builds in-memory columnar cache and replace the parent RDD if `newLevel.useMemory` is true, otherwise, fallback to the normal caching mechanism.

- `SchemaRDD.unpersist(blocking: Boolea)`

Unpersist the in-memory columnar cache and restore the parent RDD if necessary.

- `SchemaRDD.logicalPlan`

Returns the in-memory columnar version of logical plan if the `SchemaRDD` is cached in memory.

- `SQLContext.cacheTable(tableName: String)`

Forwards to `SQLContext.persistTable(tableName: String, newLevel: StorageLevel)` and leverages in-memory columnar storage if necessary.

- `SQLContext.uncacheTable(tableName: String)`

Forwards to `SQLContext.unpersistTable(tableName: String, blocking: Boolean)`, and re-registers the uncached table if it's backed with an existing RDD.

### New APIs introduced

- `SQLContext.persistTable(tableName: String, newLevel: StorageLevel)`

Equivalents to `table(tableName).persist(newLevel).registerAsTable(tableName)`.

- `SQLContext.unpersistTable(blocking: Boolean)`

Similar to `table(tableName).unpersist(blocking)`, despite that it re-registers uncached table if necessary.

### Open issues

Now we can cache / uncache a table in two styles. In the plain old way:

```scala
val schemaRdd = ...
schemaRdd.registerAsTable("t")
// Caching
cacheTable("t")
// Uncaching
uncacheTable("t")
```

In this way, the name of the cached table is guaranteed to be the same as the original table.

The new, more RDD-style way:

```scala
val schemaRdd = ...
schemaRdd.registerAsTable("t")
// Caching
schemaRdd.cache().registerAsTable("t_cached")
// Uncaching (plan A: OK)
schemaRdd.uncache()
// Uncaching (plan B: OK? A little complicated...)
table("t_cached").uncache()
// Uncaching (plan C: probably WRONG)
uncacheTable("t_cached")
```

A major difference here is that in the new way, the name of the cached table is not necessary the same as the original one. So plan C is not OK because `uncacheTable` tries to re-register the original RDD with the wrong table name, and there ends up two uncached tables named `t` and `t_cached` respectively.

In plan B, `table("t_cached")` returns a new RDD, although the memory occupied by the columnar storage is released, `schemaRdd` is left untouched and will be cached again in subsequent jobs even if no `.cache()` is not called.

Another confusion is:

```scala
val productRdd: RDD[SomeCaseClass] = ...
productRdd.cache().registerAsTable("t")
```

Although looks very similar to the second way of caching a `SchemaRDD` table, in this case, no table is cached, since `productRdd` is not a `SchemaRDD`.