title: Is it necessary to add SaveMode Delete, Update and Upsert for Spark jdbc source?
body: <p>Do you think it is necessary to add SaveMode for Delete, Update, and Upsert? Such as: </p>
  
  <ul>
  <li>SaveMode.Delete</li>
  <li>SaveMode.Update</li>
  <li>SaveMode.Upsert</li>
  </ul>
  
  <p>referring to the code: <a href="https://github.com/apache/spark/blob/47d84e4d0e56e14f9402770dceaf0b4302c00e98/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JdbcRelationProvider.scala" rel="nofollow noreferrer">JdbcRelationProvider.scala</a></p>
  
  <p>I've analyzed its code for SaveTable: <a href="https://github.com/apache/spark/blob/868e02533d76d45bff4200d07658105b6004cf46/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JdbcUtils.scala" rel="nofollow noreferrer">JdbcUtils.scala</a>, and think it is easy to extend current insert implementation with delete, update, and merge statements, e.g.:</p>
  
  <pre><code>  def getDeleteStatement(table: String, rddSchema: StructType, dialect: JdbcDialect): String = {
      val columns = rddSchema.fields.map(x =&gt; dialect.quoteIdentifier(x.name) + "=?").mkString(" AND ")
  
      s"DELETE FROM ${table.toUpperCase} WHERE $columns"
    }
  
    def getUpdateStatement(table: String, rddSchema: StructType, priKeys: Seq[String], dialect: JdbcDialect): String = {
      val fullCols = rddSchema.fields.map(x =&gt; dialect.quoteIdentifier(x.name))
      val priCols = priKeys.map(dialect.quoteIdentifier(_))
      val columns = (fullCols diff priCols).map(_ + "=?").mkString(",")
      val cnditns = priCols.map(_ + "=?").mkString(" AND ")
  
      s"UPDATE ${table.toUpperCase} SET $columns WHERE $cnditns"
    }
  
    def getMergeStatement(table: String, rddSchema: StructType, priKeys: Seq[String], dialect: JdbcDialect): String = {
      val fullCols = rddSchema.fields.map(x =&gt; dialect.quoteIdentifier(x.name))
      val priCols = priKeys.map(dialect.quoteIdentifier(_))
      val nrmCols = fullCols diff priCols
  
      val fullPart = fullCols.map(c =&gt; s"${dialect.quoteIdentifier("SRC")}.$c").mkString(",")
      val priPart = priCols.map(c =&gt; s"${dialect.quoteIdentifier("TGT")}.$c=${dialect.quoteIdentifier("SRC")}.$c").mkString(" AND ")
      val nrmPart = nrmCols.map(c =&gt; s"$c=${dialect.quoteIdentifier("SRC")}.$c").mkString(",")
  
      val columns = fullCols.mkString(",")
      val placeholders = fullCols.map(_ =&gt; "?").mkString(",")
  
      s"MERGE INTO ${table.toUpperCase} AS ${dialect.quoteIdentifier("TGT")} " +
        s"USING TABLE(VALUES($placeholders)) " +
        s"AS ${dialect.quoteIdentifier("SRC")}($columns) " +
        s"ON $priPart " +
        s"WHEN NOT MATCHED THEN INSERT ($columns) VALUES ($fullPart) " +
        s"WHEN MATCHED THEN UPDATE SET $nrmPart"
    }
  </code></pre>
  
  <p>The only extra thing is that we need provide primary keys for them, and they could call the same <strong>savePartition</strong> function just replacing insertStmt with runningStmt covering all insert/delete/update/merge. </p>
  
  <pre class="lang-scala prettyprint-override"><code>  def savePartition(
        getConnection: () =&gt; Connection,
        table: String,
        iterator: Iterator[Row],
        rddSchema: StructType,
        insertStmt: String,
        batchSize: Int,
        dialect: JdbcDialect,
        isolationLevel: Int,
        options: JDBCOptions): Iterator[Byte] 
  </code></pre>