[SPARK-43969][SQL] Refactor & Assign names to the error class _LEGACY_ERROR_TEMP_1170

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

@@ -877,10 +877,24 @@
   },
   "INSERT_COLUMN_ARITY_MISMATCH" : {
     "message" : [
-      "Cannot write to '<tableName>', <reason>:",
-      "Table columns: <tableColumns>.",
-      "Data columns: <dataColumns>."
+      "Cannot write to '<tableName>',"
     ],
+    "subClass" : {
+      "NOT_ENOUGH_DATA_COLUMNS" : {
+        "message" : [
+          "not enough data columns: ",
+          "Table columns: <tableColumns>.",
+          "Data columns: <dataColumns>."
+        ]
+      },
+      "TOO_MANY_DATA_COLUMNS" : {
+        "message" : [
+          "too many data columns: ",
+          "Table columns: <tableColumns>.",
+          "Data columns: <dataColumns>."
+        ]
+      }
+    },
     "sqlState" : "21S01"
   },
   "INSERT_PARTITION_COLUMN_ARITY_MISMATCH" : {
@@ -1651,6 +1665,11 @@
     ],
     "sqlState" : "46110"
   },
+  "NOT_SUPPORTED_COMMAND_WITHOUT_HIVE_SUPPORT" : {
+    "message" : [
+      "<cmd> is not supported, if you want to enable it, please set `spark.sql.catalogImplementation` to `hive`."
+    ]
+  },
   "NOT_SUPPORTED_IN_JDBC_CATALOG" : {
     "message" : [
       "Not supported command in JDBC catalog:"
@@ -2368,6 +2387,33 @@
       "grouping()/grouping_id() can only be used with GroupingSets/Cube/Rollup."
     ]
   },
+  "UNSUPPORTED_INSERT" : {
+    "message" : [
+      "Can't insert into the target."
+    ],
+    "subClass" : {
+      "NOT_ALLOWED" : {
+        "message" : [
+          "The target relation <relationId> does not allow insertion."
+        ]
+      },
+      "NOT_PARTITIONED" : {
+        "message" : [
+          "The target relation <relationId> is not partitioned."
+        ]
+      },
+      "RDD_BASED" : {
+        "message" : [
+          "An RDD-based table is not allowed."
+        ]
+      },
+      "READ_FROM" : {
+        "message" : [
+          "The target relation <relationId> is also being read from."
+        ]
+      }
+    }
+  },
   "UNSUPPORTED_OVERWRITE" : {
     "message" : [
       "Can't overwrite the target that is also being read from."
@@ -2929,11 +2975,6 @@
       "Window function <wf> requires window to be ordered, please add ORDER BY clause. For example SELECT <wf>(value_expr) OVER (PARTITION BY window_partition ORDER BY window_ordering) from table."
     ]
   },
-  "_LEGACY_ERROR_TEMP_1038" : {
-    "message" : [
-      "Cannot write to table due to mismatched user specified column size(<columnSize>) and data column size(<outputSize>)."
-    ]
-  },
   "_LEGACY_ERROR_TEMP_1039" : {
     "message" : [
       "Multiple time/session window expressions would result in a cartesian product of rows, therefore they are currently not supported."
@@ -3430,11 +3471,6 @@
       "Table partitions: <partColNames>."
     ]
   },
-  "_LEGACY_ERROR_TEMP_1170" : {
-    "message" : [
-      "Hive support is required to <detail>."
-    ]
-  },
   "_LEGACY_ERROR_TEMP_1171" : {
     "message" : [
       "createTableColumnTypes option column <col> not found in schema <schema>."

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala

@@ -1288,7 +1288,7 @@ class Analyzer(override val catalogManager: CatalogManager) extends RuleExecutor
 
         // Create a project if this is an INSERT INTO BY NAME query.
         val projectByName = if (i.userSpecifiedCols.nonEmpty) {
-          Some(createProjectForByNameQuery(i))
+          Some(createProjectForByNameQuery(r.table.name, i))
         } else {
           None
         }

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveInsertionBase.scala

@@ -28,12 +28,19 @@ abstract class ResolveInsertionBase extends Rule[LogicalPlan] {
   def resolver: Resolver = conf.resolver
 
   /** Add a project to use the table column names for INSERT INTO BY NAME */
-  protected def createProjectForByNameQuery(i: InsertIntoStatement): LogicalPlan = {
+  protected def createProjectForByNameQuery(
+      tblName: String,
+      i: InsertIntoStatement): LogicalPlan = {
     SchemaUtils.checkColumnNameDuplication(i.userSpecifiedCols, resolver)
 
     if (i.userSpecifiedCols.size != i.query.output.size) {
-      throw QueryCompilationErrors.writeTableWithMismatchedColumnsError(
-        i.userSpecifiedCols.size, i.query.output.size, i.query)
+      if (i.userSpecifiedCols.size > i.query.output.size) {
+        throw QueryCompilationErrors.cannotWriteNotEnoughColumnsToTableError(
+          tblName, i.userSpecifiedCols, i.query)
+      } else {
+        throw QueryCompilationErrors.cannotWriteTooManyColumnsToTableError(
+          tblName, i.userSpecifiedCols, i.query)
+      }
     }
     val projectByName = i.userSpecifiedCols.zip(i.query.output)
       .map { case (userSpecifiedCol, queryOutputCol) =>

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/TableOutputResolver.scala

@@ -47,7 +47,7 @@ object TableOutputResolver {
 
     if (actualExpectedCols.size < query.output.size) {
       throw QueryCompilationErrors.cannotWriteTooManyColumnsToTableError(
-        tableName, actualExpectedCols, query)
+        tableName, actualExpectedCols.map(_.name), query)
     }
 
     val errors = new mutable.ArrayBuffer[String]()
@@ -74,7 +74,7 @@ object TableOutputResolver {
       }
       if (actualExpectedCols.size > queryOutputCols.size) {
         throw QueryCompilationErrors.cannotWriteNotEnoughColumnsToTableError(
-          tableName, actualExpectedCols, query)
+          tableName, actualExpectedCols.map(_.name), query)
       }
 
       resolveColumnsByPosition(queryOutputCols, actualExpectedCols, conf, errors += _)

sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala

@@ -613,16 +613,6 @@ private[sql] object QueryCompilationErrors extends QueryErrorsBase {
       messageParameters = Map("wf" -> wf.toString))
   }
 
-  def writeTableWithMismatchedColumnsError(
-      columnSize: Int, outputSize: Int, t: TreeNode[_]): Throwable = {
-    new AnalysisException(
-      errorClass = "_LEGACY_ERROR_TEMP_1038",
-      messageParameters = Map(
-        "columnSize" -> columnSize.toString,
-        "outputSize" -> outputSize.toString),
-      origin = t.origin)
-  }
-
   def multiTimeWindowExpressionsNotSupportedError(t: TreeNode[_]): Throwable = {
     new AnalysisException(
       errorClass = "_LEGACY_ERROR_TEMP_1039",
@@ -1743,10 +1733,10 @@ private[sql] object QueryCompilationErrors extends QueryErrorsBase {
         "partColNames" -> partColNames.map(_.name).mkString(",")))
   }
 
-  def ddlWithoutHiveSupportEnabledError(detail: String): Throwable = {
+  def ddlWithoutHiveSupportEnabledError(cmd: String): Throwable = {
     new AnalysisException(
-      errorClass = "_LEGACY_ERROR_TEMP_1170",
-      messageParameters = Map("detail" -> detail))
+      errorClass = "NOT_SUPPORTED_COMMAND_WITHOUT_HIVE_SUPPORT",
+      messageParameters = Map("cmd" -> cmd))
   }
 
   def createTableColumnTypesOptionColumnNotFoundInSchemaError(
@@ -2056,26 +2046,26 @@ private[sql] object QueryCompilationErrors extends QueryErrorsBase {
 
   def cannotWriteTooManyColumnsToTableError(
       tableName: String,
-      expected: Seq[Attribute],
+      expected: Seq[String],
       query: LogicalPlan): Throwable = {
     new AnalysisException(
-      errorClass = "INSERT_COLUMN_ARITY_MISMATCH",
+      errorClass = "INSERT_COLUMN_ARITY_MISMATCH.TOO_MANY_DATA_COLUMNS",
       messageParameters = Map(
         "tableName" -> tableName,
-        "reason" -> "too many data columns",
-        "tableColumns" -> expected.map(c => s"'${c.name}'").mkString(", "),
-        "dataColumns" -> query.output.map(c => s"'${c.name}'").mkString(", ")))
+        "tableColumns" -> expected.map(c => toSQLId(c)).mkString(", "),
+        "dataColumns" -> query.output.map(c => toSQLId(c.name)).mkString(", ")))
   }
 
   def cannotWriteNotEnoughColumnsToTableError(
-      tableName: String, expected: Seq[Attribute], query: LogicalPlan): Throwable = {
+      tableName: String,
+      expected: Seq[String],
+      query: LogicalPlan): Throwable = {
     new AnalysisException(
-      errorClass = "INSERT_COLUMN_ARITY_MISMATCH",
+      errorClass = "INSERT_COLUMN_ARITY_MISMATCH.NOT_ENOUGH_DATA_COLUMNS",
       messageParameters = Map(
         "tableName" -> tableName,
-        "reason" -> "not enough data columns",
-        "tableColumns" -> expected.map(c => s"'${c.name}'").mkString(", "),
-        "dataColumns" -> query.output.map(c => s"'${c.name}'").mkString(", ")))
+        "tableColumns" -> expected.map(c => toSQLId(c)).mkString(", "),
+        "dataColumns" -> query.output.map(c => toSQLId(c.name)).mkString(", ")))
   }
 
   def cannotWriteIncompatibleDataToTableError(tableName: String, errors: Seq[String]): Throwable = {

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/rules.scala

@@ -391,7 +391,7 @@ object PreprocessTableInsertion extends ResolveInsertionBase {
     // Create a project if this INSERT has a user-specified column list.
     val isByName = insert.userSpecifiedCols.nonEmpty
     val query = if (isByName) {
-      createProjectForByNameQuery(insert)
+      createProjectForByNameQuery(tblName, insert)
     } else {
       insert.query
     }
@@ -400,7 +400,8 @@ object PreprocessTableInsertion extends ResolveInsertionBase {
         tblName, expectedColumns, query, byName = isByName, conf, supportColDefaultValue = true)
     } catch {
       case e: AnalysisException if staticPartCols.nonEmpty &&
-          e.getErrorClass == "INSERT_COLUMN_ARITY_MISMATCH" =>
+        (e.getErrorClass == "INSERT_COLUMN_ARITY_MISMATCH.NOT_ENOUGH_DATA_COLUMNS" ||
+          e.getErrorClass == "INSERT_COLUMN_ARITY_MISMATCH.TOO_MANY_DATA_COLUMNS") =>
         val newException = e.copy(
           errorClass = Some("INSERT_PARTITION_COLUMN_ARITY_MISMATCH"),
           messageParameters = e.messageParameters ++ Map(
@@ -502,8 +503,6 @@ object PreReadCheck extends (LogicalPlan => Unit) {
  */
 object PreWriteCheck extends (LogicalPlan => Unit) {
 
-  def failAnalysis(msg: String): Unit = { throw new AnalysisException(msg) }
-
   def apply(plan: LogicalPlan): Unit = {
     plan.foreach {
       case InsertIntoStatement(l @ LogicalRelation(relation, _, _, _), partition, _, query, _, _) =>
@@ -512,7 +511,9 @@ object PreWriteCheck extends (LogicalPlan => Unit) {
           case LogicalRelation(src, _, _, _) => src
         }
         if (srcRelations.contains(relation)) {
-          failAnalysis("Cannot insert into table that is also being read from.")
+          throw new AnalysisException(
+            errorClass = "UNSUPPORTED_INSERT.READ_FROM",
+            messageParameters = Map("relationId" -> toSQLId(relation.toString)))
         } else {
           // OK
         }
@@ -522,18 +523,25 @@ object PreWriteCheck extends (LogicalPlan => Unit) {
 
           // Right now, we do not support insert into a non-file-based data source table with
           // partition specs.
-          case _: InsertableRelation if partition.nonEmpty =>
-            failAnalysis(s"Insert into a partition is not allowed because $l is not partitioned.")
-
-          case _ => failAnalysis(s"$relation does not allow insertion.")
+          case i: InsertableRelation if partition.nonEmpty =>
+            throw new AnalysisException(
+              errorClass = "UNSUPPORTED_INSERT.NOT_PARTITIONED",
+              messageParameters = Map("relationId" -> toSQLId(i.toString)))
+
+          case _ =>
+            throw new AnalysisException(
+              errorClass = "UNSUPPORTED_INSERT.NOT_ALLOWED",
+              messageParameters = Map("relationId" -> toSQLId(relation.toString)))
         }
 
       case InsertIntoStatement(t, _, _, _, _, _)
         if !t.isInstanceOf[LeafNode] ||
           t.isInstanceOf[Range] ||
           t.isInstanceOf[OneRowRelation] ||
           t.isInstanceOf[LocalRelation] =>
-        failAnalysis(s"Inserting into an RDD-based table is not allowed.")
+        throw new AnalysisException(
+          errorClass = "UNSUPPORTED_INSERT.RDD_BASED",
+          messageParameters = Map.empty)
 
       case _ => // OK
     }

sql/core/src/test/resources/sql-tests/analyzer-results/postgreSQL/numeric.sql.out

@@ -3841,12 +3841,11 @@ INSERT INTO num_result SELECT t1.id, t2.id, t1.val, t2.val, t1.val * t2.val
 -- !query analysis
 org.apache.spark.sql.AnalysisException
 {
-  "errorClass" : "INSERT_COLUMN_ARITY_MISMATCH",
+  "errorClass" : "INSERT_COLUMN_ARITY_MISMATCH.TOO_MANY_DATA_COLUMNS",
   "sqlState" : "21S01",
   "messageParameters" : {
-    "dataColumns" : "'id', 'id', 'val', 'val', '(val * val)'",
-    "reason" : "too many data columns",
-    "tableColumns" : "'id1', 'id2', 'result'",
+    "dataColumns" : "`id`, `id`, `val`, `val`, `(val * val)`",
+    "tableColumns" : "`id1`, `id2`, `result`",
     "tableName" : "`spark_catalog`.`default`.`num_result`"
   }
 }

sql/core/src/test/resources/sql-tests/results/postgreSQL/numeric.sql.out

@@ -3832,12 +3832,11 @@ struct<>
 -- !query output
 org.apache.spark.sql.AnalysisException
 {
-  "errorClass" : "INSERT_COLUMN_ARITY_MISMATCH",
+  "errorClass" : "INSERT_COLUMN_ARITY_MISMATCH.TOO_MANY_DATA_COLUMNS",
   "sqlState" : "21S01",
   "messageParameters" : {
-    "dataColumns" : "'id', 'id', 'val', 'val', '(val * val)'",
-    "reason" : "too many data columns",
-    "tableColumns" : "'id1', 'id2', 'result'",
+    "dataColumns" : "`id`, `id`, `val`, `val`, `(val * val)`",
+    "tableColumns" : "`id1`, `id2`, `result`",
     "tableName" : "`spark_catalog`.`default`.`num_result`"
   }
 }