[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/dsl/package.scala

@@ -110,6 +110,13 @@ package object dsl {
     def isNull: Predicate = IsNull(expr)
     def isNotNull: Predicate = IsNotNull(expr)
 
+    def isTrue: Predicate = IsTrue(expr)
+    def isNotTrue: Predicate = IsNotTrue(expr)
+    def isFalse: Predicate = IsFalse(expr)
+    def isNotFalse: Predicate = IsNotFalse(expr)
+    def isUnknown: Predicate = IsUnknown(expr)
+    def isNotUnknown: Predicate = IsNotUnknown(expr)
+
     def getItem(ordinal: Expression): UnresolvedExtractValue = UnresolvedExtractValue(expr, ordinal)
     def getField(fieldName: String): UnresolvedExtractValue =
       UnresolvedExtractValue(expr, Literal(fieldName))

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

@@ -840,3 +840,81 @@ case class GreaterThanOrEqual(left: Expression, right: Expression)
 
   protected override def nullSafeEval(input1: Any, input2: Any): Any = ordering.gteq(input1, input2)
 }
+
+trait BooleanTest extends UnaryExpression with Predicate with ExpectsInputTypes {
+
+  def boolValue: Boolean
+  def whenNull: Boolean
+
+  override def nullable: Boolean = false
+  override def inputTypes: Seq[DataType] = Seq(BooleanType)
+
+  override def eval(input: InternalRow): Any = {
+    val value = child.eval(input)
+    Option(value) match {
+      case None => whenNull
+      case other => if (whenNull) {
+        value == !boolValue
+      } else {
+        value == boolValue
+      }
+    }
+  }
+
+  override def doGenCode(ctx: CodegenContext, ev: ExprCode): ExprCode = {
+    val eval = child.genCode(ctx)
+    ev.copy(code = code"""
+      ${eval.code}
+      ${CodeGenerator.javaType(dataType)} ${ev.value} = ${CodeGenerator.defaultValue(dataType)};
+      if (${eval.isNull}) {
+        ${ev.value} = $whenNull;
+      } else if ($whenNull) {
+        ${ev.value} = ${eval.value} == !$boolValue;
+      } else {
+        ${ev.value} = ${eval.value} == $boolValue;
+      }
+      """, isNull = FalseLiteral)
+  }
+}
+
+case class IsTrue(child: Expression) extends BooleanTest {
+  override def boolValue: Boolean = true
+  override def whenNull: Boolean = false
+  override def sql: String = s"(${child.sql} IS TRUE)"
+}
+
+case class IsNotTrue(child: Expression) extends BooleanTest {
+  override def boolValue: Boolean = true
+  override def whenNull: Boolean = true
+  override def sql: String = s"(${child.sql} IS NOT TRUE)"
+}
+
+case class IsFalse(child: Expression) extends BooleanTest {
+  override def boolValue: Boolean = false
+  override def whenNull: Boolean = false
+  override def sql: String = s"(${child.sql} IS FALSE)"
+}
+
+case class IsNotFalse(child: Expression) extends BooleanTest {
+  override def boolValue: Boolean = false
+  override def whenNull: Boolean = true
+  override def sql: String = s"(${child.sql} IS NOT FALSE)"
+}
+
+object IsUnknown {
+  def apply(child: Expression): Predicate = {
+    new IsNull(child) with ExpectsInputTypes {
+      override def inputTypes: Seq[DataType] = Seq(BooleanType)
+      override def sql: String = s"(${child.sql} IS UNKNOWN)"
+    }
+  }
+}
+
+object IsNotUnknown {
+  def apply(child: Expression): Predicate = {
+    new IsNotNull(child) with ExpectsInputTypes {
+      override def inputTypes: Seq[DataType] = Seq(BooleanType)
+      override def sql: String = s"(${child.sql} IS NOT UNKNOWN)"
+    }
+  }
+}

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

@@ -1210,6 +1210,7 @@ class AstBuilder(conf: SQLConf) extends SqlBaseBaseVisitor[AnyRef] with Logging
    * - (NOT) LIKE
    * - (NOT) RLIKE
    * - IS (NOT) NULL.
+   * - IS (NOT) (TRUE | FALSE | UNKNOWN)
    * - IS (NOT) DISTINCT FROM
    */
   private def withPredicate(e: Expression, ctx: PredicateContext): Expression = withOrigin(ctx) {
@@ -1243,6 +1244,18 @@ class AstBuilder(conf: SQLConf) extends SqlBaseBaseVisitor[AnyRef] with Logging
         IsNotNull(e)
       case SqlBaseParser.NULL =>
         IsNull(e)
+      case SqlBaseParser.TRUE => ctx.NOT match {
+        case null => IsTrue(e)
+        case _ => IsNotTrue(e)
+      }
+      case SqlBaseParser.FALSE => ctx.NOT match {
+        case null => IsFalse(e)
+        case _ => IsNotFalse(e)
+      }
+      case SqlBaseParser.UNKNOWN => ctx.NOT match {
+        case null => IsUnknown(e)
+        case _ => IsNotUnknown(e)
+      }
       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/PredicateSuite.scala

@@ -521,4 +521,31 @@ class PredicateSuite extends SparkFunSuite with ExpressionEvalHelper {
     val expected = "(('id = 1) OR ('id = 2))"
     assert(expression == expected)
   }
+
+  val row0 = create_row(null)
+  val row1 = create_row(false)
+  val row2 = create_row(true)
+
+  test("istrue and isnottrue") {
+    checkEvaluation(IsTrue(Literal.create(null, BooleanType)), false, row0)
+    checkEvaluation(IsNotTrue(Literal.create(null, BooleanType)), true, row0)
+    checkEvaluation(IsTrue(Literal.create(false, BooleanType)), false, row1)
+    checkEvaluation(IsNotTrue(Literal.create(false, BooleanType)), true, row1)
+    checkEvaluation(IsTrue(Literal.create(true, BooleanType)), true, row2)
+    checkEvaluation(IsNotTrue(Literal.create(true, BooleanType)), false, row2)
+  }
+
+  test("isfalse and isnotfalse") {
+    checkEvaluation(IsFalse(Literal.create(null, BooleanType)), false, row0)
+    checkEvaluation(IsNotFalse(Literal.create(null, BooleanType)), true, row0)
+    checkEvaluation(IsFalse(Literal.create(false, BooleanType)), true, row1)
+    checkEvaluation(IsNotFalse(Literal.create(false, BooleanType)), false, row1)
+    checkEvaluation(IsFalse(Literal.create(true, BooleanType)), false, row2)
+    checkEvaluation(IsNotFalse(Literal.create(true, BooleanType)), true, row2)
+  }
+
+  test("isunknown and isnotunknown") {
+    checkEvaluation(IsUnknown(Literal.create(null, BooleanType)), true, row0)
+    checkEvaluation(IsNotUnknown(Literal.create(null, BooleanType)), false, row0)
+  }
 }

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

@@ -198,6 +198,21 @@ class ExpressionParserSuite extends AnalysisTest {
     assertEqual("a = b is not null", ('a === 'b).isNotNull)
   }
 
+  test("is true | false | unknown expressions") {
+    assertEqual("a is true", 'a.isTrue)
+    assertEqual("a is not true", 'a.isNotTrue)
+    assertEqual("a is false", 'a.isFalse)
+    assertEqual("a is not false", 'a.isNotFalse)
+    assertEqual("a is unknown", 'a.isUnknown)
+    assertEqual("a is not unknown", 'a.isNotUnknown)
+    assertEqual("a = b is true", ('a === 'b).isTrue)
+    assertEqual("a = b is not true", ('a === 'b).isNotTrue)
+    assertEqual("a = b is false", ('a === 'b).isFalse)
+    assertEqual("a = b is not false", ('a === 'b).isNotFalse)
+    assertEqual("a = b is unknown", ('a === 'b).isUnknown)
+    assertEqual("a = b is not unknown", ('a === 'b).isNotUnknown)
+  }
+
   test("is distinct expressions") {
     assertEqual("a is distinct from b", !('a <=> 'b))
     assertEqual("a is not distinct from b", 'a <=> 'b)

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",