[SPARK-29837][SQL] PostgreSQL dialect: cast to boolean

### What changes were proposed in this pull request?

Make SparkSQL's `cast to boolean` behavior be consistent with PostgreSQL when
spark.sql.dialect is configured as PostgreSQL.

### Why are the changes needed?

SparkSQL and PostgreSQL have a lot different cast behavior between types by default. We should make SparkSQL's cast behavior be consistent with PostgreSQL when `spark.sql.dialect` is configured as PostgreSQL.

### Does this PR introduce any user-facing change?

Yes. If user switches to PostgreSQL dialect now, they will

* get an exception if they input a invalid string, e.g "erut", while they get `null` before;

* get an exception if they input `TimestampType`, `DateType`, `LongType`, `ShortType`, `ByteType`, `DecimalType`, `DoubleType`, `FloatType` values,  while they get `true` or `false` result before.

And here're evidences for those unsupported types from PostgreSQL:

timestamp:
```
postgres=# select cast(cast('2019-11-11' as timestamp) as boolean);
ERROR:  cannot cast type timestamp without time zone to boolean
```

date:
```
postgres=# select cast(cast('2019-11-11' as date) as boolean);
ERROR:  cannot cast type date to boolean
```

bigint:
```
postgres=# select cast(cast('20191111' as bigint) as boolean);
ERROR:  cannot cast type bigint to boolean
```

smallint:
```
postgres=# select cast(cast(2019 as smallint) as boolean);
ERROR:  cannot cast type smallint to boolean
```

bytea:
```
postgres=# select cast(cast('2019' as bytea) as boolean);
ERROR:  cannot cast type bytea to boolean
```

decimal:
```
postgres=# select cast(cast('2019' as decimal) as boolean);
ERROR:  cannot cast type numeric to boolean
```

float:
```
postgres=# select cast(cast('2019' as float) as boolean);
ERROR:  cannot cast type double precision to boolean
```

### How was this patch tested?

Added and tested manually.

Closes #26463 from Ngone51/dev-postgre-cast2bool.

Authored-by: wuyi <ngone_5451@163.com>
Signed-off-by: Wenchen Fan <wenchen@databricks.com>

Author: Ngone51

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/PostgreSQLDialect.scala

@@ -19,27 +19,27 @@ package org.apache.spark.sql.catalyst.analysis
 
 import org.apache.spark.internal.Logging
 import org.apache.spark.sql.catalyst.expressions.Cast
-import org.apache.spark.sql.catalyst.expressions.postgreSQL.PostgreCastStringToBoolean
+import org.apache.spark.sql.catalyst.expressions.postgreSQL.PostgreCastToBoolean
 import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
 import org.apache.spark.sql.catalyst.rules.Rule
 import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.types.{BooleanType, StringType}
 
 object PostgreSQLDialect {
   val postgreSQLDialectRules: List[Rule[LogicalPlan]] =
-    CastStringToBoolean ::
+    CastToBoolean ::
       Nil
 
-  object CastStringToBoolean extends Rule[LogicalPlan] with Logging {
+  object CastToBoolean extends Rule[LogicalPlan] with Logging {
     override def apply(plan: LogicalPlan): LogicalPlan = {
       // The SQL configuration `spark.sql.dialect` can be changed in runtime.
       // To make sure the configuration is effective, we have to check it during rule execution.
       val conf = SQLConf.get
       if (conf.usePostgreSQLDialect) {
         plan.transformExpressions {
-          case Cast(child, dataType, _)
-            if dataType == BooleanType && child.dataType == StringType =>
-            PostgreCastStringToBoolean(child)
+          case Cast(child, dataType, timeZoneId)
+            if child.dataType != BooleanType && dataType == BooleanType =>
+            PostgreCastToBoolean(child, timeZoneId)
         }
       } else {
         plan

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

@@ -274,7 +274,7 @@ abstract class CastBase extends UnaryExpression with TimeZoneAwareExpression wit
   private[this] def needsTimeZone: Boolean = Cast.needsTimeZone(child.dataType, dataType)
 
   // [[func]] assumes the input is no longer null because eval already does the null check.
-  @inline private[this] def buildCast[T](a: Any, func: T => Any): Any = func(a.asInstanceOf[T])
+  @inline protected def buildCast[T](a: Any, func: T => Any): Any = func(a.asInstanceOf[T])
 
   private lazy val dateFormatter = DateFormatter(zoneId)
   private lazy val timestampFormatter = TimestampFormatter.getFractionFormatter(zoneId)
@@ -377,7 +377,7 @@ abstract class CastBase extends UnaryExpression with TimeZoneAwareExpression wit
   }
 
   // UDFToBoolean
-  private[this] def castToBoolean(from: DataType): Any => Any = from match {
+  protected[this] def castToBoolean(from: DataType): Any => Any = from match {
     case StringType =>
       buildCast[UTF8String](_, s => {
         if (StringUtils.isTrueString(s)) {
@@ -782,7 +782,7 @@ abstract class CastBase extends UnaryExpression with TimeZoneAwareExpression wit
     }
   }
 
-  override def doGenCode(ctx: CodegenContext, ev: ExprCode): ExprCode = {
+  override protected def doGenCode(ctx: CodegenContext, ev: ExprCode): ExprCode = {
     val eval = child.genCode(ctx)
     val nullSafeCast = nullSafeCastFunction(child.dataType, dataType, ctx)
 
@@ -792,7 +792,7 @@ abstract class CastBase extends UnaryExpression with TimeZoneAwareExpression wit
 
   // The function arguments are: `input`, `result` and `resultIsNull`. We don't need `inputIsNull`
   // in parameter list, because the returned code will be put in null safe evaluation region.
-  private[this] type CastFunction = (ExprValue, ExprValue, ExprValue) => Block
+  protected type CastFunction = (ExprValue, ExprValue, ExprValue) => Block
 
   private[this] def nullSafeCastFunction(
       from: DataType,
@@ -1234,7 +1234,7 @@ abstract class CastBase extends UnaryExpression with TimeZoneAwareExpression wit
   private[this] def timestampToDoubleCode(ts: ExprValue): Block =
     code"$ts / (double)$MICROS_PER_SECOND"
 
-  private[this] def castToBooleanCode(from: DataType): CastFunction = from match {
+  protected[this] def castToBooleanCode(from: DataType): CastFunction = from match {
     case StringType =>
       val stringUtils = inline"${StringUtils.getClass.getName.stripSuffix("$")}"
       (c, evPrim, evNull) =>

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/postgreSQL/PostgreCastStringToBoolean.scala

@@ -0,0 +1,83 @@
+/*
+ * 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.catalyst.expressions.postgreSQL
+
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
+import org.apache.spark.sql.catalyst.expressions.{CastBase, Expression, TimeZoneAwareExpression}
+import org.apache.spark.sql.catalyst.expressions.codegen.Block._
+import org.apache.spark.sql.catalyst.util.postgreSQL.StringUtils
+import org.apache.spark.sql.types._
+import org.apache.spark.unsafe.types.UTF8String
+
+case class PostgreCastToBoolean(child: Expression, timeZoneId: Option[String])
+  extends CastBase {
+
+  override protected def ansiEnabled =
+    throw new UnsupportedOperationException("PostgreSQL dialect doesn't support ansi mode")
+
+  override def withTimeZone(timeZoneId: String): TimeZoneAwareExpression =
+    copy(timeZoneId = Option(timeZoneId))
+
+  override def checkInputDataTypes(): TypeCheckResult = child.dataType match {
+    case StringType | IntegerType | NullType =>
+      TypeCheckResult.TypeCheckSuccess
+    case _ =>
+      TypeCheckResult.TypeCheckFailure(s"cannot cast type ${child.dataType} to boolean")
+  }
+
+  override def castToBoolean(from: DataType): Any => Any = from match {
+    case StringType =>
+      buildCast[UTF8String](_, str => {
+        val s = str.trim().toLowerCase()
+        if (StringUtils.isTrueString(s)) {
+          true
+        } else if (StringUtils.isFalseString(s)) {
+          false
+        } else {
+          throw new IllegalArgumentException(s"invalid input syntax for type boolean: $s")
+        }
+      })
+    case IntegerType =>
+      super.castToBoolean(from)
+  }
+
+  override def castToBooleanCode(from: DataType): CastFunction = from match {
+    case StringType =>
+      val stringUtils = inline"${StringUtils.getClass.getName.stripSuffix("$")}"
+      (c, evPrim, evNull) =>
+        code"""
+          if ($stringUtils.isTrueString($c.trim().toLowerCase())) {
+            $evPrim = true;
+          } else if ($stringUtils.isFalseString($c.trim().toLowerCase())) {
+            $evPrim = false;
+          } else {
+            throw new IllegalArgumentException("invalid input syntax for type boolean: $c");
+          }
+        """
+
+    case IntegerType =>
+      super.castToBooleanCode(from)
+  }
+
+  override def dataType: DataType = BooleanType
+
+  override def nullable: Boolean = child.nullable
+
+  override def toString: String = s"PostgreCastToBoolean($child as ${dataType.simpleString})"
+
+  override def sql: String = s"CAST(${child.sql} AS ${dataType.sql})"
+}

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/postgreSQL/PostgreCastToBoolean.scala

@@ -16,44 +16,58 @@
  */
 package org.apache.spark.sql.catalyst.expressions.postgreSQL
 
+import java.sql.{Date, Timestamp}
+
 import org.apache.spark.SparkFunSuite
+import org.apache.spark.sql.AnalysisException
 import org.apache.spark.sql.catalyst.expressions.{ExpressionEvalHelper, Literal}
 
 class CastSuite extends SparkFunSuite with ExpressionEvalHelper {
-  private def checkPostgreCastStringToBoolean(v: Any, expected: Any): Unit = {
-    checkEvaluation(PostgreCastStringToBoolean(Literal(v)), expected)
+  private def checkPostgreCastToBoolean(v: Any, expected: Any): Unit = {
+    checkEvaluation(PostgreCastToBoolean(Literal(v), None), expected)
   }
 
   test("cast string to boolean") {
-    checkPostgreCastStringToBoolean("true", true)
-    checkPostgreCastStringToBoolean("tru", true)
-    checkPostgreCastStringToBoolean("tr", true)
-    checkPostgreCastStringToBoolean("t", true)
-    checkPostgreCastStringToBoolean("tRUe", true)
-    checkPostgreCastStringToBoolean("    tRue   ", true)
-    checkPostgreCastStringToBoolean("    tRu   ", true)
-    checkPostgreCastStringToBoolean("yes", true)
-    checkPostgreCastStringToBoolean("ye", true)
-    checkPostgreCastStringToBoolean("y", true)
-    checkPostgreCastStringToBoolean("1", true)
-    checkPostgreCastStringToBoolean("on", true)
+    checkPostgreCastToBoolean("true", true)
+    checkPostgreCastToBoolean("tru", true)
+    checkPostgreCastToBoolean("tr", true)
+    checkPostgreCastToBoolean("t", true)
+    checkPostgreCastToBoolean("tRUe", true)
+    checkPostgreCastToBoolean("    tRue   ", true)
+    checkPostgreCastToBoolean("    tRu   ", true)
+    checkPostgreCastToBoolean("yes", true)
+    checkPostgreCastToBoolean("ye", true)
+    checkPostgreCastToBoolean("y", true)
+    checkPostgreCastToBoolean("1", true)
+    checkPostgreCastToBoolean("on", true)
+
+    checkPostgreCastToBoolean("false", false)
+    checkPostgreCastToBoolean("fals", false)
+    checkPostgreCastToBoolean("fal", false)
+    checkPostgreCastToBoolean("fa", false)
+    checkPostgreCastToBoolean("f", false)
+    checkPostgreCastToBoolean("    fAlse    ", false)
+    checkPostgreCastToBoolean("    fAls    ", false)
+    checkPostgreCastToBoolean("    FAlsE    ", false)
+    checkPostgreCastToBoolean("no", false)
+    checkPostgreCastToBoolean("n", false)
+    checkPostgreCastToBoolean("0", false)
+    checkPostgreCastToBoolean("off", false)
+    checkPostgreCastToBoolean("of", false)
 
-    checkPostgreCastStringToBoolean("false", false)
-    checkPostgreCastStringToBoolean("fals", false)
-    checkPostgreCastStringToBoolean("fal", false)
-    checkPostgreCastStringToBoolean("fa", false)
-    checkPostgreCastStringToBoolean("f", false)
-    checkPostgreCastStringToBoolean("    fAlse    ", false)
-    checkPostgreCastStringToBoolean("    fAls    ", false)
-    checkPostgreCastStringToBoolean("    FAlsE    ", false)
-    checkPostgreCastStringToBoolean("no", false)
-    checkPostgreCastStringToBoolean("n", false)
-    checkPostgreCastStringToBoolean("0", false)
-    checkPostgreCastStringToBoolean("off", false)
-    checkPostgreCastStringToBoolean("of", false)
+    intercept[IllegalArgumentException](PostgreCastToBoolean(Literal("o"), None).eval())
+    intercept[IllegalArgumentException](PostgreCastToBoolean(Literal("abc"), None).eval())
+    intercept[IllegalArgumentException](PostgreCastToBoolean(Literal(""), None).eval())
+  }
 
-    checkPostgreCastStringToBoolean("o", null)
-    checkPostgreCastStringToBoolean("abc", null)
-    checkPostgreCastStringToBoolean("", null)
+  test("unsupported data types to cast to boolean") {
+    assert(PostgreCastToBoolean(Literal(new Timestamp(1)), None).checkInputDataTypes().isFailure)
+    assert(PostgreCastToBoolean(Literal(new Date(1)), None).checkInputDataTypes().isFailure)
+    assert(PostgreCastToBoolean(Literal(1.toLong), None).checkInputDataTypes().isFailure)
+    assert(PostgreCastToBoolean(Literal(1.toShort), None).checkInputDataTypes().isFailure)
+    assert(PostgreCastToBoolean(Literal(1.toByte), None).checkInputDataTypes().isFailure)
+    assert(PostgreCastToBoolean(Literal(BigDecimal(1.0)), None).checkInputDataTypes().isFailure)
+    assert(PostgreCastToBoolean(Literal(1.toDouble), None).checkInputDataTypes().isFailure)
+    assert(PostgreCastToBoolean(Literal(1.toFloat), None).checkInputDataTypes().isFailure)
   }
 }