Change all SQL type to upper case.

Author: wangyum

docs/sql-programming-guide.md

@@ -1339,7 +1339,7 @@ the following case-insensitive options:
   <tr>
     <td><code>customDataFrameColumnTypes</code></td>
     <td>
-     The DataFrame column data types to use instead of the defaults when reading data from jdbc API. (e.g: <code>"id decimal(38, 0), name string")</code>. The specified types should be valid spark sql data types. This option applies only to reading.
+     The DataFrame column data types to use instead of the defaults when reading data from jdbc API. (e.g: <code>"id DECIMAL(38, 0), name STRING")</code>. The specified types should be valid spark sql data types. This option applies only to reading.
     </td>
   </tr>
 </table>

external/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/OracleIntegrationSuite.scala

@@ -80,7 +80,7 @@ class OracleIntegrationSuite extends DockerJDBCIntegrationSuite with SharedSQLCo
     conn.commit()
 
     conn.prepareStatement(
-      "CREATE TABLE tableWithCustomSchema (id NUMBER, n1 number(1), n2 number(1))").executeUpdate()
+      "CREATE TABLE tableWithCustomSchema (id NUMBER, n1 NUMBER(1), n2 NUMBER(1))").executeUpdate()
     conn.prepareStatement(
       "INSERT INTO tableWithCustomSchema values(12312321321321312312312312123, 1, 0)").executeUpdate()
     conn.commit()
@@ -291,7 +291,7 @@ class OracleIntegrationSuite extends DockerJDBCIntegrationSuite with SharedSQLCo
     // custom schema can read data
     val props = new Properties()
     props.put("customDataFrameColumnTypes",
-      s"ID decimal(${DecimalType.MAX_PRECISION}, 0), N1 int, N2 boolean")
+      s"ID DECIMAL(${DecimalType.MAX_PRECISION}, 0), N1 INT, N2 BOOLEAN")
     val dfRead = spark.read.jdbc(jdbcUrl, "tableWithCustomSchema", props)
 
     val rows = dfRead.collect()

sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/jdbc/JdbcUtilsSuite.scala

@@ -31,55 +31,55 @@ class JdbcUtilsSuite extends SparkFunSuite {
 
   test("Parse user specified column types") {
     assert(
-      JdbcUtils.parseUserSpecifiedColumnTypes(schema, "C1 date, C2 string", caseInsensitive) ===
+      JdbcUtils.parseUserSpecifiedColumnTypes(schema, "C1 DATE, C2 STRING", caseInsensitive) ===
       StructType(Seq(StructField("C1", DateType, true), StructField("C2", StringType, true))))
-    assert(JdbcUtils.parseUserSpecifiedColumnTypes(schema, "C1 date, C2 string", caseSensitive) ===
+    assert(JdbcUtils.parseUserSpecifiedColumnTypes(schema, "C1 DATE, C2 STRING", caseSensitive) ===
       StructType(Seq(StructField("C1", DateType, true), StructField("C2", StringType, true))))
     assert(
-      JdbcUtils.parseUserSpecifiedColumnTypes(schema, "c1 date, C2 string", caseInsensitive) ===
+      JdbcUtils.parseUserSpecifiedColumnTypes(schema, "c1 DATE, C2 STRING", caseInsensitive) ===
         StructType(Seq(StructField("c1", DateType, true), StructField("C2", StringType, true))))
     assert(JdbcUtils.parseUserSpecifiedColumnTypes(
-      schema, "c1 decimal(38, 0), C2 string", caseInsensitive) ===
+      schema, "c1 DECIMAL(38, 0), C2 STRING", caseInsensitive) ===
       StructType(Seq(StructField("c1", DecimalType(38, 0), true),
         StructField("C2", StringType, true))))
 
     // Throw AnalysisException
     val duplicate = intercept[AnalysisException]{
-      JdbcUtils.parseUserSpecifiedColumnTypes(schema, "c1 date, c1 string", caseInsensitive) ===
+      JdbcUtils.parseUserSpecifiedColumnTypes(schema, "c1 DATE, c1 STRING", caseInsensitive) ===
         StructType(Seq(StructField("c1", DateType, true), StructField("c1", StringType, true)))
     }
     assert(duplicate.getMessage.contains(
       "Found duplicate column(s) in the createTableColumnTypes option value"))
 
     val allColumns = intercept[AnalysisException]{
-      JdbcUtils.parseUserSpecifiedColumnTypes(schema, "C1 string", caseSensitive) ===
+      JdbcUtils.parseUserSpecifiedColumnTypes(schema, "C1 STRING", caseSensitive) ===
         StructType(Seq(StructField("C1", DateType, true)))
     }
     assert(allColumns.getMessage.contains("Please provide all the columns,"))
 
     val caseSensitiveColumnNotFound = intercept[AnalysisException]{
-      JdbcUtils.parseUserSpecifiedColumnTypes(schema, "c1 date, C2 string", caseSensitive) ===
+      JdbcUtils.parseUserSpecifiedColumnTypes(schema, "c1 DATE, C2 STRING", caseSensitive) ===
         StructType(Seq(StructField("c1", DateType, true), StructField("C2", StringType, true)))
     }
     assert(caseSensitiveColumnNotFound.getMessage.contains(
       s"${JDBCOptions.JDBC_CUSTOM_DATAFRAME_COLUMN_TYPES} option column c1 not found in schema"))
 
     val caseInsensitiveColumnNotFound = intercept[AnalysisException]{
-      JdbcUtils.parseUserSpecifiedColumnTypes(schema, "c3 date, C2 string", caseInsensitive) ===
+      JdbcUtils.parseUserSpecifiedColumnTypes(schema, "c3 DATE, C2 STRING", caseInsensitive) ===
         StructType(Seq(StructField("c3", DateType, true), StructField("C2", StringType, true)))
     }
     assert(caseInsensitiveColumnNotFound.getMessage.contains(
       s"${JDBCOptions.JDBC_CUSTOM_DATAFRAME_COLUMN_TYPES} option column c3 not found in schema"))
 
     // Throw ParseException
-    val DataTypeNotSupported = intercept[ParseException]{
-      JdbcUtils.parseUserSpecifiedColumnTypes(schema, "c3 datee, C2 string", caseInsensitive) ===
+    val dataTypeNotSupported = intercept[ParseException]{
+      JdbcUtils.parseUserSpecifiedColumnTypes(schema, "c3 DATEE, C2 STRING", caseInsensitive) ===
         StructType(Seq(StructField("c3", DateType, true), StructField("C2", StringType, true)))
     }
-    assert(DataTypeNotSupported.getMessage.contains("DataType datee is not supported"))
+    assert(dataTypeNotSupported.getMessage.contains("DataType datee is not supported"))
 
     val mismatchedInput = intercept[ParseException]{
-      JdbcUtils.parseUserSpecifiedColumnTypes(schema, "c3 date. C2 string", caseInsensitive) ===
+      JdbcUtils.parseUserSpecifiedColumnTypes(schema, "c3 DATE. C2 STRING", caseInsensitive) ===
         StructType(Seq(StructField("c3", DateType, true), StructField("C2", StringType, true)))
     }
     assert(mismatchedInput.getMessage.contains("mismatched input '.' expecting"))

sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCSuite.scala

@@ -971,7 +971,7 @@ class JDBCSuite extends SparkFunSuite
   test("jdbc API support custom schema") {
     val parts = Array[String]("THEID < 2", "THEID >= 2")
     val props = new Properties()
-    props.put("customDataFrameColumnTypes", "NAME string, THEID bigint")
+    props.put("customDataFrameColumnTypes", "NAME STRING, THEID BIGINT")
     val schema = StructType(Seq(
       StructField("NAME", StringType, true), StructField("THEID", LongType, true)))
     val df = spark.read.jdbc(urlWithUserAndPass, "TEST.PEOPLE", parts, props)
@@ -987,7 +987,7 @@ class JDBCSuite extends SparkFunSuite
            |CREATE TEMPORARY VIEW people_view
            |USING org.apache.spark.sql.jdbc
            |OPTIONS (uRl '$url', DbTaBlE 'TEST.PEOPLE', User 'testUser', PassWord 'testPass',
-           |customDataFrameColumnTypes 'NAME string, THEID int')
+           |customDataFrameColumnTypes 'NAME STRING, THEID INT')
         """.stripMargin.replaceAll("\n", " "))
       val schema = StructType(
         Seq(StructField("NAME", StringType, true), StructField("THEID", IntegerType, true)))