[SPARK-27924][SQL] Support ANSI SQL Boolean-Predicate syntax

docs/sql-keywords.md

@@ -273,6 +273,7 @@ Below is a list of all the keywords in Spark SQL.
   <tr><td>UNCACHE</td><td>non-reserved</td><td>non-reserved</td><td>non-reserved</td></tr>
   <tr><td>UNION</td><td>reserved</td><td>strict-non-reserved</td><td>reserved</td></tr>
   <tr><td>UNIQUE</td><td>reserved</td><td>non-reserved</td><td>reserved</td></tr>
+  <tr><td>UNKNOWN</td><td>reserved</td><td>non-reserved</td><td>reserved</td></tr>
   <tr><td>UNLOCK</td><td>non-reserved</td><td>non-reserved</td><td>non-reserved</td></tr>
   <tr><td>UNSET</td><td>non-reserved</td><td>non-reserved</td><td>non-reserved</td></tr>
   <tr><td>USE</td><td>non-reserved</td><td>non-reserved</td><td>non-reserved</td></tr>

sql/catalyst/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBase.g4

@@ -660,6 +660,7 @@ predicate
     | NOT? kind=IN '(' query ')'
     | NOT? kind=(RLIKE | LIKE) pattern=valueExpression
     | IS NOT? kind=NULL
+    | IS NOT? kind=(TRUE | FALSE | UNKNOWN)
     | IS NOT? kind=DISTINCT FROM right=valueExpression
     ;
 
@@ -1330,6 +1331,7 @@ nonReserved
     | UNBOUNDED
     | UNCACHE
     | UNIQUE
+    | UNKNOWN
     | UNLOCK
     | UNSET
     | USE
@@ -1592,6 +1594,7 @@ UNBOUNDED: 'UNBOUNDED';
 UNCACHE: 'UNCACHE';
 UNION: 'UNION';
 UNIQUE: 'UNIQUE';
+UNKNOWN: 'UNKNOWN';
 UNLOCK: 'UNLOCK';
 UNSET: 'UNSET';
 USE: 'USE';

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

@@ -0,0 +1,82 @@
+/*
+ * 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
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.codegen.{CodegenContext, ExprCode}
+import org.apache.spark.sql.types.BooleanType
+
+/**
+ * String to indicate which boolean test selected.
+ */
+object BooleanTest {
+  val TRUE = "TRUE"
+  val FALSE = "FALSE"
+  val UNKNOWN = "UNKNOWN"
+
+  def calculate(input: Any, booleanValue: String): Boolean = {
+    booleanValue match {
+      case TRUE => input == true
+      case FALSE => input == false
+      case UNKNOWN => input == null
+      case _ => throw new AnalysisException("Boolean test value must be one of TRUE, " +
+        "FALSE and UNKNOWN.")
+    }
+  }
+}
+
+/**
+ * Test the value of an expression is true, false, or unknown.
+ */
+@ExpressionDescription(
+  usage = "_FUNC_(expr, booleanValue) - Returns true if `expr` equals booleanValue, " +
+    "or false otherwise.",
+  arguments = """
+    Arguments:
+      * expr - a boolean expression
+      * booleanValue - a boolean value represented by a string. booleanValue must be one
+          of TRUE, FALSE and UNKNOWN.
+  """,
+  examples = """
+    Examples:
+    > SELECT _FUNC_(1 > 2, true);
+       false
+    > SELECT _FUNC_(2 > 1, true);
+       true
+  """)
+case class BooleanTest(child: Expression, booleanValue: String)
+  extends UnaryExpression with Predicate {
+
+  override def nullable: Boolean = false
+
+  override def eval(input: InternalRow): Any = {
+    BooleanTest.calculate(child.eval(input), booleanValue)
+  }
+
+  override def doGenCode(ctx: CodegenContext, ev: ExprCode): ExprCode = {
+    defineCodeGen(ctx, ev, input =>
+      s"""
+        org.apache.spark.sql.catalyst.expressions.BooleanTest.calculate($input, "$booleanValue")
+      """
+    )
+  }
+
+  override def sql: String = s"(${child.sql} IS $booleanValue)"
+}
+

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala

@@ -1223,6 +1223,12 @@ class AstBuilder(conf: SQLConf) extends SqlBaseBaseVisitor[AnyRef] with Logging
       case c: CreateNamedStruct => c.valExprs
       case other => Seq(other)
     }
+    // Check the argument of boolean test is valid.
+    def checkBooleanTestArgs(e: Expression): Unit = e.dataType match {
+      case BooleanType | NullType =>
+      case other => throw new ParseException("argument of boolean test must be boolean or null, " +
+        s"not type $other", ctx)
+    }
 
     // Create the predicate.
     ctx.kind.getType match {
@@ -1243,6 +1249,24 @@ class AstBuilder(conf: SQLConf) extends SqlBaseBaseVisitor[AnyRef] with Logging
         IsNotNull(e)
       case SqlBaseParser.NULL =>
         IsNull(e)
+      case SqlBaseParser.TRUE if ctx.NOT != null =>
+        checkBooleanTestArgs(e)
+        Not(BooleanTest(e, BooleanTest.TRUE))
+      case SqlBaseParser.TRUE =>
+        checkBooleanTestArgs(e)
+        BooleanTest(e, BooleanTest.TRUE)
+      case SqlBaseParser.FALSE if ctx.NOT != null =>
+        checkBooleanTestArgs(e)
+        Not(BooleanTest(e, BooleanTest.FALSE))
+      case SqlBaseParser.FALSE =>
+        checkBooleanTestArgs(e)
+        BooleanTest(e, BooleanTest.FALSE)
+      case SqlBaseParser.UNKNOWN if ctx.NOT != null =>
+        checkBooleanTestArgs(e)
+        Not(BooleanTest(e, BooleanTest.UNKNOWN))
+      case SqlBaseParser.UNKNOWN =>
+        checkBooleanTestArgs(e)
+        BooleanTest(e, BooleanTest.UNKNOWN)
       case SqlBaseParser.DISTINCT if ctx.NOT != null =>
         EqualNullSafe(e, expression(ctx.right))
       case SqlBaseParser.DISTINCT =>

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/BooleanExpressionsSuite.scala

@@ -0,0 +1,54 @@
+/*
+ * 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
+
+import org.apache.spark.SparkFunSuite
+import org.apache.spark.sql.catalyst.expressions.BooleanTest._
+import org.apache.spark.sql.types._
+
+class BooleanExpressionsSuite extends SparkFunSuite with ExpressionEvalHelper {
+
+  val row0 = create_row(null)
+  val row1 = create_row(false)
+  val row2 = create_row(true)
+
+  test("istrue and isnottrue") {
+    checkEvaluation(BooleanTest(Literal.create(null, NullType), TRUE), false, row0)
+    checkEvaluation(Not(BooleanTest(Literal.create(null, NullType), TRUE)), true, row0)
+    checkEvaluation(BooleanTest(Literal.create(false, BooleanType), TRUE), false, row1)
+    checkEvaluation(Not(BooleanTest(Literal.create(false, BooleanType), TRUE)), true, row1)
+    checkEvaluation(BooleanTest(Literal.create(true, BooleanType), TRUE), true, row2)
+    checkEvaluation(Not(BooleanTest(Literal.create(true, BooleanType), TRUE)), false, row2)
+  }
+
+  test("isfalse and isnotfalse") {
+    checkEvaluation(BooleanTest(Literal.create(null, NullType), FALSE), false, row0)
+    checkEvaluation(Not(BooleanTest(Literal.create(null, NullType), FALSE)), true, row0)
+    checkEvaluation(BooleanTest(Literal.create(false, BooleanType), FALSE), true, row1)
+    checkEvaluation(Not(BooleanTest(Literal.create(false, BooleanType), FALSE)), false, row1)
+    checkEvaluation(BooleanTest(Literal.create(true, BooleanType), FALSE), false, row2)
+    checkEvaluation(Not(BooleanTest(Literal.create(true, BooleanType), FALSE)), true, row2)
+  }
+
+  test("isunknown and isnotunknown") {
+    checkEvaluation(BooleanTest(Literal.create(null, NullType), UNKNOWN), true, row0)
+    checkEvaluation(Not(BooleanTest(Literal.create(null, NullType), UNKNOWN)), false, row0)
+  }
+
+}
+

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/TableIdentifierParserSuite.scala

@@ -536,6 +536,7 @@ class TableIdentifierParserSuite extends SparkFunSuite with SQLHelper {
     "uncache",
     "union",
     "unique",
+    "unknown",
     "unlock",
     "unset",
     "use",
@@ -621,6 +622,7 @@ class TableIdentifierParserSuite extends SparkFunSuite with SQLHelper {
     "trailing",
     "union",
     "unique",
+    "unknown",
     "user",
     "using",
     "when",