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

Author: sabir-akhadov

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>`"

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

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 =>

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")
+    }
+  }
 }