[SPARK-7188] added python support for math DataFrame functions

Adds support for the math functions for DataFrames in PySpark.

rxin I love Davies.

Author: Burak Yavuz <brkyvz@gmail.com>

Closes #5750 from brkyvz/python-math-udfs and squashes the following commits:

7c4f563 [Burak Yavuz] removed is_math
3c4adde [Burak Yavuz] cleanup imports
d5dca3f [Burak Yavuz] moved math functions to mathfunctions
25e6534 [Burak Yavuz] addressed comments v2.0
d3f7e0f [Burak Yavuz] addressed comments and added tests
7b7d7c4 [Burak Yavuz] remove tests for removed methods
33c2c15 [Burak Yavuz] fixed python style
3ee0c05 [Burak Yavuz] added python functions

3 files changed, 29 insertions, 117 deletions

diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/mathfuncs/binary.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/mathfuncs/binary.scala
index 5b4d912a64..fcc06d3aa1 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/mathfuncs/binary.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/mathfuncs/binary.scala
@@ -65,12 +65,6 @@ abstract class BinaryMathExpression(f: (Double, Double) => Double, name: String)
   }
 }
 
-case class Pow(left: Expression, right: Expression) extends BinaryMathExpression(math.pow, "POWER")
-
-case class Hypot(
-    left: Expression,
-    right: Expression) extends BinaryMathExpression(math.hypot, "HYPOT")
-
 case class Atan2(
     left: Expression,
     right: Expression) extends BinaryMathExpression(math.atan2, "ATAN2") {
@@ -91,3 +85,9 @@ case class Atan2(
     }
   }
 }
+
+case class Hypot(
+    left: Expression,
+    right: Expression) extends BinaryMathExpression(math.hypot, "HYPOT")
+
+case class Pow(left: Expression, right: Expression) extends BinaryMathExpression(math.pow, "POWER")
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/mathfuncs/unary.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/mathfuncs/unary.scala
index 96cb77d487..dc68469e06 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/mathfuncs/unary.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/mathfuncs/unary.scala
@@ -25,27 +25,16 @@ import org.apache.spark.sql.types._
  * input format, therefore these functions extend `ExpectsInputTypes`.
  * @param name The short name of the function
  */
-abstract class MathematicalExpression(name: String)
+abstract class MathematicalExpression(f: Double => Double, name: String)
   extends UnaryExpression with Serializable with ExpectsInputTypes {
   self: Product =>
   type EvaluatedType = Any
 
+  override def expectedChildTypes: Seq[DataType] = Seq(DoubleType)
   override def dataType: DataType = DoubleType
   override def foldable: Boolean = child.foldable
   override def nullable: Boolean = true
   override def toString: String = s"$name($child)"
-}
-
-/**
- * A unary expression specifically for math functions that take a `Double` as input and return
- * a `Double`.
- * @param f The math function.
- * @param name The short name of the function
- */
-abstract class MathematicalExpressionForDouble(f: Double => Double, name: String)
-  extends MathematicalExpression(name) { self: Product =>
-  
-  override def expectedChildTypes: Seq[DataType] = Seq(DoubleType)
 
   override def eval(input: Row): Any = {
     val evalE = child.eval(input)
@@ -58,111 +47,46 @@ abstract class MathematicalExpressionForDouble(f: Double => Double, name: String
   }
 }
 
-/**
- * A unary expression specifically for math functions that take an `Int` as input and return
- * an `Int`.
- * @param f The math function.
- * @param name The short name of the function
- */
-abstract class MathematicalExpressionForInt(f: Int => Int, name: String)
-  extends MathematicalExpression(name) { self: Product =>
+case class Acos(child: Expression) extends MathematicalExpression(math.acos, "ACOS")
 
-  override def dataType: DataType = IntegerType
-  override def expectedChildTypes: Seq[DataType] = Seq(IntegerType)
+case class Asin(child: Expression) extends MathematicalExpression(math.asin, "ASIN")
 
-  override def eval(input: Row): Any = {
-    val evalE = child.eval(input)
-    if (evalE == null) null else f(evalE.asInstanceOf[Int])
-  }
-}
+case class Atan(child: Expression) extends MathematicalExpression(math.atan, "ATAN")
 
-/**
- * A unary expression specifically for math functions that take a `Float` as input and return
- * a `Float`.
- * @param f The math function.
- * @param name The short name of the function
- */
-abstract class MathematicalExpressionForFloat(f: Float => Float, name: String)
-  extends MathematicalExpression(name) { self: Product =>
+case class Cbrt(child: Expression) extends MathematicalExpression(math.cbrt, "CBRT")
 
-  override def dataType: DataType = FloatType
-  override def expectedChildTypes: Seq[DataType] = Seq(FloatType)
+case class Ceil(child: Expression) extends MathematicalExpression(math.ceil, "CEIL")
 
-  override def eval(input: Row): Any = {
-    val evalE = child.eval(input)
-    if (evalE == null) {
-      null
-    } else {
-      val result = f(evalE.asInstanceOf[Float])
-      if (result.isNaN) null else result
-    }
-  }
-}
-
-/**
- * A unary expression specifically for math functions that take a `Long` as input and return
- * a `Long`.
- * @param f The math function.
- * @param name The short name of the function
- */
-abstract class MathematicalExpressionForLong(f: Long => Long, name: String)
-  extends MathematicalExpression(name) { self: Product =>
-
-  override def dataType: DataType = LongType
-  override def expectedChildTypes: Seq[DataType] = Seq(LongType)
-
-  override def eval(input: Row): Any = {
-    val evalE = child.eval(input)
-    if (evalE == null) null else f(evalE.asInstanceOf[Long])
-  }
-}
-
-case class Sin(child: Expression) extends MathematicalExpressionForDouble(math.sin, "SIN")
-
-case class Asin(child: Expression) extends MathematicalExpressionForDouble(math.asin, "ASIN")
-
-case class Sinh(child: Expression) extends MathematicalExpressionForDouble(math.sinh, "SINH")
-
-case class Cos(child: Expression) extends MathematicalExpressionForDouble(math.cos, "COS")
+case class Cos(child: Expression) extends MathematicalExpression(math.cos, "COS")
 
-case class Acos(child: Expression) extends MathematicalExpressionForDouble(math.acos, "ACOS")
+case class Cosh(child: Expression) extends MathematicalExpression(math.cosh, "COSH")
 
-case class Cosh(child: Expression) extends MathematicalExpressionForDouble(math.cosh, "COSH")
+case class Exp(child: Expression) extends MathematicalExpression(math.exp, "EXP")
 
-case class Tan(child: Expression) extends MathematicalExpressionForDouble(math.tan, "TAN")
+case class Expm1(child: Expression) extends MathematicalExpression(math.expm1, "EXPM1")
 
-case class Atan(child: Expression) extends MathematicalExpressionForDouble(math.atan, "ATAN")
+case class Floor(child: Expression) extends MathematicalExpression(math.floor, "FLOOR")
 
-case class Tanh(child: Expression) extends MathematicalExpressionForDouble(math.tanh, "TANH")
+case class Log(child: Expression) extends MathematicalExpression(math.log, "LOG")
 
-case class Ceil(child: Expression) extends MathematicalExpressionForDouble(math.ceil, "CEIL")
+case class Log10(child: Expression) extends MathematicalExpression(math.log10, "LOG10")
 
-case class Floor(child: Expression) extends MathematicalExpressionForDouble(math.floor, "FLOOR")
+case class Log1p(child: Expression) extends MathematicalExpression(math.log1p, "LOG1P")
 
-case class Rint(child: Expression) extends MathematicalExpressionForDouble(math.rint, "ROUND")
+case class Rint(child: Expression) extends MathematicalExpression(math.rint, "ROUND")
 
-case class Cbrt(child: Expression) extends MathematicalExpressionForDouble(math.cbrt, "CBRT")
+case class Signum(child: Expression) extends MathematicalExpression(math.signum, "SIGNUM")
 
-case class Signum(child: Expression) extends MathematicalExpressionForDouble(math.signum, "SIGNUM")
+case class Sin(child: Expression) extends MathematicalExpression(math.sin, "SIN")
 
-case class ISignum(child: Expression) extends MathematicalExpressionForInt(math.signum, "ISIGNUM")
+case class Sinh(child: Expression) extends MathematicalExpression(math.sinh, "SINH")
 
-case class FSignum(child: Expression) extends MathematicalExpressionForFloat(math.signum, "FSIGNUM")
+case class Tan(child: Expression) extends MathematicalExpression(math.tan, "TAN")
 
-case class LSignum(child: Expression) extends MathematicalExpressionForLong(math.signum, "LSIGNUM")
+case class Tanh(child: Expression) extends MathematicalExpression(math.tanh, "TANH")
 
 case class ToDegrees(child: Expression) 
-  extends MathematicalExpressionForDouble(math.toDegrees, "DEGREES")
+  extends MathematicalExpression(math.toDegrees, "DEGREES")
 
 case class ToRadians(child: Expression) 
-  extends MathematicalExpressionForDouble(math.toRadians, "RADIANS")
-
-case class Log(child: Expression) extends MathematicalExpressionForDouble(math.log, "LOG")
-
-case class Log10(child: Expression) extends MathematicalExpressionForDouble(math.log10, "LOG10")
-
-case class Log1p(child: Expression) extends MathematicalExpressionForDouble(math.log1p, "LOG1P")
-
-case class Exp(child: Expression) extends MathematicalExpressionForDouble(math.exp, "EXP")
-
-case class Expm1(child: Expression) extends MathematicalExpressionForDouble(math.expm1, "EXPM1")
+  extends MathematicalExpression(math.toRadians, "RADIANS")
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/ExpressionEvaluationSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/ExpressionEvaluationSuite.scala
index 5390ce43c6..fa71001c93 100644
--- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/ExpressionEvaluationSuite.scala
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/ExpressionEvaluationSuite.scala
@@ -1253,18 +1253,6 @@ class ExpressionEvaluationSuite extends ExpressionEvaluationBaseSuite {
     unaryMathFunctionEvaluation[Double](Signum, math.signum)
   }
 
-  test("isignum") {
-    unaryMathFunctionEvaluation[Int](ISignum, math.signum, (-5 to 5))
-  }
-
-  test("fsignum") {
-    unaryMathFunctionEvaluation[Float](FSignum, math.signum, (-5 to 5).map(_.toFloat))
-  }
-
-  test("lsignum") {
-    unaryMathFunctionEvaluation[Long](LSignum, math.signum, (5 to 5).map(_.toLong))
-  }
-
   test("log") {
     unaryMathFunctionEvaluation(Log, math.log, (0 to 20).map(_ * 0.1))
     unaryMathFunctionEvaluation(Log, math.log, (-5 to -1).map(_ * 0.1), true)