[SC-86316] Convert to Delta can ignore casting partition values

In Convert To Delta, we throw an error when failing to cast a partition value. This PR add a flag so that we can silence the error and fill in nulls. new unit test GitOrigin-RevId: 67eacea8338d2cadbff52babaad715f3fa363a35
varuy322 · Oct 28, 2021 · 61cf007 · 61cf007
1 parent 323e909
commit 61cf007
Show file tree

Hide file tree

Showing 2 changed files with 18 additions and 8 deletions.
diff --git a/core/src/main/scala/org/apache/spark/sql/delta/commands/ConvertToDeltaCommand.scala b/core/src/main/scala/org/apache/spark/sql/delta/commands/ConvertToDeltaCommand.scala
@@ -227,9 +227,8 @@ abstract class ConvertToDeltaCommandBase(
 
       if (mergedConfig != deltaLogConfig) {
         if (deltaLogConfig.nonEmpty &&
-          spark.sessionState.conf.getConf(DeltaSQLConf.DELTA_CONVERT_METADATA_CHECK_ENABLED)) {
-          throw DeltaErrors
-            .convertMetastoreMetadataMismatchException(tableProps, deltaLogConfig)
+            conf.getConf(DeltaSQLConf.DELTA_CONVERT_METADATA_CHECK_ENABLED)) {
+          throw DeltaErrors.convertMetastoreMetadataMismatchException(tableProps, deltaLogConfig)
         }
         val newMetadata = txn.metadata.copy(
           configuration = mergedConfig
@@ -685,11 +684,13 @@ object ConvertToDeltaCommand {
 
       val tz = Option(conf.sessionLocalTimeZone)
       // Check if the partition value can be casted to the provided type
-      partValues.literals.zip(partitionFields).foreach { case (literal, field) =>
-        if (literal.eval() != null && Cast(literal, field.dataType, tz).eval() == null) {
-          val partitionValue = Cast(literal, StringType, tz).eval()
-          val partitionValueStr = Option(partitionValue).map(_.toString).orNull
-          throw DeltaErrors.castPartitionValueException(partitionValueStr, field.dataType)
+      if (!conf.getConf(DeltaSQLConf.DELTA_CONVERT_PARTITION_VALUES_IGNORE_CAST_FAILURE)) {
+        partValues.literals.zip(partitionFields).foreach { case (literal, field) =>
+          if (literal.eval() != null && Cast(literal, field.dataType, tz).eval() == null) {
+            val partitionValue = Cast(literal, StringType, tz).eval()
+            val partitionValueStr = Option(partitionValue).map(_.toString).orNull
+            throw DeltaErrors.castPartitionValueException(partitionValueStr, field.dataType)
+          }
         }
       }
 

diff --git a/core/src/main/scala/org/apache/spark/sql/delta/sources/DeltaSQLConf.scala b/core/src/main/scala/org/apache/spark/sql/delta/sources/DeltaSQLConf.scala
@@ -67,6 +67,15 @@ trait DeltaSQLConfBase {
       .booleanConf
       .createWithDefault(true)
 
+  val DELTA_CONVERT_PARTITION_VALUES_IGNORE_CAST_FAILURE =
+    buildConf("convert.partitionValues.ignoreCastFailure")
+      .doc(
+        """ When converting to Delta, ignore the failure when casting a partition value to
+        | the specified data type, in which case the partition column will be filled with null.
+        """.stripMargin)
+      .booleanConf
+      .createWithDefault(false)
+
   val DELTA_SNAPSHOT_PARTITIONS =
     buildConf("snapshotPartitions")
       .internal()