[SPARK-30352][SQL] DataSourceV2: Add CURRENT_CATALOG function

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

@@ -490,6 +490,7 @@ object FunctionRegistry {
     expression[InputFileBlockLength]("input_file_block_length"),
     expression[MonotonicallyIncreasingID]("monotonically_increasing_id"),
     expression[CurrentDatabase]("current_database"),
+    expression[CurrentCatalog]("current_catalog"),
     expression[CallMethodViaReflection]("reflect"),
     expression[CallMethodViaReflection]("java_method"),
     expression[SparkVersion]("version"),

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

@@ -116,6 +116,24 @@ case class CurrentDatabase() extends LeafExpression with Unevaluable {
   override def prettyName: String = "current_database"
 }
 
+/**
+ * Returns the current catalog.
+ */
+@ExpressionDescription(
+  usage = "_FUNC_() - Returns the current catalog.",
+  examples = """
+    Examples:
+      > SELECT _FUNC_();
+       spark_catalog
+  """,
+  since = "3.0.0")
+case class CurrentCatalog() extends LeafExpression with Unevaluable {
+  override def dataType: DataType = StringType
+  override def foldable: Boolean = true
+  override def nullable: Boolean = false
+  override def prettyName: String = "current_catalog"
+}
+
 // scalastyle:off line.size.limit
 @ExpressionDescription(
   usage = """_FUNC_() - Returns an universally unique identifier (UUID) string. The value is returned as a canonical UUID 36-character string.""",

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

@@ -130,6 +130,7 @@ abstract class Optimizer(catalogManager: CatalogManager)
       ReplaceExpressions,
       ComputeCurrentTime,
       GetCurrentDatabase(catalogManager),
+      GetCurrentCatalog(catalogManager),
       RewriteDistinctAggregates,
       ReplaceDeduplicateWithAggregate) ::
     //////////////////////////////////////////////////////////////////////////////////////////
@@ -220,6 +221,7 @@ abstract class Optimizer(catalogManager: CatalogManager)
       ReplaceExpressions.ruleName ::
       ComputeCurrentTime.ruleName ::
       GetCurrentDatabase(catalogManager).ruleName ::
+      GetCurrentCatalog(catalogManager).ruleName ::
       RewriteDistinctAggregates.ruleName ::
       ReplaceDeduplicateWithAggregate.ruleName ::
       ReplaceIntersectWithSemiJoin.ruleName ::

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

@@ -88,3 +88,14 @@ case class GetCurrentDatabase(catalogManager: CatalogManager) extends Rule[Logic
     }
   }
 }
+
+/** Replaces the expression of [[CurrentCatalog]] with the current catalog name. */
+case class GetCurrentCatalog(catalogManager: CatalogManager) extends Rule[LogicalPlan] {
+  def apply(plan: LogicalPlan): LogicalPlan = {
+    val currentCatalog = catalogManager.currentCatalog.name()
+
+    plan transformAllExpressions {
+      case CurrentCatalog() => Literal.create(currentCatalog, StringType)
+    }
+  }
+}

sql/core/src/test/resources/sql-tests/inputs/misc-functions.sql

@@ -8,3 +8,6 @@ select typeof(cast(1.0 as float)), typeof(1.0D), typeof(1.2);
 select typeof(date '1986-05-23'),  typeof(timestamp '1986-05-23'), typeof(interval '23 days');
 select typeof(x'ABCD'), typeof('SPARK');
 select typeof(array(1, 2)), typeof(map(1, 2)), typeof(named_struct('a', 1, 'b', 'spark'));
+
+-- get current_catalog
+select current_catalog();

sql/core/src/test/resources/sql-tests/results/misc-functions.sql.out

@@ -1,5 +1,5 @@
 -- Automatically generated by SQLQueryTestSuite
--- Number of queries: 7
+-- Number of queries: 8
 
 
 -- !query 0
@@ -56,3 +56,11 @@ select typeof(array(1, 2)), typeof(map(1, 2)), typeof(named_struct('a', 1, 'b',
 struct<typeof(array(1, 2)):string,typeof(map(1, 2)):string,typeof(named_struct(a, 1, b, spark)):string>
 -- !query 6 output
 array<int>	map<int,int>	struct<a:int,b:string>
+
+
+-- !query 7
+select current_catalog()
+-- !query 7 schema
+struct<current_catalog():string>
+-- !query 7 output
+spark_catalog