diff --git a/python/pyspark/sql/functions.py b/python/pyspark/sql/functions.py index 0e76182e0e02d..a47436f8c237f 100644 --- a/python/pyspark/sql/functions.py +++ b/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 = { diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/FunctionRegistry.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/FunctionRegistry.scala index 11538bd31b4fd..2a52763ecfbd0 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/FunctionRegistry.scala +++ b/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"), diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/CurrentUser.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/CurrentUser.scala new file mode 100644 index 0000000000000..9eb7aca65cf7a --- /dev/null +++ b/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" + +} diff --git a/sql/core/src/main/scala/org/apache/spark/sql/functions.scala b/sql/core/src/main/scala/org/apache/spark/sql/functions.scala index 53b2552fa3b36..b69af7e00c5b6 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/functions.scala +++ b/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 ////////////////////////////////////////////////////////////////////////////////////////////// diff --git a/sql/core/src/test/scala/org/apache/spark/sql/DataFrameFunctionsSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/DataFrameFunctionsSuite.scala index 50e475984f458..5165c79f29ff4 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/DataFrameFunctionsSuite.scala +++ b/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") diff --git a/sql/core/src/test/scala/org/apache/spark/sql/MiscFunctionsSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/MiscFunctionsSuite.scala index a5b08f717767f..cee58904b25a9 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/MiscFunctionsSuite.scala +++ 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 {