[SPARK-54810] Path

common/utils/src/main/resources/error/error-conditions.json

@@ -1929,6 +1929,12 @@
     ],
     "sqlState" : "23505"
   },
+  "DUPLICATE_PATH_ENTRY" : {
+    "message" : [
+      "Duplicate path entry <pathEntry>. The session path cannot contain the same catalog.schema more than once (including after expanding shortcuts like DEFAULT_PATH or SYSTEM_PATH)."
+    ],
+    "sqlState" : "42P10"
+  },
   "DUPLICATE_ROUTINE_PARAMETER_ASSIGNMENT" : {
     "message" : [
       "Call to routine <routineName> is invalid because it includes multiple argument assignments to the same parameter name <parameterName>."
@@ -7615,6 +7621,11 @@
           "Cannot have VARIANT type columns in DataFrame which calls set operations (INTERSECT, EXCEPT, etc.), but the type of column <colName> is <dataType>."
         ]
       },
+      "SET_PATH_VIA_SET" : {
+        "message" : [
+          "The session path cannot be set using the SET statement. Use SET PATH = ... instead."
+        ]
+      },
       "SET_PROPERTIES_AND_DBPROPERTIES" : {
         "message" : [
           "set PROPERTIES and DBPROPERTIES at the same time."

docs/sql-ref-ansi-compliance.md

@@ -477,7 +477,10 @@ Below is a list of all the keywords in Spark SQL.
 |CROSS|reserved|strict-non-reserved|reserved|
 |CUBE|non-reserved|non-reserved|reserved|
 |CURRENT|non-reserved|non-reserved|reserved|
+|CURRENT_DATABASE|non-reserved|non-reserved|non-reserved|
 |CURRENT_DATE|reserved|non-reserved|reserved|
+|CURRENT_PATH|non-reserved|non-reserved|reserved|
+|CURRENT_SCHEMA|non-reserved|non-reserved|non-reserved|
 |CURRENT_TIME|reserved|non-reserved|reserved|
 |CURRENT_TIMESTAMP|reserved|non-reserved|reserved|
 |CURRENT_USER|reserved|non-reserved|reserved|
@@ -500,6 +503,7 @@ Below is a list of all the keywords in Spark SQL.
 |DEFAULT|non-reserved|non-reserved|non-reserved|
 |DEFINED|non-reserved|non-reserved|non-reserved|
 |DEFINER|non-reserved|non-reserved|non-reserved|
+|DEFAULT_PATH|non-reserved|non-reserved|not a keyword|
 |DELAY|non-reserved|non-reserved|non-reserved|
 |DELETE|non-reserved|non-reserved|reserved|
 |DELIMITED|non-reserved|non-reserved|non-reserved|
@@ -667,6 +671,7 @@ Below is a list of all the keywords in Spark SQL.
 |PARTITION|non-reserved|non-reserved|reserved|
 |PARTITIONED|non-reserved|non-reserved|non-reserved|
 |PARTITIONS|non-reserved|non-reserved|non-reserved|
+|PATH|non-reserved|non-reserved|not a keyword|
 |PERCENT|non-reserved|non-reserved|non-reserved|
 |PIVOT|non-reserved|non-reserved|non-reserved|
 |PLACING|non-reserved|non-reserved|non-reserved|
@@ -750,6 +755,7 @@ Below is a list of all the keywords in Spark SQL.
 |SUBSTR|non-reserved|non-reserved|non-reserved|
 |SUBSTRING|non-reserved|non-reserved|non-reserved|
 |SYNC|non-reserved|non-reserved|non-reserved|
+|SYSTEM_PATH|non-reserved|non-reserved|not a keyword|
 |SYSTEM_TIME|non-reserved|non-reserved|non-reserved|
 |SYSTEM_VERSION|non-reserved|non-reserved|non-reserved|
 |TABLE|reserved|non-reserved|reserved|

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

@@ -196,7 +196,10 @@ CREATE: 'CREATE';
 CROSS: 'CROSS';
 CUBE: 'CUBE';
 CURRENT: 'CURRENT';
+CURRENT_DATABASE: 'CURRENT_DATABASE';
 CURRENT_DATE: 'CURRENT_DATE';
+CURRENT_PATH: 'CURRENT_PATH';
+CURRENT_SCHEMA: 'CURRENT_SCHEMA';
 CURRENT_TIME: 'CURRENT_TIME';
 CURRENT_TIMESTAMP: 'CURRENT_TIMESTAMP';
 CURRENT_USER: 'CURRENT_USER';
@@ -217,6 +220,7 @@ DEC: 'DEC';
 DECIMAL: 'DECIMAL';
 DECLARE: 'DECLARE';
 DEFAULT: 'DEFAULT';
+DEFAULT_PATH: 'DEFAULT_PATH';
 DEFINED: 'DEFINED';
 DEFINER: 'DEFINER';
 DELAY: 'DELAY';
@@ -385,6 +389,7 @@ OVERWRITE: 'OVERWRITE';
 PARTITION: 'PARTITION';
 PARTITIONED: 'PARTITIONED';
 PARTITIONS: 'PARTITIONS';
+PATH: 'PATH';
 PERCENTLIT: 'PERCENT';
 PIVOT: 'PIVOT';
 PLACING: 'PLACING';
@@ -470,6 +475,7 @@ SUBSTRING: 'SUBSTRING';
 SYNC: 'SYNC';
 SYSTEM_TIME: 'SYSTEM_TIME';
 SYSTEM_VERSION: 'SYSTEM_VERSION';
+SYSTEM_PATH: 'SYSTEM_PATH';
 TABLE: 'TABLE';
 TABLES: 'TABLES';
 TABLESAMPLE: 'TABLESAMPLE';

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

@@ -446,6 +446,7 @@ setResetStatement
     | SET TIME ZONE interval                                           #setTimeZone
     | SET TIME ZONE timezone                                           #setTimeZone
     | SET TIME ZONE .*?                                                #setTimeZone
+    | SET PATH EQ pathElement (COMMA pathElement)*                      #setPath
     | SET variable assignmentList                                      #setVariable
     | SET variable LEFT_PAREN multipartIdentifierList RIGHT_PAREN EQ
         LEFT_PAREN query RIGHT_PAREN                                   #setVariable
@@ -457,6 +458,15 @@ setResetStatement
     | RESET .*?                                                        #resetConfiguration
     ;
 
+pathElement
+    : DEFAULT_PATH
+    | SYSTEM_PATH
+    | PATH
+    | CURRENT_DATABASE (LEFT_PAREN RIGHT_PAREN)?
+    | CURRENT_SCHEMA (LEFT_PAREN RIGHT_PAREN)?
+    | multipartIdentifier
+    ;
+
 executeImmediate
     : EXECUTE IMMEDIATE queryParam=expression (INTO targetVariable=multipartIdentifierList)? executeImmediateUsing?
     ;
@@ -1276,7 +1286,7 @@ datetimeUnit
     ;
 
 primaryExpression
-    : name=(CURRENT_DATE | CURRENT_TIMESTAMP | CURRENT_USER | USER | SESSION_USER | CURRENT_TIME)             #currentLike
+    : name=(CURRENT_DATABASE | CURRENT_DATE | CURRENT_PATH | CURRENT_SCHEMA | CURRENT_TIME | CURRENT_TIMESTAMP | CURRENT_USER | USER | SESSION_USER) (LEFT_PAREN RIGHT_PAREN)?             #currentLike
     | name=(TIMESTAMPADD | DATEADD | DATE_ADD) LEFT_PAREN (unit=datetimeUnit | invalidUnit=stringLit) COMMA unitsAmount=valueExpression COMMA timestamp=valueExpression RIGHT_PAREN             #timestampadd
     | name=(TIMESTAMPDIFF | DATEDIFF | DATE_DIFF | TIMEDIFF) LEFT_PAREN (unit=datetimeUnit | invalidUnit=stringLit) COMMA startTimestamp=valueExpression COMMA endTimestamp=valueExpression RIGHT_PAREN    #timestampdiff
     | CASE whenClause+ (ELSE elseExpression=expression)? END                                   #searchedCase
@@ -1939,6 +1949,9 @@ ansiNonReserved
     | CURSOR
     | CUBE
     | CURRENT
+    | CURRENT_DATABASE
+    | CURRENT_PATH
+    | CURRENT_SCHEMA
     | DATA
     | DATABASE
     | DATABASES
@@ -1957,6 +1970,7 @@ ansiNonReserved
     | DEFAULT
     | DEFINED
     | DEFINER
+    | DEFAULT_PATH
     | DELAY
     | DELETE
     | DELIMITED
@@ -2088,6 +2102,7 @@ ansiNonReserved
     | PARTITION
     | PARTITIONED
     | PARTITIONS
+    | PATH
     | PERCENTLIT
     | PIVOT
     | PLACING
@@ -2163,6 +2178,7 @@ ansiNonReserved
     | SUBSTR
     | SUBSTRING
     | SYNC
+    | SYSTEM_PATH
     | SYSTEM_TIME
     | SYSTEM_VERSION
     | TABLES
@@ -2316,7 +2332,10 @@ nonReserved
     | CUBE
     | CURRENT
     | CURSOR
+    | CURRENT_DATABASE
     | CURRENT_DATE
+    | CURRENT_PATH
+    | CURRENT_SCHEMA
     | CURRENT_TIME
     | CURRENT_TIMESTAMP
     | CURRENT_USER
@@ -2336,6 +2355,7 @@ nonReserved
     | DECIMAL
     | DECLARE
     | DEFAULT
+    | DEFAULT_PATH
     | DEFINED
     | DEFINER
     | DELAY
@@ -2507,6 +2527,7 @@ nonReserved
     | PROCEDURES
     | PROPERTIES
     | PURGE
+    | PATH
     | QUARTER
     | QUERY
     | RANGE
@@ -2576,6 +2597,7 @@ nonReserved
     | SUBSTR
     | SUBSTRING
     | SYNC
+    | SYSTEM_PATH
     | SYSTEM_TIME
     | SYSTEM_VERSION
     | TABLE

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

@@ -290,13 +290,20 @@ object Analyzer {
  */
 class Analyzer(
     override val catalogManager: CatalogManager,
-    private[sql] val sharedRelationCache: RelationCache = RelationCache.empty)
+    private[sql] val sharedRelationCache: RelationCache = RelationCache.empty,
+    private[sql] val sessionConf: Option[SQLConf] = None)
   extends RuleExecutor[LogicalPlan]
   with CheckAnalysis with AliasHelper with SQLConfHelper with ColumnResolutionHelper {
 
+  /** Conf to use for path-based resolution and error messages; uses session conf when available. */
+  private[sql] def resolutionConf: SQLConf = sessionConf.getOrElse(SQLConf.get)
+
+  override protected def confForRoutineResolution: SQLConf = resolutionConf
+
   private val v1SessionCatalog: SessionCatalog = catalogManager.v1SessionCatalog
   private val relationResolution = new RelationResolution(catalogManager, sharedRelationCache)
-  private val functionResolution = new FunctionResolution(catalogManager, relationResolution)
+  private val functionResolution = new FunctionResolution(catalogManager, relationResolution,
+    resolutionConf)
 
   override protected def validatePlanChanges(
       previousPlan: LogicalPlan,
@@ -317,20 +324,22 @@ class Analyzer(
     if (plan.analyzed) {
       plan
     } else {
+      def runAnalysis(): LogicalPlan = HybridAnalyzer.fromLegacyAnalyzer(
+        legacyAnalyzer = this, tracker = tracker).apply(plan)
+      def runWithConf(): LogicalPlan = sessionConf match {
+        case Some(c) => SQLConf.withExistingConf(c) { runAnalysis() }
+        case None => runAnalysis()
+      }
       if (AnalysisContext.get.isDefault) {
         AnalysisContext.reset()
         try {
-          AnalysisHelper.markInAnalyzer {
-            HybridAnalyzer.fromLegacyAnalyzer(legacyAnalyzer = this, tracker = tracker).apply(plan)
-          }
+          AnalysisHelper.markInAnalyzer { runWithConf() }
         } finally {
           AnalysisContext.reset()
         }
       } else {
         AnalysisContext.withNewAnalysisContext {
-          AnalysisHelper.markInAnalyzer {
-            HybridAnalyzer.fromLegacyAnalyzer(legacyAnalyzer = this, tracker = tracker).apply(plan)
-          }
+          AnalysisHelper.markInAnalyzer { runWithConf() }
         }
       }
     }
@@ -2063,7 +2072,9 @@ class Analyzer(
                 case FunctionType.NotFound =>
                   val catalogPath =
                     catalogManager.currentCatalog.name +: catalogManager.currentNamespace
-                  val searchPath = SQLConf.get.resolutionSearchPath(catalogPath.toSeq)
+                  val pathEntries = resolutionConf.effectivePathEntries
+                    .getOrElse(Seq(catalogPath.toSeq))
+                  val searchPath = resolutionConf.resolutionSearchPath(pathEntries)
                     .map(_.quoted)
                   throw QueryCompilationErrors.unresolvedRoutineError(
                     nameParts,

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

@@ -43,6 +43,11 @@ trait CheckAnalysis extends LookupCatalog with QueryErrorsBase with PlanToString
 
   protected def isView(nameParts: Seq[String]): Boolean
 
+  protected def conf: org.apache.spark.sql.internal.SQLConf
+
+  /** Conf for routine resolution/errors; override in Analyzer to use session conf. */
+  protected def confForRoutineResolution: SQLConf = conf
+
   import org.apache.spark.sql.connector.catalog.CatalogV2Implicits._
 
   /**
@@ -304,7 +309,9 @@ trait CheckAnalysis extends LookupCatalog with QueryErrorsBase with PlanToString
 
       case u: UnresolvedFunctionName =>
         val catalogPath = currentCatalog.name +: catalogManager.currentNamespace
-        val searchPath = SQLConf.get.resolutionSearchPath(catalogPath.toSeq)
+        val pathEntries = confForRoutineResolution.effectivePathEntries
+          .getOrElse(Seq(catalogPath.toSeq))
+        val searchPath = confForRoutineResolution.resolutionSearchPath(pathEntries)
           .map(_.quoted)
         throw QueryCompilationErrors.unresolvedRoutineError(
           u.multipartIdentifier,

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

@@ -218,10 +218,15 @@ trait SimpleFunctionRegistryBase[T] extends FunctionRegistryBase[T] with Logging
   protected val functionBuilders =
     new mutable.HashMap[FunctionIdentifier, (ExpressionInfo, FunctionBuilder)]
 
-  // Resolution of the function name is always case insensitive, but the database name
-  // depends on the caller
-  private def normalizeFuncName(name: FunctionIdentifier): FunctionIdentifier = {
-    FunctionIdentifier(name.funcName.toLowerCase(Locale.ROOT), name.database)
+  // All function identifiers must be fully qualified (3-part: catalog.database.funcName).
+  // Normalization lowercases all parts for case-insensitive lookup.
+  protected def normalizeFuncName(name: FunctionIdentifier): FunctionIdentifier = {
+    assert(name.database.isDefined && name.catalog.isDefined,
+      s"Function identifier must be fully qualified (3-part): $name")
+    new FunctionIdentifier(
+      name.funcName.toLowerCase(Locale.ROOT),
+      name.database.map(_.toLowerCase(Locale.ROOT)),
+      name.catalog.map(_.toLowerCase(Locale.ROOT)))
   }
 
   override def registerFunction(
@@ -336,6 +341,16 @@ class SimpleFunctionRegistry
   }
 }
 
+/**
+ * A mixin for builtin-only registries that accepts any identifier and normalizes
+ * it to the builtin 3-part key (system.builtin.funcName). This allows callers to
+ * look up builtins by simple name without constructing a fully qualified identifier.
+ */
+trait BuiltinRegistryMixin[T] extends SimpleFunctionRegistryBase[T] {
+  override protected def normalizeFuncName(name: FunctionIdentifier): FunctionIdentifier =
+    super.normalizeFuncName(FunctionRegistry.builtinFunctionIdentifier(name.funcName))
+}
+
 object EmptyFunctionRegistry
     extends EmptyFunctionRegistryBase[Expression]
     with FunctionRegistry {
@@ -347,6 +362,13 @@ object FunctionRegistry {
 
   type FunctionBuilder = Seq[Expression] => Expression
 
+  /** Returns the 3-part identifier for a builtin function: system.builtin.funcName. */
+  private[sql] def builtinFunctionIdentifier(name: String): FunctionIdentifier =
+    new FunctionIdentifier(
+      name.toLowerCase(Locale.ROOT),
+      Some(CatalogManager.BUILTIN_NAMESPACE),
+      Some(CatalogManager.SYSTEM_CATALOG_NAME))
+
   val FUNC_ALIAS = TreeNodeTag[String]("functionAliasName")
 
   // ==============================================================================================
@@ -832,6 +854,7 @@ object FunctionRegistry {
     expression[CurrentDatabase]("current_database"),
     expression[CurrentDatabase]("current_schema", true, Some("3.4.0")),
     expression[CurrentCatalog]("current_catalog"),
+    expression[CurrentPath]("current_path", true, Some("4.2.0")),
     expression[CurrentUser]("current_user"),
     expression[CurrentUser]("user", true, Some("3.4.0")),
     expression[CurrentUser]("session_user", true, Some("4.0.0")),
@@ -1003,19 +1026,21 @@ object FunctionRegistry {
     expression[ToProtobuf]("to_protobuf")
   )
 
+  // BuiltinRegistryMixin normalizes any name to the builtin 3-part key (system.builtin.name).
   val builtin: SimpleFunctionRegistry = {
-    val fr = new SimpleFunctionRegistry
+    val fr = new SimpleFunctionRegistry with BuiltinRegistryMixin[Expression]
     expressions.foreach {
       case (name, (info, builder)) =>
-        fr.internalRegisterFunction(FunctionIdentifier(name), info, builder)
+        fr.registerFunction(FunctionIdentifier(name), info, builder)
     }
     fr
   }
 
   val functionSet: Set[FunctionIdentifier] = builtin.listFunction().toSet
 
   /** Registry for internal functions used by Connect and the Column API. */
-  private[sql] val internal: SimpleFunctionRegistry = new SimpleFunctionRegistry
+  private[sql] val internal: SimpleFunctionRegistry =
+    new SimpleFunctionRegistry with BuiltinRegistryMixin[Expression]
 
   private[spark] def registerInternalExpression[T <: Expression : ClassTag](
       name: String,
@@ -1030,7 +1055,8 @@ object FunctionRegistry {
     } else {
       builder
     }
-    internal.internalRegisterFunction(FunctionIdentifier(name), info, newBuilder)
+    // BuiltinRegistryMixin normalizes to the builtin 3-part key (system.builtin.name).
+    internal.registerFunction(FunctionIdentifier(name), info, newBuilder)
   }
 
   registerInternalExpression[Product]("product")
@@ -1301,11 +1327,12 @@ object TableFunctionRegistry {
     PythonWorkerLogs.functionBuilder
   )
 
+  // BuiltinRegistryMixin normalizes any name to the builtin 3-part key (system.builtin.name).
   val builtin: SimpleTableFunctionRegistry = {
-    val fr = new SimpleTableFunctionRegistry
+    val fr = new SimpleTableFunctionRegistry with BuiltinRegistryMixin[LogicalPlan]
     logicalPlans.foreach {
       case (name, (info, builder)) =>
-        fr.internalRegisterFunction(FunctionIdentifier(name), info, builder)
+        fr.registerFunction(FunctionIdentifier(name), info, builder)
     }
     fr
   }