[WIP][SQL] Clarify schema mismatch types in insertInto error

README.md

@@ -163,3 +163,4 @@ in the online documentation for an overview on how to configure Spark.
 
 Please review the [Contribution to Spark guide](https://spark.apache.org/contributing.html)
 for information on how to get started contributing to the project.
+# trigger build

common/utils/src/main/resources/error/error-conditions.json

@@ -2229,6 +2229,16 @@
     },
     "sqlState" : "21S01"
   },
+  "INSERT_INTO_SCHEMA_MISMATCH": {
+  "message": "InsertInto schema mismatch at column '%columnName%':\n- DataFrame column has type %sourceType%\n- Target table column '%targetColumnName%' has type %targetType%\n- Columns matched by position",
+  "sqlState": "42K09",
+  "parameters": [
+    "columnName",
+    "sourceType",
+    "targetColumnName",
+    "targetType"
+    ]
+  },
   "INSERT_PARTITION_COLUMN_ARITY_MISMATCH" : {
     "message" : [
       "Cannot write to '<tableName>', <reason>:",

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

@@ -570,7 +570,16 @@ object TableOutputResolver extends SQLConfHelper with Logging {
 
     val canWriteExpr = canWrite(
       tableName, queryExpr.dataType, attrTypeWithoutCharVarchar,
-      byName, conf, addError, colPath)
+      byName, conf, errorMsg => {
+        val colName = colPath.lastOption.getOrElse(tableAttr.name)
+        addError(QueryCompilationErrors.insertIntoSchemaMismatchError(
+        colName,
+        queryExpr.dataType.simpleString,
+        tableAttr.name,
+        attrTypeWithoutCharVarchar.simpleString
+        ))
+      }
+      , colPath)
 
     if (canWriteExpr) outputField else None
   }

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

@@ -576,6 +576,23 @@ private[sql] object QueryCompilationErrors extends QueryErrorsBase with Compilat
       origin = t.origin)
   }
 
+  def insertIntoSchemaMismatchError(
+    columnName: String,
+    sourceType: String,
+    targetColumnName: String,
+    targetType: String): AnalysisException = {
+    new AnalysisException(
+      errorClass = "INSERT_INTO_SCHEMA_MISMATCH",
+      messageParameters = Map(
+        "columnName" -> columnName,
+        "sourceType" -> sourceType,
+        "targetColumnName" -> targetColumnName,
+        "targetType" -> targetType
+        )
+    )
+  }
+
+
   def writeIntoViewNotAllowedError(identifier: TableIdentifier, t: TreeNode[_]): Throwable = {
     new AnalysisException(
       errorClass = "_LEGACY_ERROR_TEMP_1011",

trigger.txt

@@ -0,0 +1,2 @@
+# trigger re-run
+# trigger re-run