[SPARK-8220][SQL]Add positive identify function

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

@@ -120,6 +120,7 @@ object FunctionRegistry {
     expression[Log2]("log2"),
     expression[Pow]("pow"),
     expression[Pow]("power"),
+    expression[UnaryPositive]("positive"),
     expression[Rint]("rint"),
     expression[Signum]("sign"),
     expression[Signum]("signum"),

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

@@ -58,6 +58,15 @@ case class UnaryMinus(child: Expression) extends UnaryArithmetic {
   protected override def evalInternal(evalE: Any) = numeric.negate(evalE)
 }
 
+case class UnaryPositive(child: Expression) extends UnaryArithmetic {
+  override def toString: String = s"positive($child)"
+
+  override def genCode(ctx: CodeGenContext, ev: GeneratedExpressionCode): String =
+    defineCodeGen(ctx, ev, c => c)
+
+  protected override def evalInternal(evalE: Any) = evalE
+}
+
 case class Sqrt(child: Expression) extends UnaryArithmetic {
   override def dataType: DataType = DoubleType
   override def nullable: Boolean = true

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala

@@ -52,6 +52,7 @@ object DefaultOptimizer extends Optimizer {
       LikeSimplification,
       BooleanSimplification,
       PushPredicateThroughJoin,
+      RemovePositive,
       SimplifyFilters,
       SimplifyCasts,
       SimplifyCaseConversionExpressions) ::
@@ -632,6 +633,15 @@ object SimplifyCasts extends Rule[LogicalPlan] {
   }
 }
 
+/**
+ * Removes [[UnaryPositive]] identify function
+ */
+object RemovePositive extends Rule[LogicalPlan] {
+  def apply(plan: LogicalPlan): LogicalPlan = plan transformAllExpressions {
+    case UnaryPositive(child) => child
+  }
+}
+
 /**
  * Combines two adjacent [[Limit]] operators into one, merging the
  * expressions into one single expression.

sql/core/src/test/scala/org/apache/spark/sql/MathExpressionsSuite.scala

@@ -262,4 +262,11 @@ class MathExpressionsSuite extends QueryTest {
       ctx.sql("SELECT negative(1), negative(0), negative(-1)"),
       Row(-1, 0, 1))
   }
+
+  test("positive") {
+    val df = Seq((1, -1, "abc")).toDF("a", "b", "c")
+    checkAnswer(df.selectExpr("positive(a)"), Row(1))
+    checkAnswer(df.selectExpr("positive(b)"), Row(-1))
+    checkAnswer(df.selectExpr("positive(c)"), Row("abc"))
+  }
 }