[SPARK-28151][SQL] Fix MsSqlServerDialect Byte/Short/Float type mappings ( DRAFT)

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

@@ -439,6 +439,10 @@ object JdbcUtils extends Logging {
       (rs: ResultSet, row: InternalRow, pos: Int) =>
         row.setLong(pos, rs.getLong(pos + 1))
 
+    case ByteType =>
+      (rs: ResultSet, row: InternalRow, pos: Int) =>
+        row.setByte(pos, rs.getByte(pos + 1))
+
     case ShortType =>
       (rs: ResultSet, row: InternalRow, pos: Int) =>
         row.setShort(pos, rs.getShort(pos + 1))
@@ -797,6 +801,7 @@ object JdbcUtils extends Logging {
       tableSchema: StructType,
       customSchema: String,
       nameEquality: Resolver): StructType = {
+
     if (null != customSchema && customSchema.nonEmpty) {
       val userSchema = CatalystSqlParser.parseTableSchema(customSchema)
 

sql/core/src/main/scala/org/apache/spark/sql/jdbc/MsSqlServerDialect.scala

@@ -30,7 +30,13 @@ private object MsSqlServerDialect extends JdbcDialect {
       // String is recommend by Microsoft SQL Server for datetimeoffset types in non-MS clients
       Option(StringType)
     } else {
-      None
+      val answer = sqlType match {
+        case java.sql.Types.TINYINT  => Option(ByteType)
+        case java.sql.Types.SMALLINT => Option(ShortType)
+        case java.sql.Types.REAL     => Option(FloatType)
+        case _ => None
+      }
+      answer
     }
   }
 
@@ -39,6 +45,8 @@ private object MsSqlServerDialect extends JdbcDialect {
     case StringType => Some(JdbcType("NVARCHAR(MAX)", java.sql.Types.NVARCHAR))
     case BooleanType => Some(JdbcType("BIT", java.sql.Types.BIT))
     case BinaryType => Some(JdbcType("VARBINARY(MAX)", java.sql.Types.VARBINARY))
+    case ByteType => Option(JdbcType("TINYINT", java.sql.Types.TINYINT))
+    case ShortType => Option(JdbcType("SMALLINT", java.sql.Types.SMALLINT))
     case _ => None
   }
 

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

@@ -895,6 +895,24 @@ class JDBCSuite extends QueryTest
       "BIT")
     assert(msSqlServerDialect.getJDBCType(BinaryType).map(_.databaseTypeDefinition).get ==
       "VARBINARY(MAX)")
+    assert(msSqlServerDialect.getJDBCType(ByteType).map(_.databaseTypeDefinition).get ==
+      "TINYINT")
+
+    assert(msSqlServerDialect.getJDBCType(ShortType).map(_.databaseTypeDefinition).get ==
+      "SMALLINT")
+  }
+
+  test("MsSqlServerDialect catalyst type mapping") {
+    val msSqlServerDialect = JdbcDialects.get("jdbc:sqlserver")
+    val metadata = new MetadataBuilder().putLong("scale", 1)
+
+    assert(msSqlServerDialect.getCatalystType(java.sql.Types.TINYINT,"TINYINT",1,metadata).get ==
+          ByteType)
+    assert(msSqlServerDialect.getCatalystType(java.sql.Types.SMALLINT,"SMALLINT",1,metadata).get ==
+      ShortType)
+    assert(msSqlServerDialect.getCatalystType(java.sql.Types.REAL,"REAL",1,metadata).get ==
+      FloatType)
+
   }
 
   test("table exists query by jdbc dialect") {