[MSPARK-107]idField information is lost in MapRDBDataFrameWriterFunctions.saveToMapRDB (apache#186)

(cherry picked from commit 5430a1d)

Author: rsotn-mapr

external/maprdb/src/main/scala/com/mapr/db/spark/sql/DefaultSource.scala

@@ -48,12 +48,14 @@ class DefaultSource extends DataSourceRegister
                               parameters: Map[String, String], data: DataFrame): BaseRelation = {
 
     require(parameters.get("tableName").isDefined)
+    require(parameters.get("idFieldPath").isDefined)
+    val idFieldPath = parameters("idFieldPath")
     //val cParser = new ConditionParser();
     val condition : Option[QueryCondition] = parameters.get("QueryCondition").map(cond => ConditionImpl.parseFrom(ByteBuffer.wrap(cond.getBytes)))
     lazy val tableExists = DBClient().tableExists(parameters.get("tableName").get)
     lazy val tableName = parameters.get("tableName").get
     lazy val createTheTable = if (tableExists) false else true
-    lazy val bulkMode  = parameters.get("bulkMode").getOrElse("false").toBoolean
+    lazy val bulkMode = parameters.get("bulkMode").getOrElse("false").toBoolean
     val operation = parameters.get("Operation").getOrElse("ErrorIfExists")
     mode match {
       case ErrorIfExists => {}
@@ -62,25 +64,25 @@ class DefaultSource extends DataSourceRegister
 
     operation match {
       case "Insert" => {
-        MapRSpark.insert(data, tableName,"_id", createTable = createTheTable, bulkInsert = bulkMode)
+        MapRSpark.insert(data, tableName,idFieldPath, createTable = createTheTable, bulkInsert = bulkMode)
       }
 
       case "InsertOrReplace" => {
-        MapRSpark.save(data, tableName,"_id",createTable = createTheTable, bulkInsert = bulkMode)
+        MapRSpark.save(data, tableName,idFieldPath,createTable = createTheTable, bulkInsert = bulkMode)
       }
 
       case "ErrorIfExists" => {
         if (tableExists) throw new TableExistsException("Table: " + tableName + " already Exists")
-        else MapRSpark.save(data, tableName,"_id",createTable = true, bulkInsert = bulkMode)
+        else MapRSpark.save(data, tableName,idFieldPath,createTable = true, bulkInsert = bulkMode)
       }
 
       case "Overwrite" => {
         DBClient().deleteTable(tableName)
-        MapRSpark.save(data, tableName,"_id",createTable = true, bulkInsert = bulkMode)
+        MapRSpark.save(data, tableName,idFieldPath,createTable = true, bulkInsert = bulkMode)
       }
 
       case "Update" => {
-        MapRSpark.update(data, tableName,"_id",createTable = false, bulkInsert = bulkMode)
+        MapRSpark.update(data, tableName,idFieldPath,createTable = false, bulkInsert = bulkMode)
       }
 
       case _ => throw new UnsupportedOperationException("Not supported operation")    }

external/maprdb/src/main/scala/com/mapr/db/spark/sql/MapRDBDataFrameWriterFunctions.scala

@@ -7,6 +7,6 @@ import org.apache.spark.sql.DataFrameWriter
 private[spark] case class MapRDBDataFrameWriterFunctions(@transient dfw: DataFrameWriter[_]) extends LoggingTrait {
 
   def saveToMapRDB(tableName: String, idFieldPath: String = "_id",
-                   bulkInsert : Boolean = false): Unit = MapRSpark.save(dfw, tableName)
+                   bulkInsert : Boolean = false): Unit = MapRSpark.save(dfw, tableName, idFieldPath, bulkInsert)
 
 }

external/maprdb/src/main/scala/com/mapr/db/spark/utils/MapRSpark.scala

@@ -53,8 +53,17 @@ object MapRSpark {
     documentRdd.updateToMapRDB(tableName, mutation, getID, condition)
   }
 
-  def save(dfw: DataFrameWriter[_], tableName: String): Unit = {
-    dfw.format(defaultSource).option("tableName", tableName).save()
+  def save(
+           dfw: DataFrameWriter[_],
+           tableName: String,
+           idFieldPath: String,
+           bulkInsert: Boolean
+          ): Unit = {
+    dfw.format(defaultSource)
+      .option("tableName", tableName)
+      .option("idFieldPath", idFieldPath)
+      .option("bulkMode", bulkInsert)
+      .save()
   }
 
   def load(sc: SparkContext, tableName: String) : MapRDBBaseRDD[OJAIDocument] = {