Disallow overwriteSchema with dynamic partitions overwrite

Disallow overwriteSchema when partitionOverwriteMode is set to dynamic. Otherwise, the table might become corrupted as schemas of newly written partitions would not match the non-overwritten partitions. GitOrigin-RevId: 1012793448c1ffed9a3f8bde507d9fe1ee183803
delta-io · May 10, 2023 · 579a315 · 579a315
1 parent 6556d6f
commit 579a315
Show file tree

Hide file tree

Showing 4 changed files with 36 additions and 2 deletions.
diff --git a/core/src/main/resources/error/delta-error-classes.json b/core/src/main/resources/error/delta-error-classes.json
@@ -1346,6 +1346,12 @@
     ],
     "sqlState" : "0A000"
   },
+  "DELTA_OVERWRITE_SCHEMA_WITH_DYNAMIC_PARTITION_OVERWRITE" : {
+    "message" : [
+      "'overwriteSchema' cannot be used in dynamic partition overwrite mode."
+    ],
+    "sqlState" : "42613"
+  },
   "DELTA_PARTITION_COLUMN_CAST_FAILED" : {
     "message" : [
       "Failed to cast value `<value>` to `<dataType>` for partition column `<columnName>`"

diff --git a/core/src/main/scala/org/apache/spark/sql/delta/DeltaErrors.scala b/core/src/main/scala/org/apache/spark/sql/delta/DeltaErrors.scala
@@ -2403,6 +2403,12 @@ trait DeltaErrorsBase
     )
   }
 
+  def overwriteSchemaUsedWithDynamicPartitionOverwrite(): Throwable = {
+    new DeltaIllegalArgumentException(
+      errorClass = "DELTA_OVERWRITE_SCHEMA_WITH_DYNAMIC_PARTITION_OVERWRITE"
+    )
+  }
+
   def replaceWhereUsedInOverwrite(): Throwable = {
     new DeltaAnalysisException(
       errorClass = "DELTA_REPLACE_WHERE_IN_OVERWRITE", messageParameters = Array.empty

diff --git a/core/src/main/scala/org/apache/spark/sql/delta/commands/WriteIntoDelta.scala b/core/src/main/scala/org/apache/spark/sql/delta/commands/WriteIntoDelta.scala
@@ -235,6 +235,10 @@ case class WriteIntoDelta(
       } else options.isDynamicPartitionOverwriteMode
     }
 
+    if (useDynamicPartitionOverwriteMode && canOverwriteSchema) {
+      throw DeltaErrors.overwriteSchemaUsedWithDynamicPartitionOverwrite()
+    }
+
     // Validate partition predicates
     var containsDataFilters = false
     val replaceWhere = options.replaceWhere.flatMap { replace =>

diff --git a/core/src/test/scala/org/apache/spark/sql/delta/DeltaOptionSuite.scala b/core/src/test/scala/org/apache/spark/sql/delta/DeltaOptionSuite.scala
@@ -19,11 +19,10 @@ package org.apache.spark.sql.delta
 import java.util.Locale
 
 // scalastyle:off import.ordering.noEmptyLine
-import org.apache.spark.sql.delta.DeltaOptions.PARTITION_OVERWRITE_MODE_OPTION
+import org.apache.spark.sql.delta.DeltaOptions.{OVERWRITE_SCHEMA_OPTION, PARTITION_OVERWRITE_MODE_OPTION}
 import org.apache.spark.sql.delta.actions.{Action, FileAction}
 import org.apache.spark.sql.delta.sources.DeltaSQLConf
 import org.apache.spark.sql.delta.test.DeltaSQLCommandTest
-import org.apache.spark.sql.delta.test.DeltaTestImplicits._
 import org.apache.spark.sql.delta.util.FileNames
 import org.apache.commons.io.FileUtils
 import org.apache.parquet.format.CompressionCodec
@@ -283,4 +282,23 @@ class DeltaOptionSuite extends QueryTest
       }
     }
   }
+
+  test("overwriteSchema=true should be invalid with partitionOverwriteMode=dynamic") {
+    withTempDir { tempDir =>
+      val e = intercept[DeltaIllegalArgumentException] {
+        withSQLConf(DeltaSQLConf.DYNAMIC_PARTITION_OVERWRITE_ENABLED.key -> "true") {
+          Seq(1, 2, 3).toDF
+            .withColumn("part", $"value" % 2)
+            .write
+            .mode("overwrite")
+            .format("delta")
+            .partitionBy("part")
+            .option(OVERWRITE_SCHEMA_OPTION, "true")
+            .option(PARTITION_OVERWRITE_MODE_OPTION, "dynamic")
+            .save(tempDir.getAbsolutePath)
+        }
+      }
+      assert(e.getErrorClass == "DELTA_OVERWRITE_SCHEMA_WITH_DYNAMIC_PARTITION_OVERWRITE")
+    }
+  }
 }