[SPARK-21957][SQL] Add current_user function

python/pyspark/sql/functions.py

@@ -1296,6 +1296,15 @@ def hash(*cols):
     return Column(jc)
 
 
+@since(2.3)
+def current_user():
+    """
+    Returns current user executing the query as a :class:`StringType` column column.
+    """
+    sc = SparkContext._active_spark_context
+    return Column(sc._jvm.functions.current_user())
+
+
 # ---------------------- String/Binary functions ------------------------------
 
 _string_functions = {

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

@@ -418,6 +418,7 @@ object FunctionRegistry {
     expression[Sha1]("sha"),
     expression[Sha1]("sha1"),
     expression[Sha2]("sha2"),
+    expression[CurrentUser]("current_user"),
     expression[SparkPartitionID]("spark_partition_id"),
     expression[InputFileName]("input_file_name"),
     expression[InputFileBlockStart]("input_file_block_start"),

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

@@ -0,0 +1,47 @@
+/*
+ * 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.SparkContext
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.codegen.CodegenFallback
+import org.apache.spark.sql.types.{DataType, StringType}
+
+/**
+ * Return the user executing the current query.
+ *
+ * There is no code generation since this expression should get constant folded by the optimizer.
+ */
+@ExpressionDescription(
+  usage = """
+    _FUNC_() - Returns the current user executing the query.
+  """,
+  since = "2.3.0")
+case class CurrentUser() extends LeafExpression with CodegenFallback {
+  override def foldable: Boolean = true
+  override def nullable: Boolean = false
+  override def dataType: DataType = StringType
+
+  /** Returns the result of evaluating this expression on a given input Row */
+  override def eval(input: InternalRow): Any = {
+    SparkContext.getActive.map(_.sparkUser).getOrElse("")
+  }
+
+  override def prettyName: String = "current_user"
+
+}

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

@@ -2148,6 +2148,16 @@ object functions {
     new Murmur3Hash(cols.map(_.expr))
   }
 
+  /**
+   * Returns current user executing the query as a string column.
+   *
+   * @group misc_funcs
+   * @since 2.3.0
+   */
+  def current_user(): Column = withExpr {
+    CurrentUser()
+  }
+
   //////////////////////////////////////////////////////////////////////////////////////////////
   // String functions
   //////////////////////////////////////////////////////////////////////////////////////////////

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

@@ -209,6 +209,13 @@ class DataFrameFunctionsSuite extends QueryTest with SharedSQLContext {
       Row(2743272264L, 2180413220L))
   }
 
+  test("misc current_user function") {
+    checkAnswer(
+      spark.sql("select current_user()"),
+      Row(spark.sparkContext.sparkUser)
+    )
+  }
+
   test("string function find_in_set") {
     val df = Seq(("abc,b,ab,c,def", "abc,b,ab,c,def")).toDF("a", "b")
 

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

@@ -17,6 +17,7 @@
 
 package org.apache.spark.sql
 
+import org.apache.spark.sql.functions.current_user
 import org.apache.spark.sql.test.SharedSQLContext
 
 class MiscFunctionsSuite extends QueryTest with SharedSQLContext {
@@ -31,6 +32,13 @@ class MiscFunctionsSuite extends QueryTest with SharedSQLContext {
         s"java_method('$className', 'method1', a, b)"),
       Row("m1one", "m1one"))
   }
+
+  test("current_user") {
+    val df = Seq(1, 2).toDF("a")
+    checkAnswer(
+      df.select(current_user()).limit(1),
+      Row(spark.sparkContext.sparkUser))
+  }
 }
 
 object ReflectClass {