[SPARK-21957][SQL][FOLLOWUP] Support CURRENT_USER without tailing par…

…entheses

### What changes were proposed in this pull request?

A followup for 345d35e, in this PR we support CURRENT_USER without tailing parentheses in default mode. And for ANSI mode, we can only use CURRENT_USER without tailing parentheses because it is a reserved keyword that cannot be used as a function name

### Why are the changes needed?

1. make it the same as current_date/current_timestamp
2. better ANSI compliance
### Does this PR introduce _any_ user-facing change?

no, just a followup

### How was this patch tested?

new tests

Closes apache#32770 from yaooqinn/SPARK-21957-F.

Authored-by: Kent Yao <yao@apache.org>
Signed-off-by: Wenchen Fan <wenchen@databricks.com>

sql/catalyst/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBase.g4

@@ -816,7 +816,7 @@ valueExpression
     ;
 
 primaryExpression
-    : name=(CURRENT_DATE | CURRENT_TIMESTAMP)                                                  #currentDatetime
+    : name=(CURRENT_DATE | CURRENT_TIMESTAMP | CURRENT_USER)                                   #currentLike
     | CASE whenClause+ (ELSE elseExpression=expression)? END                                   #searchedCase
     | CASE value=expression whenClause+ (ELSE elseExpression=expression)? END                  #simpleCase
     | name=(CAST | TRY_CAST) '(' expression AS dataType ')'                                    #cast

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

@@ -1652,6 +1652,7 @@ class Analyzer(override val catalogManager: CatalogManager)
   private val literalFunctions: Seq[(String, () => Expression, Expression => String)] = Seq(
     (CurrentDate().prettyName, () => CurrentDate(), toPrettySQL(_)),
     (CurrentTimestamp().prettyName, () => CurrentTimestamp(), toPrettySQL(_)),
+    (CurrentUser().prettyName, () => CurrentUser(), toPrettySQL),
     (VirtualColumn.hiveGroupingIdName, () => GroupingID(Nil), _ => VirtualColumn.hiveGroupingIdName)
   )
 

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala

@@ -1683,13 +1683,15 @@ class AstBuilder extends SqlBaseBaseVisitor[AnyRef] with SQLConfHelper with Logg
     }
   }
 
-  override def visitCurrentDatetime(ctx: CurrentDatetimeContext): Expression = withOrigin(ctx) {
+  override def visitCurrentLike(ctx: CurrentLikeContext): Expression = withOrigin(ctx) {
     if (conf.ansiEnabled) {
       ctx.name.getType match {
         case SqlBaseParser.CURRENT_DATE =>
           CurrentDate()
         case SqlBaseParser.CURRENT_TIMESTAMP =>
           CurrentTimestamp()
+        case SqlBaseParser.CURRENT_USER =>
+          CurrentUser()
       }
     } else {
       // If the parser is not in ansi mode, we should return `UnresolvedAttribute`, in case there

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

@@ -18,6 +18,8 @@
 package org.apache.spark.sql
 
 import org.apache.spark.{SPARK_REVISION, SPARK_VERSION_SHORT}
+import org.apache.spark.sql.catalyst.parser.ParseException
+import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.test.SharedSparkSession
 
 class MiscFunctionsSuite extends QueryTest with SharedSparkSession {
@@ -42,8 +44,17 @@ class MiscFunctionsSuite extends QueryTest with SharedSparkSession {
   }
 
   test("SPARK-21957: get current_user in normal spark apps") {
-    val df = sql("select current_user()")
-    checkAnswer(df, Row(spark.sparkContext.sparkUser))
+    val user = spark.sparkContext.sparkUser
+    withSQLConf(SQLConf.ANSI_ENABLED.key -> "false") {
+      val df = sql("select current_user(), current_user")
+      checkAnswer(df, Row(user, user))
+    }
+    withSQLConf(SQLConf.ANSI_ENABLED.key -> "true") {
+      val df = sql("select current_user")
+      checkAnswer(df, Row(spark.sparkContext.sparkUser))
+      val e = intercept[ParseException](sql("select current_user()"))
+      assert(e.getMessage.contains("current_user"))
+    }
   }
 }
 

sql/hive-thriftserver/src/test/scala/org/apache/spark/sql/hive/thriftserver/ThriftServerWithSparkContextSuite.scala

@@ -20,7 +20,7 @@ package org.apache.spark.sql.hive.thriftserver
 import java.sql.SQLException
 import java.util.concurrent.atomic.AtomicBoolean
 
-import org.apache.hive.service.cli.HiveSQLException
+import org.apache.hive.service.cli.{HiveSQLException, OperationHandle}
 
 import org.apache.spark.TaskKilled
 import org.apache.spark.scheduler.{SparkListener, SparkListenerTaskEnd}
@@ -125,9 +125,23 @@ trait ThriftServerWithSparkContextSuite extends SharedThriftServer {
     withCLIServiceClient(clientUser) { client =>
       val sessionHandle = client.openSession(clientUser, "")
       val confOverlay = new java.util.HashMap[java.lang.String, java.lang.String]
-      val opHandle = client.executeStatement(sessionHandle, sql, confOverlay)
-      val rowSet = client.fetchResults(opHandle)
-      assert(rowSet.toTRowSet.getColumns.get(0).getStringVal.getValues.get(0) === clientUser)
+      val exec: String => OperationHandle = client.executeStatement(sessionHandle, _, confOverlay)
+
+      exec(s"set ${SQLConf.ANSI_ENABLED.key}=false")
+
+      val opHandle1 = exec("select current_user(), current_user")
+      val rowSet1 = client.fetchResults(opHandle1)
+      rowSet1.toTRowSet.getColumns.forEach { col =>
+        assert(col.getStringVal.getValues.get(0) === clientUser)
+      }
+
+      exec(s"set ${SQLConf.ANSI_ENABLED.key}=true")
+      val opHandle2 = exec("select current_user")
+      assert(client.fetchResults(opHandle2).toTRowSet.getColumns.get(0)
+        .getStringVal.getValues.get(0) === clientUser)
+
+      val e = intercept[HiveSQLException](exec("select current_user()"))
+      assert(e.getMessage.contains("current_user"))
     }
   }
 }