Adjusted comments

apache · Oct 16, 2024 · ab114c9 · ab114c9
1 parent eae01b3
commit ab114c9
Show file tree

Hide file tree

Showing 5 changed files with 20 additions and 15 deletions.
diff --git a/common/utils/src/main/resources/error/error-conditions.json b/common/utils/src/main/resources/error/error-conditions.json
@@ -2138,6 +2138,12 @@
     },
     "sqlState" : "22022"
   },
+  "INVALID_CORRUPT_RECORD_TYPE" : {
+    "message" : [
+      "The column <columnName> for corrupt records must have the nullable STRING type, but got <actualType>."
+    ],
+    "sqlState" : "42804"
+  },
   "INVALID_CURSOR" : {
     "message" : [
       "The cursor is invalid."
@@ -4852,11 +4858,6 @@
     ],
     "sqlState" : "0A000"
   },
-  "UNSUPPORTED_DATA_TYPE_FOR_CORRUPTED_RECORD" : {
-    "message" : [
-      "The field <columnName> for corrupt records must be string type and nullable."
-    ]
-  },
   "UNSUPPORTED_DATA_TYPE_FOR_ENCODER" : {
     "message" : [
       "Cannot create encoder for <dataType>. Please use a different output data type for your UDF or DataFrame."

diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/ExprUtils.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/ExprUtils.scala
@@ -84,7 +84,7 @@ object ExprUtils extends QueryErrorsBase {
       val f = schema(corruptFieldIndex)
       if (!f.dataType.isInstanceOf[StringType] || !f.nullable) {
         throw QueryCompilationErrors.invalidFieldTypeForCorruptRecordError(
-          columnNameOfCorruptRecord)
+          columnNameOfCorruptRecord, f.dataType)
       }
     }
   }

diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala
@@ -1327,10 +1327,11 @@ private[sql] object QueryCompilationErrors extends QueryErrorsBase with Compilat
       messageParameters = Map.empty)
   }
 
-  def invalidFieldTypeForCorruptRecordError(columnName: String): Throwable = {
+  def invalidFieldTypeForCorruptRecordError(columnName: String, actualType: DataType): Throwable = {
     new AnalysisException(
-      errorClass = "UNSUPPORTED_DATA_TYPE_FOR_CORRUPTED_RECORD",
-      messageParameters = Map("columnName" -> toSQLId(columnName)))
+      errorClass = "INVALID_CORRUPT_RECORD_TYPE",
+      messageParameters = Map(
+        "columnName" -> toSQLId(columnName), "actualType" -> toSQLType(actualType)))
   }
 
   def dataTypeUnsupportedByClassError(x: DataType, className: String): Throwable = {

diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/csv/CSVSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/csv/CSVSuite.scala
@@ -1510,8 +1510,9 @@ abstract class CSVSuite
             .csv(testFile(valueMalformedFile))
             .collect()
         },
-        condition = "UNSUPPORTED_DATA_TYPE_FOR_CORRUPTED_RECORD",
-        parameters = Map("columnName" -> toSQLId(columnNameOfCorruptRecord))
+        condition = "INVALID_CORRUPT_RECORD_TYPE",
+        parameters = Map(
+          "columnName" -> toSQLId(columnNameOfCorruptRecord), "actualType" -> "\"INT\"")
       )
     }
   }

diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/json/JsonSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/json/JsonSuite.scala
@@ -2117,8 +2117,9 @@ abstract class JsonSuite
           .schema(schema)
           .json(corruptRecords)
       },
-      condition = "UNSUPPORTED_DATA_TYPE_FOR_CORRUPTED_RECORD",
-      parameters = Map("columnName" -> toSQLId(columnNameOfCorruptRecord))
+      condition = "INVALID_CORRUPT_RECORD_TYPE",
+      parameters = Map(
+        "columnName" -> toSQLId(columnNameOfCorruptRecord), "actualType" -> "\"INT\"")
     )
 
     // We use `PERMISSIVE` mode by default if invalid string is given.
@@ -2134,8 +2135,9 @@ abstract class JsonSuite
             .json(path)
             .collect()
         },
-        condition = "UNSUPPORTED_DATA_TYPE_FOR_CORRUPTED_RECORD",
-        parameters = Map("columnName" -> toSQLId(columnNameOfCorruptRecord))
+        condition = "INVALID_CORRUPT_RECORD_TYPE",
+        parameters = Map(
+          "columnName" -> toSQLId(columnNameOfCorruptRecord), "actualType" -> "\"INT\"")
       )
     }
   }