[SPARK-28023][SQL] Add trim logic in UTF8String's toInt/toLong to make it consistent with other string-numeric casting

common/unsafe/src/main/java/org/apache/spark/unsafe/types/UTF8String.java

@@ -1077,26 +1077,31 @@ public static class IntWrapper implements Serializable {
    * @return true if the parsing was successful else false
    */
   public boolean toLong(LongWrapper toLongResult) {
-    if (numBytes == 0) {
-      return false;
-    }
+    if (this.numBytes == 0) return false;
+    int offset = 0;
+    while (offset < this.numBytes && getByte(offset) == ' ') offset++;
+    if (offset == this.numBytes) return false;
 
-    byte b = getByte(0);
+    int end = this.numBytes - 1;
+    while (end > offset && getByte(end) == ' ') end--;
+
+    int numBytes = end - offset + 1;
+
+    byte b = getByte(offset);
     final boolean negative = b == '-';
-    int offset = 0;
     if (negative || b == '+') {
-      offset++;
       if (numBytes == 1) {
         return false;
       }
+      offset++;
     }
 
     final byte separator = '.';
     final int radix = 10;
     final long stopValue = Long.MIN_VALUE / radix;
     long result = 0;
 
-    while (offset < numBytes) {
+    while (offset <= end) {
       b = getByte(offset);
       offset++;
       if (b == separator) {
@@ -1131,7 +1136,7 @@ public boolean toLong(LongWrapper toLongResult) {
     // This is the case when we've encountered a decimal separator. The fractional
     // part will not change the number, but we will verify that the fractional part
     // is well formed.
-    while (offset < numBytes) {
+    while (offset <= end) {
       byte currentByte = getByte(offset);
       if (currentByte < '0' || currentByte > '9') {
         return false;
@@ -1168,26 +1173,31 @@ public boolean toLong(LongWrapper toLongResult) {
    * @return true if the parsing was successful else false
    */
   public boolean toInt(IntWrapper intWrapper) {
-    if (numBytes == 0) {
-      return false;
-    }
+    if (this.numBytes == 0) return false;
+    int offset = 0;
+    while (offset < this.numBytes && getByte(offset) == ' ') offset++;
+    if (offset == this.numBytes) return false;
 
-    byte b = getByte(0);
+    int end = this.numBytes - 1;
+    while (end > offset && getByte(end) == ' ') end--;
+
+    int numBytes = end - offset + 1;
+
+    byte b = getByte(offset);
     final boolean negative = b == '-';
-    int offset = 0;
     if (negative || b == '+') {
-      offset++;
       if (numBytes == 1) {
         return false;
       }
+      offset++;
     }
 
     final byte separator = '.';
     final int radix = 10;
     final int stopValue = Integer.MIN_VALUE / radix;
     int result = 0;
 
-    while (offset < numBytes) {
+    while (offset <= end) {
       b = getByte(offset);
       offset++;
       if (b == separator) {
@@ -1222,7 +1232,7 @@ public boolean toInt(IntWrapper intWrapper) {
     // This is the case when we've encountered a decimal separator. The fractional
     // part will not change the number, but we will verify that the fractional part
     // is well formed.
-    while (offset < numBytes) {
+    while (offset <= end) {
       byte currentByte = getByte(offset);
       if (currentByte < '0' || currentByte > '9') {
         return false;

sql/core/src/test/resources/sql-tests/inputs/cast.sql

@@ -60,3 +60,12 @@ DESC FUNCTION EXTENDED boolean;
 -- cast string to interval and interval to string
 SELECT CAST('interval 3 month 1 hour' AS interval);
 SELECT CAST(interval 3 month 1 hour AS string);
+
+-- trim string before cast to numeric
+select cast(' 1' as tinyint);
+select cast(' 1' as smallint);
+select cast(' 1' as INT);
+select cast(' 1' as bigint);
+select cast(' 1' as float);
+select cast(' 1 ' as DOUBLE);
+select cast('1.0 ' as DEC);

sql/core/src/test/resources/sql-tests/inputs/comparator.sql

@@ -1,3 +1,12 @@
 -- binary type
 select x'00' < x'0f';
 select x'00' < x'ff';
+
+-- trim string to numeric
+select '1 ' = 1Y;
+select '1 ' = 1S;
+select '1 ' = 1;
+select ' 1' = 1L;
+select ' 1' = cast(1.0 as float);
+select ' 1.0 ' = 1.0D;
+select ' 1.0 ' = 1.0BD;

sql/core/src/test/resources/sql-tests/results/cast.sql.out

@@ -1,5 +1,5 @@
 -- Automatically generated by SQLQueryTestSuite
--- Number of queries: 35
+-- Number of queries: 42
 
 
 -- !query 0
@@ -287,3 +287,59 @@ SELECT CAST(interval 3 month 1 hour AS string)
 struct<CAST(INTERVAL '3 months 1 hours' AS STRING):string>
 -- !query 34 output
 3 months 1 hours
+
+
+-- !query 35
+select cast(' 1' as tinyint)
+-- !query 35 schema
+struct<CAST( 1 AS TINYINT):tinyint>
+-- !query 35 output
+1
+
+
+-- !query 36
+select cast(' 1' as smallint)
+-- !query 36 schema
+struct<CAST( 1 AS SMALLINT):smallint>
+-- !query 36 output
+1
+
+
+-- !query 37
+select cast(' 1' as INT)
+-- !query 37 schema
+struct<CAST( 1 AS INT):int>
+-- !query 37 output
+1
+
+
+-- !query 38
+select cast(' 1' as bigint)
+-- !query 38 schema
+struct<CAST( 1 AS BIGINT):bigint>
+-- !query 38 output
+1
+
+
+-- !query 39
+select cast(' 1' as float)
+-- !query 39 schema
+struct<CAST( 1 AS FLOAT):float>
+-- !query 39 output
+1.0
+
+
+-- !query 40
+select cast(' 1 ' as DOUBLE)
+-- !query 40 schema
+struct<CAST( 1  AS DOUBLE):double>
+-- !query 40 output
+1.0
+
+
+-- !query 41
+select cast('1.0 ' as DEC)
+-- !query 41 schema
+struct<CAST(1.0  AS DECIMAL(10,0)):decimal(10,0)>
+-- !query 41 output
+NULL

sql/core/src/test/resources/sql-tests/results/comparator.sql.out

@@ -1,5 +1,5 @@
 -- Automatically generated by SQLQueryTestSuite
--- Number of queries: 2
+-- Number of queries: 9
 
 
 -- !query 0
@@ -16,3 +16,59 @@ select x'00' < x'ff'
 struct<(X'00' < X'FF'):boolean>
 -- !query 1 output
 true
+
+
+-- !query 2
+select '1 ' = 1Y
+-- !query 2 schema
+struct<(CAST(1  AS TINYINT) = 1):boolean>
+-- !query 2 output
+true
+
+
+-- !query 3
+select '1 ' = 1S
+-- !query 3 schema
+struct<(CAST(1  AS SMALLINT) = 1):boolean>
+-- !query 3 output
+true
+
+
+-- !query 4
+select '1 ' = 1
+-- !query 4 schema
+struct<(CAST(1  AS INT) = 1):boolean>
+-- !query 4 output
+true
+
+
+-- !query 5
+select ' 1' = 1L
+-- !query 5 schema
+struct<(CAST( 1 AS BIGINT) = 1):boolean>
+-- !query 5 output
+true
+
+
+-- !query 6
+select ' 1' = cast(1.0 as float)
+-- !query 6 schema
+struct<(CAST( 1 AS FLOAT) = CAST(1.0 AS FLOAT)):boolean>
+-- !query 6 output
+true
+
+
+-- !query 7
+select ' 1.0 ' = 1.0D
+-- !query 7 schema
+struct<(CAST( 1.0  AS DOUBLE) = 1.0):boolean>
+-- !query 7 output
+true
+
+
+-- !query 8
+select ' 1.0 ' = 1.0BD
+-- !query 8 schema
+struct<(CAST( 1.0  AS DOUBLE) = CAST(1.0 AS DOUBLE)):boolean>
+-- !query 8 output
+true