[CARMEL-2663] Supporting insert overwrite with static/dynamic partiti…

…ons (delta-io#41)
LantaoJin · Apr 27, 2020 · 815e98e · 815e98e
1 parent 8871c00
commit 815e98e
Showing 1 changed file with 16 additions and 1 deletion.
diff --git a/src/main/scala/org/apache/spark/sql/delta/DeltaLog.scala b/src/main/scala/org/apache/spark/sql/delta/DeltaLog.scala
@@ -692,10 +692,25 @@ class DeltaLog private(
           overwrite: Boolean,
           metrics: Map[String, SQLMetric]): Unit = {
         val mode = if (overwrite) SaveMode.Overwrite else SaveMode.Append
+        val deltaOptions = data.queryExecution.logical match {
+          case InsertIntoDataSource(
+              LogicalRelation(_: HadoopFsRelation with InsertableRelation, _, Some(_), _),
+              _, _, staticPartitions, _, _) =>
+            val predicates = staticPartitions.map {
+              case (key, value) => key + "=" + value
+            }.mkString(" AND ")
+            if (predicates.isEmpty) {
+              Map.empty[String, String]
+            } else {
+              Map(DeltaOptions.REPLACE_WHERE_OPTION -> predicates)
+            }
+          case _ =>
+            Map.empty[String, String]
+        }
         WriteIntoDelta(
           deltaLog = DeltaLog.this,
           mode = mode,
-          new DeltaOptions(Map.empty[String, String], spark.sessionState.conf),
+          new DeltaOptions(deltaOptions, spark.sessionState.conf),
           partitionColumns = partitionSchema.map(_.name),
           configuration = Map.empty,
           data = data,