[SQL] Clarify schema mismatch types in insertInto error

Author: a1noh

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",