Support index upsert

[zhexuany]

In this PR, we support pk and index upsert.

[marsishandsome]

the following test fails

class TestSuite extends BaseDataSourceTest("test_datasource_basic2") {
  private val row3 = Row(3)
  private val row4 = Row(4)

  private val schema = StructType(
    List(
      StructField("i", IntegerType)
    )
  )

  override def beforeAll(): Unit = {
    super.beforeAll()

    dropTable()
    jdbcUpdate(s"create table $dbtable(i int primary key)")
  }

  test("Test Write Append") {
    val data: RDD[Row] = sc.makeRDD(List(row3, row4))
    val df = sqlContext.createDataFrame(data, schema)

    df.write
      .format("tidb")
      .options(tidbOptions)
      .option("database", database)
      .option("table", table)
      .mode("append")
      .save()

    testTiDBSelect(Seq(row3, row4))
  }

  override def afterAll(): Unit =
    try {
      dropTable()
    } finally {
      super.afterAll()
    }
}

error message

== Results ==
!== Correct Answer - 2 ==   == Spark Answer - 1 ==
![3]                        [4]
![4]                        

[zhexuany]

@marsishandsome The test case is due to a bug in encodeRow.
Before

  // We could not set nil value into kv.
    if (values.length == 0) {
      return new byte[] {Codec.NULL_FLAG};
    }

After

  // We could not set nil value into kv.
    if (cdo.toBytes().length == 0) {
      return new byte[] {Codec.NULL_FLAG};
    }

[zhexuany]

/run-all-tests

[marsishandsome]

rebase #843 to let build success

[marsishandsome]

pls update documents in the same PR

[zhexuany]

/run-all-tests

[marsishandsome]

pls add some test in DeduplicateSuite to test the following code

 uniqueIndices.foreach { index =>
      {
        mutableRdd = mutableRdd
          .map { wrappedRow =>
            val indexKey = buildUniqueIndexKey(wrappedRow.row, index)
            (indexKey, wrappedRow)
          }
          .groupByKey()
          .map(_._2.head)
      }
    }

[marsishandsome]

would u pls provide a test design document?

[marsishandsome]

change rdd.map to rdd.mapPartitions, cause if use rdd.map the following code will be called rdd.size times

val snapshot = TiSessionCache.getSession(tiConf).createSnapshot(startTs)

[zhexuany]

It is planned to be done when we support batch get.

[marsishandsome]

it can be done without batch get.

[marsishandsome]

mostly LGTM

[marsishandsome]

pls update this line

TiSpark's `Append` means upsert. If primary key is same, data will be updated; if no same primary key exists, data will be inserted.

[marsishandsome]

LGTM

[zhexuany]

/run-all-tests

[windtalker]

LGTM