Refactor first set of 20 error messages.

This PR is going to refactor the existing exceptions in our code base, so that they are well organized and queryable. This commit refactor the first 20 error messages.

To handle compilation and error class handling, there are two more classes created:
1. The DeltaThrowable: this is used as the based interface of delta code and contains all necessary implementation for error message framework. With its help, the compilation error of OSS Delta caused by outdated Apache Spark can be avoided.

2. The DeltaThrowableHelper: This is used to pick the error class template from JSON files. It handles all exceptions' message for Delta.

Thus, there are 2 JSON files we read:
1. error-classes.json: it stores the error classes for exceptions inside Apache Spark code base. It will be maintained by the Apache Spark community.
2. delta-error-classes.json: it stores the error classes for exceptions inside our Delta code base. This is maintained by Delta Lake.

Note:
0. The error-classes.json and delta-error-classes.json are not allowed to include same error class.
1. Test cases for all error classes are covered.
2. No duplicate error classes in Delta
4. No error classes are shared by Delta and Spark
5. Delta error classes are correctly formatted
6. Delta message format invariants

GitOrigin-RevId: cff83f531faeaa97391f8debad7e8dbe7f18f632

Author: kamcheungting-db

core/src/main/resources/error/delta-error-classes.json

@@ -0,0 +1,77 @@
+{
+  "COMPLEX_TYPE_COLUMN_CANNOT_CONTAIN_NULL_TYPE" : {
+    "message" : [ " Found nested NullType in column %s which is of %s. Delta doesn't support writing NullType in complex types." ],
+    "sqlState" : "22005"
+  },
+  "CONFLICT_SET_COLUMN" : {
+    "message" : [ "There is a conflict from these SET columns: %s." ],
+    "sqlState" : "42000"
+  },
+  "DELTA_MERGE_UNEXPECTED_ASSIGNMENT_KEY" : {
+    "message" : [ "Unexpected assignment key: %s - %s" ],
+    "sqlState" : "22005"
+  },
+  "DELTA_TABLE_ONLY_OPERATION" : {
+    "message" : [ "%s is not a Delta table. %s is only supported for Delta tables." ],
+    "sqlState" : "0A000"
+  },
+  "FAILED_CHECKPOINT" : {
+    "message" : [ "Cannot rename %s to %s" ]
+  },
+  "INVALID_CHARACTERS_IN_COLUMN_NAME" : {
+    "message" : [ "Attribute name \"%s\" contains invalid character(s) among \" ,;{}()\\\\n\\\\t=\". Please use alias to rename it." ],
+    "sqlState" : "42000"
+  },
+  "INVALID_COPY_ENCRYPTION" : {
+    "message" : [ "Invalid %s. COPY INTO source %s." ],
+    "sqlState" : "42000"
+  },
+  "INVALID_GENERATED_COLUMN_REFERENCES" : {
+    "message" : [ "A generated column cannot use a non-existent column or another generated column" ],
+    "sqlState" : "42000"
+  },
+  "MERGE_INCOMPATIBLE_DECIMAL_TYPE" : {
+    "message" : [ "Failed to merge decimal types with incompatible %s" ],
+    "sqlState" : "22005"
+  },
+  "MISSING_DELTA_TABLE" : {
+    "message" : [ "%s is not a Delta table." ],
+    "sqlState" : "42000"
+  },
+  "MISSING_NOT_NULL_COLUMN_VALUE" : {
+    "message" : [ "Column %s, which has a NOT NULL constraint, is missing from the data being written into the table." ],
+    "sqlState" : "42000"
+  },
+  "MISSING_SET_COLUMN" : {
+    "message" : [ "SET column %s not found given columns: %s." ],
+    "sqlState" : "42000"
+  },
+  "UNSUPPORTED_COLUMN_MAPPING_CONVERT_TO_DELTA" : {
+    "message" : [ "The configuration '%s' cannot be set to `%s` when using CONVERT TO DELTA." ],
+    "sqlState" : "0A000"
+  },
+  "UNSUPPORTED_COLUMN_MAPPING_MODE_CHANGE" : {
+    "message" : [ "Changing column mapping mode from '%s' to '%s' is not supported." ],
+    "sqlState" : "0A000"
+  },
+  "UNSUPPORTED_COLUMN_MAPPING_PROTOCOL" : {
+    "message" : [ "", "Your current table protocol version does not support changing column mapping modes", "using %s.", "", "Required Delta protocol version for column mapping:", "%s", "Your table's current Delta protocol version:", "%s", "", "Please upgrade your table's protocol version using ALTER TABLE SET TBLPROPERTIES and try again.", "", "" ],
+    "sqlState" : "0A000"
+  },
+  "UNSUPPORTED_COLUMN_MAPPING_SCHEMA_CHANGE" : {
+    "message" : [ "", "Schema change is detected:", "", "old schema:", "%s", "", "new schema:", "%s", "", "Schema changes are not allowed during the change of column mapping mode.", "", "" ],
+    "sqlState" : "0A000"
+  },
+  "UNSUPPORTED_COLUMN_MAPPING_WRITE" : {
+    "message" : [ "Writing data with column mapping mode is not supported." ],
+    "sqlState" : "0A000"
+  },
+  "UNSUPPORTED_MANIFEST_GENERATION_WITH_COLUMN_MAPPING" : {
+    "message" : [ "Manifest generation is not supported for tables that leverage column mapping, as external readers cannot read these Delta tables. See Databricks documentation for more details." ],
+    "sqlState" : "0A000"
+  },
+  "UNSUPPORTED_NESTED_COLUMN_IN_BLOOM_FILTER" : {
+    "message" : [ "Creating a bloom filer index on a nested column is currently unsupported: %s" ],
+    "sqlState" : "0A000"
+  }
+}

core/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/deltaMerge.scala

@@ -18,6 +18,7 @@ package org.apache.spark.sql.catalyst.plans.logical
 
 import java.util.Locale
 
+import org.apache.spark.sql.delta.DeltaAnalysisException
 import org.apache.spark.sql.delta.schema.SchemaMergingUtils
 import org.apache.spark.sql.delta.sources.DeltaSQLConf
 
@@ -150,7 +151,9 @@ object DeltaMergeIntoClause {
         case Assignment(key: UnresolvedAttribute, expr) => DeltaMergeAction(key.nameParts, expr)
         case Assignment(key: Attribute, expr) => DeltaMergeAction(Seq(key.name), expr)
         case other =>
-          throw new AnalysisException(s"Unexpected assignment key: ${other.getClass} - $other")
+          throw new DeltaAnalysisException(
+            errorClass = "DELTA_MERGE_UNEXPECTED_ASSIGNMENT_KEY",
+            messageParameters = Array(s"${other.getClass}", s"$other"))
       }
     }
   }

core/src/main/scala/org/apache/spark/sql/delta/Checkpoints.scala

@@ -376,7 +376,7 @@ object Checkpoints extends DeltaLogging {
         } else {
           // There should be only one writer writing the checkpoint file, so there must be
           // something wrong here.
-          throw new IllegalStateException(s"Cannot rename $src to $dest")
+          throw DeltaErrors.failOnCheckpoint(src, dest)
         }
       } finally {
         if (!renameDone) {

core/src/main/scala/org/apache/spark/sql/delta/DeltaAnalysisException.scala

@@ -0,0 +1,28 @@
+/*
+ * Copyright (2021) The Delta Lake Project Authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.delta
+
+import org.apache.spark.sql.AnalysisException
+
+class DeltaAnalysisException(
+    errorClass: String, messageParameters: Array[String], cause: Option[Throwable] = None)
+  extends AnalysisException(
+    DeltaThrowableHelper.getMessage(errorClass, messageParameters),
+    errorClass = Some(errorClass),
+    messageParameters = messageParameters,
+    cause = cause)
+  with DeltaThrowable

core/src/main/scala/org/apache/spark/sql/delta/DeltaErrors.scala

@@ -157,6 +157,16 @@ object DeltaErrors
     new AnalysisException(s"Can't resolve column ${name} in ${schema.treeString}")
   }
 
+  def failOnCheckpoint(src: Path, dest: Path): DeltaIllegalStateException = {
+    failOnCheckpoint(src.toString, dest.toString)
+  }
+
+  def failOnCheckpoint(src: String, dest: String): DeltaIllegalStateException = {
+    new DeltaIllegalStateException(
+      errorClass = "FAILED_CHECKPOINT",
+      messageParameters = Array(s"$src", s"$dest"))
+  }
+
 
   def formatColumn(colName: String): String = s"`$colName`"
 
@@ -175,9 +185,9 @@ object DeltaErrors
   }
 
   def notNullColumnMissingException(constraint: Constraints.NotNull): Throwable = {
-    new InvariantViolationException(s"Column ${UnresolvedAttribute(constraint.column).name}" +
-      s", which has a NOT NULL constraint, is missing from the data being " +
-      s"written into the table.")
+    new InvariantViolationException(
+      errorClass = "MISSING_NOT_NULL_COLUMN_VALUE",
+      messageParameters = Array(s"${UnresolvedAttribute(constraint.column).name}"))
   }
 
   def nestedNotNullConstraint(
@@ -267,13 +277,16 @@ object DeltaErrors
   }
 
   def notADeltaTableException(deltaTableIdentifier: DeltaTableIdentifier): Throwable = {
-    new AnalysisException(s"$deltaTableIdentifier is not a Delta table.")
+    new DeltaAnalysisException(
+      errorClass = "MISSING_DELTA_TABLE",
+      messageParameters = Array(s"$deltaTableIdentifier"))
   }
 
   def notADeltaTableException(
       operation: String, deltaTableIdentifier: DeltaTableIdentifier): Throwable = {
-    new AnalysisException(s"$deltaTableIdentifier is not a Delta table. " +
-      s"$operation is only supported for Delta tables.")
+    new DeltaAnalysisException(
+      errorClass = "DELTA_TABLE_ONLY_OPERATION",
+      messageParameters = Array(s"$deltaTableIdentifier", s"$operation"))
   }
 
   def notADeltaTableException(operation: String): Throwable = {
@@ -305,10 +318,9 @@ object DeltaErrors
   }
 
   def invalidColumnName(name: String): Throwable = {
-    new AnalysisException(
-      s"""Attribute name "$name" contains invalid character(s) among " ,;{}()\\n\\t=".
-         |Please use alias to rename it.
-       """.stripMargin.split("\n").mkString(" ").trim)
+    new DeltaAnalysisException(
+      errorClass = "INVALID_CHARACTERS_IN_COLUMN_NAME",
+      messageParameters = Array(name))
   }
 
   def invalidPartitionColumn(e: AnalysisException): Throwable = {
@@ -627,13 +639,15 @@ object DeltaErrors
   }
 
   def updateSetColumnNotFoundException(col: String, colList: Seq[String]): Throwable = {
-    new AnalysisException(
-      s"SET column ${formatColumn(col)} not found given columns: ${formatColumnList(colList)}.")
+    new DeltaAnalysisException(
+      errorClass = "MISSING_SET_COLUMN",
+      messageParameters = Array(formatColumn(col), formatColumnList(colList)))
   }
 
   def updateSetConflictException(cols: Seq[String]): Throwable = {
-    new AnalysisException(
-      s"There is a conflict from these SET columns: ${formatColumnList(cols)}.")
+    new DeltaAnalysisException(
+      errorClass = "CONFLICT_SET_COLUMN",
+      messageParameters = Array(formatColumnList(cols)))
   }
 
   def updateNonStructTypeFieldNotSupportedException(col: String, s: DataType): Throwable = {
@@ -654,8 +668,9 @@ object DeltaErrors
   }
 
   def bloomFilterOnNestedColumnNotSupportedException(name: String): Throwable = {
-    new AnalysisException(
-      s"Creating a bloom filer index on a nested column is currently unsupported: $name")
+    new DeltaAnalysisException(
+      errorClass = "UNSUPPORTED_NESTED_COLUMN_IN_BLOOM_FILTER",
+      messageParameters = Array(name))
   }
 
   def bloomFilterOnColumnTypeNotSupportedException(name: String, dataType: DataType): Throwable = {
@@ -1004,6 +1019,18 @@ object DeltaErrors
         s"COPY INTO source encryption currently only supports s3/s3n/s3a/abfss.")
   }
 
+  def copyIntoEncryptionSseCRequired(): Throwable = {
+    new DeltaIllegalArgumentException(
+      errorClass = "INVALID_COPY_ENCRYPTION",
+      messageParameters = Array("encryption type", "encryption must specify 'TYPE' = 'AWS_SSE_C'"))
+  }
+
+  def copyIntoEncryptionMasterKeyRequired(): Throwable = {
+    new DeltaIllegalArgumentException(
+      errorClass = "INVALID_COPY_ENCRYPTION",
+      messageParameters = Array("encryption arguments", "encryption must specify a MASTER_KEY"))
+  }
+
   def copyIntoCredentialsNotAllowedOn(scheme: String): Throwable = {
      new IllegalArgumentException(
       s"Invalid scheme $scheme. " +
@@ -1146,9 +1173,8 @@ object DeltaErrors
   }
 
   def generatedColumnsReferToWrongColumns(e: AnalysisException): Throwable = {
-    new AnalysisException(
-      "A generated column cannot use a non-existent column or another generated column",
-      cause = Some(e))
+    new DeltaAnalysisException(
+      errorClass = "INVALID_GENERATED_COLUMN_REFERENCES", Array.empty, cause = Some(e))
   }
 
   def generatedColumnsUpdateColumnType(current: StructField, update: StructField): Throwable = {
@@ -1235,39 +1261,36 @@ object DeltaErrors
   }
 
   def changeColumnMappingModeNotSupported(oldMode: String, newMode: String): Throwable = {
-    new ColumnMappingUnsupportedException("Changing column mapping mode from" +
-      s" '$oldMode' to '$newMode' is not supported.")
+    new DeltaColumnMappingUnsupportedException(
+      errorClass = "UNSUPPORTED_COLUMN_MAPPING_MODE_CHANGE",
+      messageParameters = Array(oldMode, newMode))
+  }
+
+  def writesWithColumnMappingNotSupported: Throwable = {
+    new DeltaColumnMappingUnsupportedException(
+      errorClass = "UNSUPPORTED_COLUMN_MAPPING_WRITE")
   }
 
   def generateManifestWithColumnMappingNotSupported: Throwable = {
-    new ColumnMappingUnsupportedException("Manifest generation is not supported for tables that " +
-      "leverage column mapping, as external readers cannot read these Delta tables. " +
-      "See Databricks documentation for more details.")
+    new DeltaColumnMappingUnsupportedException(
+      errorClass = "UNSUPPORTED_MANIFEST_GENERATION_WITH_COLUMN_MAPPING")
   }
 
   def convertToDeltaWithColumnMappingNotSupported(mode: DeltaColumnMappingMode): Throwable = {
-    new ColumnMappingUnsupportedException(
-      s"The configuration '${DeltaConfigs.COLUMN_MAPPING_MODE.defaultTablePropertyKey}' " +
-        s"cannot be set to `${mode.name}` when using CONVERT TO DELTA.")
+    new DeltaColumnMappingUnsupportedException(
+      errorClass = "UNSUPPORTED_COLUMN_MAPPING_CONVERT_TO_DELTA",
+      messageParameters = Array(
+        DeltaConfigs.COLUMN_MAPPING_MODE.defaultTablePropertyKey,
+        mode.name))
   }
 
   def changeColumnMappingModeOnOldProtocol(oldProtocol: Protocol): Throwable = {
-    // scalastyle:off line.size.limit
-    new ColumnMappingUnsupportedException(
-      s"""
-         |Your current table protocol version does not support changing column mapping modes
-         |using ${DeltaConfigs.COLUMN_MAPPING_MODE.key}.
-         |
-         |Required Delta protocol version for column mapping:
-         |${DeltaColumnMapping.MIN_PROTOCOL_VERSION}
-         |
-         |Your table's current Delta protocol version:
-         |$oldProtocol
-         |
-         |Please upgrade your table's protocol version using ALTER TABLE SET TBLPROPERTIES and try again.
-         |
-         |""".stripMargin)
-    // scalastyle:on line.size.limit
+    new DeltaColumnMappingUnsupportedException(
+      errorClass = "UNSUPPORTED_COLUMN_MAPPING_PROTOCOL",
+      messageParameters = Array(
+        s"${DeltaConfigs.COLUMN_MAPPING_MODE.key}",
+        s"${DeltaColumnMapping.MIN_PROTOCOL_VERSION.toString}",
+        s"$oldProtocol"))
   }
 
   def columnRenameNotSupported(spark: SparkSession, protocol: Protocol): Throwable = {
@@ -1301,19 +1324,11 @@ object DeltaErrors
   def schemaChangeDuringMappingModeChangeNotSupported(
       oldSchema: StructType,
       newSchema: StructType): Throwable =
-    new ColumnMappingUnsupportedException(
-      s"""
-         |Schema change is detected:
-         |
-         |old schema:
-         |${formatSchema(oldSchema)}
-         |
-         |new schema:
-         |${formatSchema(newSchema)}
-         |
-         |Schema changes are not allowed during the change of column mapping mode.
-         |
-         |""".stripMargin)
+    new DeltaColumnMappingUnsupportedException(
+      errorClass = "UNSUPPORTED_COLUMN_MAPPING_SCHEMA_CHANGE",
+      messageParameters = Array(
+        formatSchema(oldSchema),
+        formatSchema(newSchema)))
 
   def foundInvalidCharsInColumnNames(cause: Throwable): Throwable = {
     var adviceMsg = "Please use alias to rename it."
@@ -1625,6 +1640,35 @@ class MetadataMismatchErrorBuilder {
   }
 }
 
+class DeltaColumnMappingUnsupportedException(
+    errorClass: String,
+    messageParameters: Array[String] = Array.empty)
+  extends ColumnMappingUnsupportedException(
+    DeltaThrowableHelper.getMessage(errorClass, messageParameters))
+    with DeltaThrowable {
+  override def getErrorClass: String = errorClass
+}
+
+class DeltaIllegalArgumentException(
+    errorClass: String,
+    messageParameters: Array[String] = Array.empty,
+    cause: Throwable = null)
+  extends IllegalArgumentException(
+    DeltaThrowableHelper.getMessage(errorClass, messageParameters), cause)
+    with DeltaThrowable {
+  override def getErrorClass: String = errorClass
+}
+
+class DeltaIllegalStateException(
+    errorClass: String,
+    messageParameters: Array[String] = Array.empty,
+    cause: Throwable = null)
+  extends IllegalStateException(
+    DeltaThrowableHelper.getMessage(errorClass, messageParameters), cause)
+  with DeltaThrowable {
+  override def getErrorClass: String = errorClass
+}
+
 /** Errors thrown around column mapping. */
 class ColumnMappingUnsupportedException(msg: String)
   extends UnsupportedOperationException(msg)