Revert "[SPARK-56750] default path config"

This reverts commit 6cc6e8a.

Author: yaooqinn

.cursor/rules/apache-spark-scala-lint-before-commit.mdc

@@ -12,8 +12,6 @@ Before the first code edit or running test in a session, ensure a clean working
    - **New edits**: ask the user to choose: create a new git worktree from `<upstream>/master` and work from there (recommended), or create and switch to a new branch from `<upstream>/master`.
    - **Running tests**: use `<upstream>/master`.
 
-5. **Commits:** Before committing or pushing changes that touch Scala (or when unsure), run `./dev/lint-scala` from the repo root and fix all failures. Do not consider a change merge-ready until lint passes; CI runs the same checks.
-
 ## Development Notes
 
 SQL golden file tests are managed by `SQLQueryTestSuite` and its variants. Read the class documentation before running or updating these tests. DO NOT edit the generated golden files (`.sql.out`) directly. Always regenerate them when needed, and carefully review the diff to make sure it's expected.

AGENTS.md

@@ -217,10 +217,6 @@ singleTableSchema
     : colTypeList EOF
     ;
 
-singlePathElementList
-    : pathElement (COMMA pathElement)* EOF
-    ;
-
 singleRoutineParamList
     : colDefinitionList EOF
     ;

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

@@ -1976,15 +1976,14 @@ class Analyzer(
      * This is used for special syntax transformations (e.g., COUNT(*) -> COUNT(1)) that
      * should only apply to builtin functions, not to user-defined functions.
      *
-     * When the effective SQL PATH puts `system.session` before `system.builtin`, temp
-     * functions shadow builtins, so an unqualified name that matches a temp function
-     * should NOT be treated as builtin.
+     * In legacy mode (sessionOrder="first"), temp functions shadow builtins, so an
+     * unqualified name that matches a temp function should NOT be treated as builtin.
      */
     private def matchesFunctionName(nameParts: Seq[String], expectedName: String): Boolean = {
       if (!FunctionResolution.isUnqualifiedOrBuiltinFunctionName(nameParts, expectedName)) {
         return false
       }
-      if (nameParts.size == 1 && functionResolution.isSessionBeforeBuiltinInPath) {
+      if (nameParts.size == 1 && conf.sessionFunctionResolutionOrder == "first") {
         val v1Catalog = catalogManager.v1SessionCatalog
         !v1Catalog.isTemporaryFunction(FunctionIdentifier(nameParts.head))
       } else {

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

@@ -76,22 +76,6 @@ class FunctionResolution(
     nameParts.length == 3 &&
       nameParts.head.equalsIgnoreCase(CatalogManager.SYSTEM_CATALOG_NAME)
 
-  /**
-   * True iff `system.session` is searched before `system.builtin` in the effective SQL PATH.
-   *
-   * Drives the `count(*) -> count(1)` rewrite (which must skip transformation when a temp
-   * `count` shadows the builtin) and the `SessionCatalog` security check that blocks creating
-   * a temp function with a builtin's name. Reads the live PATH via `CatalogManager` and
-   * applies the same kinds extraction that drives `SessionCatalog`'s fast-path provider, so
-   * the predicate stays in sync with the lookup loop's actual order.
-   */
-  def isSessionBeforeBuiltinInPath: Boolean = {
-    val path = catalogManager.sqlResolutionPathEntries(
-      catalogManager.currentCatalog.name(), catalogManager.currentNamespace.toSeq)
-    CatalogManager.systemFunctionKindsFromPath(path).headOption
-      .contains(org.apache.spark.sql.catalyst.catalog.SessionCatalog.Temp)
-  }
-
   /**
    * Produces the ordered list of candidate names for resolution. Expansion happens in two cases:
    *

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

@@ -20,20 +20,17 @@ package org.apache.spark.sql.catalyst.analysis.resolver
 import scala.util.Random
 
 import org.apache.spark.sql.AnalysisException
-import org.apache.spark.sql.catalyst.FunctionIdentifier
 import org.apache.spark.sql.catalyst.analysis.{
   FunctionResolution,
   UnresolvedFunction,
-  UnresolvedSeed,
-  UnresolvedStar
+  UnresolvedSeed
 }
 import org.apache.spark.sql.catalyst.expressions.{
   BinaryArithmetic,
   Collate,
   Expression,
   ExpressionWithRandomSeed,
   InheritAnalysisRules,
-  Literal,
   ResolvedCollation,
   TryEval,
   UnresolvedCollation,
@@ -63,7 +60,7 @@ import org.apache.spark.sql.catalyst.util.CollationFactory
  */
 class FunctionResolver(
     expressionResolver: ExpressionResolver,
-    protected val functionResolution: FunctionResolution,
+    functionResolution: FunctionResolution,
     aggregateExpressionResolver: AggregateExpressionResolver,
     binaryArithmeticResolver: BinaryArithmeticResolver)
     extends TreeNodeResolver[UnresolvedFunction, Expression]
@@ -120,10 +117,9 @@ class FunctionResolver(
    *  - Apply timezone, if the resulting expression is [[TimeZoneAwareExpression]].
    */
   override def resolve(unresolvedFunction: UnresolvedFunction): Expression = {
-    val effectiveUnresolved = restoreCountStarAfterParserForTempShadow(unresolvedFunction)
     val expressionInfo = functionResolution.lookupBuiltinOrTempFunction(
-      nameParts = effectiveUnresolved.nameParts,
-      unresolvedFunc = Some(effectiveUnresolved)
+      nameParts = unresolvedFunction.nameParts,
+      unresolvedFunc = Some(unresolvedFunction)
     )
     val expressionResolutionContext = expressionResolutionContextStack.peek()
 
@@ -137,7 +133,7 @@ class FunctionResolver(
 
     val functionWithResolvedChildren =
       withResolvedChildren(
-        unresolvedExpression = effectiveUnresolved,
+        unresolvedExpression = unresolvedFunction,
         resolveChild = expressionResolver.resolve _
       ).asInstanceOf[UnresolvedFunction]
 
@@ -149,33 +145,6 @@ class FunctionResolver(
     }
   }
 
-  /**
-   * SQL parsing turns `COUNT(*)` into `COUNT(1)` before analysis ([[AstBuilder]]). When PATH
-   * orders session before builtin and a temporary `count` shadows the builtin, that rewrite must
-   * be undone so arguments expand from the child operator output (aligning with fixed-point
-   * [[org.apache.spark.sql.catalyst.analysis.Analyzer.ResolveReferences]] / `matchesFunctionName`).
-   */
-  private def restoreCountStarAfterParserForTempShadow(
-      unresolvedFunction: UnresolvedFunction): UnresolvedFunction = {
-    if (unresolvedFunction.nameParts.length != 1 ||
-      unresolvedFunction.isDistinct ||
-      !unresolvedFunction.nameParts.head.equalsIgnoreCase("count")) {
-      return unresolvedFunction
-    }
-    unresolvedFunction.arguments match {
-      case Seq(lit: Literal)
-          if lit.value != null &&
-            lit.value == 1 &&
-            functionResolution.isSessionBeforeBuiltinInPath &&
-            functionResolution.catalogManager.v1SessionCatalog.isTemporaryFunction(
-              FunctionIdentifier(unresolvedFunction.nameParts.head)) =>
-        val expanded = expressionResolver.expandStarExpressions(Seq(UnresolvedStar(None)))
-        unresolvedFunction.copy(arguments = expanded)
-      case _ =>
-        unresolvedFunction
-    }
-  }
-
   private def handlePartiallyResolvedFunction(partiallyResolvedFunction: Expression): Expression = {
     val expressionResolutionContext = expressionResolutionContextStack.peek()
     val resolvedFunction = partiallyResolvedFunction match {

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/FunctionResolver.scala

@@ -19,9 +19,7 @@ package org.apache.spark.sql.catalyst.analysis.resolver
 
 import java.util.Locale
 
-import org.apache.spark.sql.catalyst.FunctionIdentifier
 import org.apache.spark.sql.catalyst.analysis.{
-  FunctionResolution,
   ResolvedStar,
   Star,
   UnresolvedFunction,
@@ -37,7 +35,6 @@ import org.apache.spark.sql.internal.SQLConf
  */
 trait FunctionResolverUtils {
   protected def expressionResolver: ExpressionResolver
-  protected def functionResolution: FunctionResolution
   protected def conf: SQLConf
 
   private val scopes = expressionResolver.getNameScopes
@@ -102,21 +99,7 @@ trait FunctionResolverUtils {
       unresolvedFunction: UnresolvedFunction,
       normalizeFunctionName: Boolean = true
   ): Boolean = {
-    !unresolvedFunction.isDistinct &&
-      isCount(unresolvedFunction, normalizeFunctionName) &&
-      !isUnqualifiedCountShadowedByTemp(unresolvedFunction)
-  }
-
-  /**
-   * Keep single-pass behavior aligned with fixed-point: when PATH puts system.session before
-   * system.builtin and a temp `count` exists, unqualified `count(*)` must not be rewritten to
-   * `count(1)`.
-   */
-  private def isUnqualifiedCountShadowedByTemp(unresolvedFunction: UnresolvedFunction): Boolean = {
-    unresolvedFunction.nameParts.length == 1 &&
-      functionResolution.isSessionBeforeBuiltinInPath &&
-      functionResolution.catalogManager.v1SessionCatalog
-        .isTemporaryFunction(FunctionIdentifier(unresolvedFunction.nameParts.head))
+    !unresolvedFunction.isDistinct && isCount(unresolvedFunction, normalizeFunctionName)
   }
 
   private def isCount(

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/FunctionResolverUtils.scala

@@ -33,7 +33,7 @@ import org.apache.spark.sql.errors.QueryCompilationErrors
  */
 class HigherOrderFunctionResolver(
     protected val expressionResolver: ExpressionResolver,
-    protected val functionResolution: FunctionResolution)
+    functionResolution: FunctionResolution)
     extends TreeNodeResolver[UnresolvedFunction, Expression]
     with ProducesUnresolvedSubtree
     with CoercesExpressionTypes

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/HigherOrderFunctionResolver.scala

@@ -107,9 +107,6 @@ class ResolutionValidator {
         validateRepartitionByExpression(repartitionByExpression)
       case sample: Sample =>
         validateSample(sample)
-      case deserializeToObject: DeserializeToObject =>
-        validate(deserializeToObject.child)
-        handleOperatorOutput(deserializeToObject)
       case generate: Generate =>
         validateGenerate(generate)
       case expand: Expand =>
@@ -121,9 +118,6 @@ class ResolutionValidator {
         validateRelation(multiInstanceRelation)
       case supervisingCommand: SupervisingCommand =>
         validateSupervisingCommand(supervisingCommand)
-      case cmd: Command if cmd.children.isEmpty =>
-        // Session side-effect commands (e.g. SET PATH) are leaves with no query subtree.
-        ()
     }
 
     operator match {

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/ResolutionValidator.scala

@@ -136,15 +136,6 @@ class Resolver(
    */
   private val planRewriter = new PlanRewriter(planRewriteRules, extendedRewriteRules)
 
-  /**
-   * Runs session post-resolution rules (see [[extendedRewriteRules]]) on a subtree. Used for
-   * operators such as [[DeserializeToObject]] whose deserializer is resolved by the same rules as
-   * the end-of-resolution [[PlanRewriter]] batch, but which must be applied before single-pass
-   * validation asserts the operator is fully resolved.
-   */
-  private def applyExtendedRewriteRulesLocally(plan: LogicalPlan): LogicalPlan =
-    extendedRewriteRules.foldLeft(plan) { case (p, rule) => rule(p) }
-
   /**
    * [[relationMetadataProvider]] is used to resolve metadata for relations. It's initialized with
    * the default implementation [[MetadataResolver]] here and is called in
@@ -334,9 +325,6 @@ class Resolver(
             resolveRepartition(repartition)
           case sample: Sample =>
             resolveSample(sample)
-          case deserializeToObject: DeserializeToObject =>
-            applyExtendedRewriteRulesLocally(
-              deserializeToObject.copy(child = resolve(deserializeToObject.child)))
           case _ =>
             tryDelegateResolutionToExtension(unresolvedPlan).getOrElse {
               handleUnmatchedOperator(unresolvedPlan)