refactor second batch and test

scottsand-db · Mar 26, 2024 · ddcccf4 · ddcccf4
1 parent dbed7a9
commit ddcccf4
Show file tree

Hide file tree

Showing 4 changed files with 137 additions and 21 deletions.
diff --git a/spark/src/main/resources/error/delta-error-classes.json b/spark/src/main/resources/error/delta-error-classes.json
@@ -1329,6 +1329,12 @@
     ],
     "sqlState" : "40000"
   },
+  "DELTA_MERGE_ADD_VOID_COLUMN" : {
+    "message" : [
+      "Cannot add column <newColumn> with type VOID. Please explicitly specify a non-void type."
+    ],
+    "sqlState" : "42K09"
+  },
   "DELTA_MERGE_INCOMPATIBLE_DATATYPE" : {
     "message" : [
       "Failed to merge incompatible data types <currentDataType> and <updateDataType>"
@@ -1525,6 +1531,12 @@
     ],
     "sqlState" : "42621"
   },
+  "DELTA_NON_DETERMINISTIC_EXPRESSION_IN_GENERATED_COLUMN" : {
+    "message" : [
+      "Found <expr>. A generated column cannot use a non deterministic expression."
+    ],
+    "sqlState" : "42621"
+  },
   "DELTA_NON_DETERMINISTIC_FUNCTION_NOT_SUPPORTED" : {
     "message" : [
       "Non-deterministic functions are not supported in the <operation> <expression>"
@@ -2713,5 +2725,29 @@
       "<configuration>",
       "If you would like to merge the configurations (update existing fields and insert new ones), set the SQL configuration `<metadataCheckSqlConf>` to false."
     ]
+  },
+  "_LEGACY_ERROR_TEMP_DELTA_0004" : {
+    "message" : [
+      "Cannot <operation> column <columnName> because this column is referenced by the following",
+      "check constraint(s):",
+      "<constraints>"
+    ]
+  },
+  "_LEGACY_ERROR_TEMP_DELTA_0005" : {
+    "message" : [
+      "Cannot <operation> column <columnName> because this column is referenced by the following",
+      "generated column(s):",
+      "<generatedColumns>"
+    ]
+  },
+  "_LEGACY_ERROR_TEMP_DELTA_0006" : {
+    "message" : [
+      "Inconsistent IDENTITY metadata for column <colName> detected: <hasStart>, <hasStep>, <hasInsert>"
+    ]
+  },
+  "_LEGACY_ERROR_TEMP_DELTA_0007" : {
+    "message" : [
+      "<message>"
+    ]
   }
 }
diff --git a/spark/src/main/scala/org/apache/spark/sql/delta/DeltaErrors.scala b/spark/src/main/scala/org/apache/spark/sql/delta/DeltaErrors.scala
@@ -1739,8 +1739,9 @@ trait DeltaErrorsBase
   }
 
   def generatedColumnsNonDeterministicExpression(expr: Expression): Throwable = {
-    new AnalysisException(
-      s"Found ${expr.sql}. A generated column cannot use a non deterministic expression")
+    new DeltaAnalysisException(
+      errorClass = "DELTA_NON_DETERMINISTIC_EXPRESSION_IN_GENERATED_COLUMN",
+      messageParameters = Array(s"${expr.sql}"))
   }
 
   def generatedColumnsAggregateExpression(expr: Expression): Throwable = {
@@ -2074,24 +2075,20 @@ trait DeltaErrorsBase
       operation: String,
       columnName: String,
       constraints: Map[String, String]): Throwable = {
-    val plural = if (constraints.size > 1) "s" else ""
-    new AnalysisException(
-      s"""
-        |Cannot $operation column $columnName because this column is referenced by the following
-        | check constraint$plural:\n\t${constraints.mkString("\n\t")}
-        |""".stripMargin)
+    new DeltaAnalysisException(
+      errorClass = "_LEGACY_ERROR_TEMP_DELTA_0004",
+      messageParameters = Array(operation, toSQLId(columnName), constraints.mkString("\n"))
+    )
   }
 
   def foundViolatingGeneratedColumnsForColumnChange(
       operation: String,
       columnName: String,
       fields: Seq[StructField]): Throwable = {
-    val plural = if (fields.size > 1) "s" else ""
-    new AnalysisException(
-      s"""
-         |Cannot $operation column $columnName because this column is referenced by the following
-         | generated column$plural:\n\t${fields.map(_.name).mkString("\n\t")}
-         |""".stripMargin)
+    new DeltaAnalysisException(
+      errorClass = "_LEGACY_ERROR_TEMP_DELTA_0005",
+      messageParameters = Array(operation, toSQLId(columnName), fields.map(_.name).mkString("\n"))
+    )
   }
 
   def missingColumnsInInsertInto(column: String): Throwable = {
@@ -2416,8 +2413,10 @@ trait DeltaErrorsBase
       hasStart: Boolean,
       hasStep: Boolean,
       hasInsert: Boolean): Throwable = {
-    new AnalysisException(s"Inconsistent IDENTITY metadata for column $colName " +
-      s"detected: $hasStart, $hasStep, $hasInsert")
+    new DeltaAnalysisException(
+      errorClass = "_LEGACY_ERROR_TEMP_DELTA_0006",
+      messageParameters = Array(toSQLId(colName), s"$hasStart", s"$hasStep", s"$hasInsert")
+    )
   }
 
   def activeSparkSessionNotFound(): Throwable = {
@@ -3267,6 +3266,13 @@ trait DeltaErrorsBase
       errorClass = "DELTA_CREATE_TABLE_SET_CLUSTERING_TABLE_FEATURE_NOT_ALLOWED",
       messageParameters = Array(tableFeature))
   }
+
+  def mergeAddVoidColumn(columnName: String): Throwable = {
+    new DeltaAnalysisException(
+      errorClass = "DELTA_MERGE_ADD_VOID_COLUMN",
+      messageParameters = Array(toSQLId(columnName))
+    )
+  }
 }
 
 object DeltaErrors extends DeltaErrorsBase
@@ -3436,7 +3442,10 @@ class MetadataMismatchErrorBuilder {
   }
 
   def finalizeAndThrow(conf: SQLConf): Unit = {
-    throw new AnalysisException(bits.mkString("\n"))
+    throw new DeltaAnalysisException(
+      errorClass = "_LEGACY_ERROR_TEMP_DELTA_0007",
+      messageParameters = Array(bits.mkString("\n"))
+    )
   }
 }
 

diff --git a/spark/src/main/scala/org/apache/spark/sql/delta/commands/MergeIntoCommandBase.scala b/spark/src/main/scala/org/apache/spark/sql/delta/commands/MergeIntoCommandBase.scala
@@ -160,10 +160,7 @@ trait MergeIntoCommandBase extends LeafRunnableCommand
       // queries that add void columns.
       val newNullColumn = SchemaUtils.findNullTypeColumn(migratedSchema.get)
       if (newNullColumn.isDefined) {
-        throw new AnalysisException(
-          s"""Cannot add column '${newNullColumn.get}' with type 'void'. Please explicitly specify a
-              |non-void type.""".stripMargin.replaceAll("\n", " ")
-        )
+        throw DeltaErrors.mergeAddVoidColumn(newNullColumn.get)
       }
     }
 

diff --git a/spark/src/test/scala/org/apache/spark/sql/delta/DeltaErrorsSuite.scala b/spark/src/test/scala/org/apache/spark/sql/delta/DeltaErrorsSuite.scala
@@ -2886,6 +2886,80 @@ trait DeltaErrorsSuiteBase
         Some("0AKDC"),
         Some("Operation not allowed: FOO_OP cannot be performed on a view."))
     }
+    {
+      val expr = "1".expr
+      val e = intercept[DeltaAnalysisException] {
+        throw DeltaErrors.generatedColumnsNonDeterministicExpression(expr)
+      }
+      checkErrorMessage(
+        e,
+        Some("DELTA_NON_DETERMINISTIC_EXPRESSION_IN_GENERATED_COLUMN"),
+        Some("42621"),
+        Some(s"Found ${expr.sql}. A generated column cannot use a non deterministic expression.")
+      )
+    }
+    {
+      val e = intercept[DeltaAnalysisException] {
+        throw DeltaErrors.foundViolatingConstraintsForColumnChange(
+          "UPDATE", "col1", Map("foo" -> "bar"))
+      }
+      checkErrorMessage(
+        e,
+        Some("_LEGACY_ERROR_TEMP_DELTA_0004"),
+        None,
+        Some(
+          s"""Cannot UPDATE column `col1` because this column is referenced by the following
+             |check constraint(s):
+             |foo -> bar""".stripMargin)
+      )
+    }
+    {
+      val e = intercept[DeltaAnalysisException] {
+        throw DeltaErrors.foundViolatingGeneratedColumnsForColumnChange(
+          "UPDATE", "col1", Seq(StructField("col2", IntegerType)))
+      }
+      checkErrorMessage(
+        e,
+        Some("_LEGACY_ERROR_TEMP_DELTA_0005"),
+        None,
+        Some(
+          s"""Cannot UPDATE column `col1` because this column is referenced by the following
+             |generated column(s):
+             |col2""".stripMargin)
+      )
+    }
+    {
+      val e = intercept[DeltaAnalysisException] {
+        throw DeltaErrors.identityColumnInconsistentMetadata("col1", true, true, true)
+      }
+      checkErrorMessage(
+        e,
+        Some("_LEGACY_ERROR_TEMP_DELTA_0006"),
+        None,
+        Some(s"Inconsistent IDENTITY metadata for column `col1` detected: true, true, true")
+      )
+    }
+    {
+      val errorBuilder = new MetadataMismatchErrorBuilder()
+      val schema1 = StructType(Seq(StructField("c0", IntegerType)))
+      val schema2 = StructType(Seq(StructField("c0", StringType)))
+      errorBuilder.addSchemaMismatch(schema1, schema2, "id")
+      val e = intercept[DeltaAnalysisException] {
+        errorBuilder.finalizeAndThrow(spark.sessionState.conf)
+      }
+      assert(e.getErrorClass == "_LEGACY_ERROR_TEMP_DELTA_0007")
+    }
+    {
+      val e = intercept[DeltaAnalysisException] {
+        throw DeltaErrors.mergeAddVoidColumn("fooCol")
+      }
+      checkErrorMessage(
+        e,
+        Some("DELTA_MERGE_ADD_VOID_COLUMN"),
+        Some("42K09"),
+        Some(s"Cannot add column `fooCol` with type VOID. Please explicitly specify a non-void type.")
+      )
+    }
   }
 }