[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

@@ -553,6 +553,29 @@ public UTF8String trim() {
     return copyUTF8String(s, e);
   }
 
+  /**
+   * Trims space characters (ASCII 32) from both ends of this string.
+   *
+   * @return this string with no spaces at the start or end
+   */
+  public UTF8String nonCopyTrim() {
+    int s = 0;
+    // skip all of the space (0x20) in the left side
+    while (s < this.numBytes && getByte(s) == 0x20) s++;
+    if (s == this.numBytes) {
+      // Everything trimmed
+      return EMPTY_UTF8;
+    }
+    // skip all of the space (0x20) in the right side
+    int e = this.numBytes - 1;
+    while (e > s && getByte(e) == 0x20) e--;
+    if (s == 0 && e == numBytes - 1) {
+      // Nothing trimmed
+      return this;
+    }
+    return new UTF8String(base, offset + s, e - s + 1);
+  }
+
   /**
    * Trims instances of the given trim string from both ends of this string.
    *

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/Cast.scala

@@ -484,7 +484,7 @@ abstract class CastBase extends UnaryExpression with TimeZoneAwareExpression wit
   private[this] def castToLong(from: DataType): Any => Any = from match {
     case StringType =>
       val result = new LongWrapper()
-      buildCast[UTF8String](_, s => if (s.toLong(result)) result.value else null)
+      buildCast[UTF8String](_, s => if (s.nonCopyTrim().toLong(result)) result.value else null)
     case BooleanType =>
       buildCast[Boolean](_, b => if (b) 1L else 0L)
     case DateType =>
@@ -501,7 +501,7 @@ abstract class CastBase extends UnaryExpression with TimeZoneAwareExpression wit
   private[this] def castToInt(from: DataType): Any => Any = from match {
     case StringType =>
       val result = new IntWrapper()
-      buildCast[UTF8String](_, s => if (s.toInt(result)) result.value else null)
+      buildCast[UTF8String](_, s => if (s.nonCopyTrim().toInt(result)) result.value else null)
     case BooleanType =>
       buildCast[Boolean](_, b => if (b) 1 else 0)
     case DateType =>
@@ -1418,7 +1418,7 @@ abstract class CastBase extends UnaryExpression with TimeZoneAwareExpression wit
       (c, evPrim, evNull) =>
         code"""
           UTF8String.IntWrapper $wrapper = new UTF8String.IntWrapper();
-          if ($c.toInt($wrapper)) {
+          if ($c.nonCopyTrim().toInt($wrapper)) {
             $evPrim = $wrapper.value;
           } else {
             $evNull = true;
@@ -1447,7 +1447,7 @@ abstract class CastBase extends UnaryExpression with TimeZoneAwareExpression wit
       (c, evPrim, evNull) =>
         code"""
           UTF8String.LongWrapper $wrapper = new UTF8String.LongWrapper();
-          if ($c.toLong($wrapper)) {
+          if ($c.nonCopyTrim().toLong($wrapper)) {
             $evPrim = $wrapper.value;
           } else {
             $evNull = true;

sql/core/benchmarks/CastBenchmark-results.txt

@@ -0,0 +1,12 @@
+================================================================================================
+Benchmark trim the string
+================================================================================================
+
+Java HotSpot(TM) 64-Bit Server VM 1.8.0_231-b11 on Mac OS X 10.15.1
+Intel(R) Core(TM) i5-5287U CPU @ 2.90GHz
+Benchmark trim the string:                Best Time(ms)   Avg Time(ms)   Stdev(ms)    Rate(M/s)   Per Row(ns)   Relative
+------------------------------------------------------------------------------------------------------------------------
+cast(str as int) as c_int                          6263           8132         NaN          1.3         764.6       1.0X
+cast(str as long) as c_long                        8199           9737         NaN          1.0        1000.9       0.8X
+
+

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

@@ -60,3 +60,7 @@ 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);
+
+select cast(' 1' as int);
+select cast(' 1' as bigint);
+select cast(' 1' as double);

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: 38
 
 
 -- !query 0
@@ -287,3 +287,27 @@ 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 int)
+-- !query 35 schema
+struct<CAST( 1 AS INT):int>
+-- !query 35 output
+1
+
+
+-- !query 36
+select cast(' 1' as bigint)
+-- !query 36 schema
+struct<CAST( 1 AS BIGINT):bigint>
+-- !query 36 output
+1
+
+
+-- !query 37
+select cast(' 1' as double)
+-- !query 37 schema
+struct<CAST( 1 AS DOUBLE):double>
+-- !query 37 output
+1.0

sql/core/src/test/scala/org/apache/spark/sql/execution/benchmark/CastBenchmark.scala

@@ -0,0 +1,60 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.benchmark
+
+import org.apache.spark.benchmark.Benchmark
+
+/**
+ * Benchmark trim the string when casting string type to Boolean/Numeric types.
+ * To run this benchmark:
+ * {{{
+ *   1. without sbt:
+ *      bin/spark-submit --class <this class> --jars <spark core test jar> <spark sql test jar>
+ *   2. build/sbt "sql/test:runMain <this class>"
+ *   3. generate result: SPARK_GENERATE_BENCHMARK_FILES=1 build/sbt "sql/test:runMain <this class>"
+ *      Results will be written to "benchmarks/CastBenchmark-results.txt".
+ * }}}
+ */
+object CastBenchmark extends SqlBasedBenchmark {
+
+  override def runBenchmarkSuite(mainArgs: Array[String]): Unit = {
+
+    val title = "Benchmark trim the string"
+    runBenchmark(title) {
+      withTempPath { dir =>
+        val N = 500L << 14
+        val df = spark.range(N)
+        val withoutWhitespace = "withoutWhitespace"
+        val withWhitespace = "withWhitespace"
+//        val types = Seq("int", "long", "float", "double", "decimal", "boolean")
+        val types = Seq("int", "long")
+
+        df.selectExpr(s"concat('${" " * 5}', id, '${" " * 5}') as str")
+          .write.mode("overwrite").parquet(dir.getCanonicalPath)
+        val benchmark = new Benchmark(title, N, minNumIters = 3, output = output)
+        types.foreach { t =>
+          val expr = s"cast(str as $t) as c_$t"
+          benchmark.addCase(expr) { _ =>
+            spark.read.parquet(dir.getCanonicalPath).selectExpr(expr).collect()
+          }
+        }
+        benchmark.run()
+      }
+    }
+  }
+}