[SPARK-24393][SQL] SQL builtin: isinf

R/pkg/NAMESPACE

@@ -281,6 +281,8 @@ exportMethods("%<=>%",
               "initcap",
               "input_file_name",
               "instr",
+              "isInf",
+              "isinf",
               "isNaN",
               "isNotNull",
               "isNull",

R/pkg/R/column.R

@@ -67,7 +67,7 @@ operators <- list(
   # we can not override `&&` and `||`, so use `&` and `|` instead
   "&" = "and", "|" = "or", "^" = "pow"
 )
-column_functions1 <- c("asc", "desc", "isNaN", "isNull", "isNotNull")
+column_functions1 <- c("asc", "desc", "isInf", "isNaN", "isNull", "isNotNull")
 column_functions2 <- c("like", "rlike", "getField", "getItem", "contains")
 
 createOperator <- function(op) {

R/pkg/R/functions.R

@@ -907,6 +907,17 @@ setMethod("initcap",
             column(jc)
           })
 
+#' @details
+#' \code{isinf}: Returns true if the column is Infinity.
+#' @rdname column_nonaggregate_functions
+#' @note isinf since 2.4.0
+setMethod("isinf",
+          signature(x = "Column"),
+          function(x) {
+            jc <- callJStatic("org.apache.spark.sql.functions", "isinf", x@jc)
+            column(jc)
+          })
+
 #' @details
 #' \code{isnan}: Returns true if the column is NaN.
 #' @rdname column_nonaggregate_functions

R/pkg/R/generics.R

@@ -695,6 +695,12 @@ setGeneric("getField", function(x, ...) { standardGeneric("getField") })
 #' @rdname columnfunctions
 setGeneric("getItem", function(x, ...) { standardGeneric("getItem") })
 
+#' @rdname columnfunctions
+setGeneric("isInf", function(x) { standardGeneric("isInf") })
+
+#' @rdname columnfunctions
+setGeneric("isinf", function(x) { standardGeneric("isinf") })
+
 #' @rdname columnfunctions
 setGeneric("isNaN", function(x) { standardGeneric("isNaN") })
 

python/pyspark/sql/functions.py

@@ -468,6 +468,18 @@ def input_file_name():
     return Column(sc._jvm.functions.input_file_name())
 
 
+@since(2.4)
+def isinf(col):
+    """An expression that returns true iff the column is Inf.
+
+    >>> df = spark.createDataFrame([(1.0, float('inf')), (float('inf'), 2.0)], ("a", "b"))
+    >>> df.select(isinf("a").alias("r1"), isinf(df.a).alias("r2")).collect()
+    [Row(r1=False, r2=False), Row(r1=True, r2=True)]
+    """
+    sc = SparkContext._active_spark_context
+    return Column(sc._jvm.functions.isinf(_to_java_column(col)))
+
+
 @since(1.6)
 def isnan(col):
     """An expression that returns true iff the column is NaN.

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

@@ -198,6 +198,7 @@ object FunctionRegistry {
     expression[If]("if"),
     expression[Inline]("inline"),
     expressionGeneratorOuter[Inline]("inline_outer"),
+    expression[IsInf]("isinf"),
     expression[IsNaN]("isnan"),
     expression[IfNull]("ifnull"),
     expression[IsNull]("isnull"),

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

@@ -199,6 +199,50 @@ case class Nvl2(expr1: Expression, expr2: Expression, expr3: Expression, child:
   override def sql: String = s"$prettyName(${expr1.sql}, ${expr2.sql}, ${expr3.sql})"
 }
 
+/**
+ * Evaluates to `true` iff it's Infinity.
+ */
+@ExpressionDescription(
+  usage = "_FUNC_(expr) - Returns true if expr evaluates to infinite else returns false ",
+  examples = """
+    Examples:
+      > SELECT _FUNC_(1/0);
+       true
+      > SELECT _FUNC_(5);
+       false
+  """)
+case class IsInf(child: Expression) extends UnaryExpression
+  with Predicate with ImplicitCastInputTypes {
+
+  override def inputTypes: Seq[AbstractDataType] = Seq(TypeCollection(DoubleType, FloatType))
+
+  override def nullable: Boolean = false
+
+  override def eval(input: InternalRow): Boolean = {
+    val value = child.eval(input)
+    if (value == null) {
+      false
+    } else {
+      child.dataType match {
+        case DoubleType => value.asInstanceOf[Double].isInfinity
+        case FloatType => value.asInstanceOf[Float].isInfinity
+      }
+    }
+  }
+
+
+  override def doGenCode(ctx: CodegenContext, ev: ExprCode): ExprCode = {
+    val eval = child.genCode(ctx)
+    child.dataType match {
+      case DoubleType | FloatType =>
+        ev.copy(code = code"""
+          ${eval.code}
+          ${CodeGenerator.javaType(dataType)} ${ev.value} = \
+          !${eval.isNull} && Double.isInfinite(${eval.value});""",
+          isNull = FalseLiteral)
+    }
+  }
+}
 
 /**
  * Evaluates to `true` iff it's NaN.

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

@@ -56,6 +56,18 @@ class NullExpressionsSuite extends SparkFunSuite with ExpressionEvalHelper {
     assert(ex.contains("Null value appeared in non-nullable field"))
   }
 
+  test("IsInf") {
+    checkEvaluation(IsInf(Literal(Double.PositiveInfinity)), true)
+    checkEvaluation(IsInf(Literal(Double.NegativeInfinity)), true)
+    checkEvaluation(IsInf(Literal(Float.PositiveInfinity)), true)
+    checkEvaluation(IsInf(Literal(Float.NegativeInfinity)), true)
+    checkEvaluation(IsInf(Literal.create(null, DoubleType)), false)
+    checkEvaluation(IsInf(Literal(Float.MaxValue)), false)
+    checkEvaluation(IsInf(Literal(5.5f)), false)
+    checkEvaluation(IsInf(Literal(Float.NaN)), expected = false)
+    checkEvaluation(IsInf(Literal(Double.NaN)), expected = false)
+  }
+
   test("IsNaN") {
     checkEvaluation(IsNaN(Literal(Double.NaN)), true)
     checkEvaluation(IsNaN(Literal(Float.NaN)), true)

sql/core/src/main/scala/org/apache/spark/sql/Column.scala

@@ -557,6 +557,14 @@ class Column(val expr: Expression) extends Logging {
     (this >= lowerBound) && (this <= upperBound)
   }
 
+  /**
+   * True if the current expression is Inf.
+   *
+   * @group expr_ops
+   * @since 2.4.0
+   */
+  def isInf: Column = withExpr { IsInf(expr) }
+
   /**
    * True if the current expression is NaN.
    *

sql/core/src/main/scala/org/apache/spark/sql/functions.scala

@@ -1107,6 +1107,14 @@ object functions {
    */
   def input_file_name(): Column = withExpr { InputFileName() }
 
+  /**
+   * Return true iff the column is Infinity.
+   *
+   * @group normal_funcs
+   * @since 2.4.0
+   */
+  def isinf(e: Column): Column = withExpr { IsInf(e.expr) }
+
   /**
    * Return true iff the column is NaN.
    *