[SPARK-59144][SQL] Route plain SQL/JSON constructor calls through routine resolution

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

@@ -2105,21 +2105,11 @@ 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.
+     * Mirrors function resolution precedence, including SQL PATH shadowing for unqualified names
+     * and `spark.sql.legacy.persistentCatalogFirst` for two-part `builtin.name` references.
      */
-    private def matchesFunctionName(nameParts: Seq[String], expectedName: String): Boolean = {
-      if (!FunctionResolution.isUnqualifiedOrBuiltinFunctionName(nameParts, expectedName)) {
-        return false
-      }
-      if (nameParts.size == 1 && functionResolution.isSessionBeforeBuiltinInPath) {
-        val v1Catalog = catalogManager.v1SessionCatalog
-        !v1Catalog.isTemporaryFunction(FunctionIdentifier(nameParts.head))
-      } else {
-        true
-      }
-    }
+    private def matchesFunctionName(nameParts: Seq[String], expectedName: String): Boolean =
+      functionResolution.functionNameResolvesToBuiltin(nameParts, expectedName)
 
     /**
      * Expands the matching attribute.*'s in `child`'s output.
@@ -2135,6 +2125,13 @@ class Analyzer(
           // the same qualification.
           f0.copy(arguments = Seq(Literal(1)))
         case f1: UnresolvedFunction if containsStar(f1.arguments) =>
+          // A routed SQL/JSON constructor (json_array(*)) forbids a bare `*`; reject it rather than
+          // expand below. A nested star (json_array(array(*))) is expanded bottom-up before we get
+          // here, and count(*) is rewritten above -- so only a bare `*` reaches this guard.
+          if (functionResolution.resolvesToStarDisallowedJsonConstructor(f1.nameParts)) {
+            throw QueryCompilationErrors.invalidStarUsageError(
+              s"expression `${f1.prettyName}`", extractStar(f1.arguments))
+          }
           // SPECIAL CASE: We want to block count(tblName.*) because in spark, count(tblName.*) will
           // be expanded while count(*) will be converted to count(1). They will produce different
           // results and confuse users if there are any null values. For count(t1.*, t2.*), it is

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

@@ -1025,7 +1025,13 @@ object FunctionRegistry {
     expression[SchemaOfJson]("schema_of_json"),
     expression[LengthOfJsonArray]("json_array_length"),
     expression[JsonObjectKeys]("json_object_keys"),
-    expression[JsonTypeof]("json_typeof")
+    expression[JsonTypeof]("json_typeof"),
+    // Built-in forms of the SQL:2016 JSON constructor and path functions, resolved for plain calls
+    // that carry no SQL/JSON clauses (the dedicated grammar handles clause-bearing syntax).
+    expressionBuilder("json_value", JsonValueExpressionBuilder),
+    expressionBuilder("json_query", JsonQueryExpressionBuilder),
+    expressionBuilder("json_exists", JsonExistsExpressionBuilder),
+    expressionBuilder("json_array", JsonArrayExpressionBuilder)
   )
 
   private def variantExpressions: Seq[FunctionRegistryEntry] = Seq(

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/FunctionResolution.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. Uses the consolidated
-   * snapshot helper (SPARK-56939) so the (catalog, namespace, path) triple is observed
-   * atomically.
-   */
-  def isSessionBeforeBuiltinInPath: Boolean = {
-    catalogManager.sessionFunctionKindsForUnqualifiedResolution().headOption
-      .contains(org.apache.spark.sql.catalyst.catalog.SessionCatalog.Temp)
-  }
-
   /**
    * Produces the ordered list of candidate names for resolution. Expansion happens in two cases:
    *
@@ -404,6 +388,87 @@ class FunctionResolution(
     }
   }
 
+  /**
+   * Returns whether an unqualified function name reaches `system.builtin` before any temp or
+   * persistent function in the effective SQL PATH. When a temp or persistent function shadows the
+   * builtin, special-syntax rewrites (e.g. `count(*) -> count(1)`) must not fire, since the name no
+   * longer refers to Spark's builtin.
+   */
+  def unqualifiedFunctionResolvesToBuiltinBeforeAnyShadow(functionName: String): Boolean = {
+    // Walk the PATH in order and stop at the first entry that owns the name. The default order puts
+    // system.builtin first, so the common case returns on the first entry with no catalog lookup;
+    // only a custom PATH that lists a persistent catalog ahead of system.builtin reaches the probe
+    // below (one lookup per such preceding entry, recomputed on each call -- not cached).
+    sqlResolutionPathEntriesForAnalysis.foreach { pathEntry =>
+      val candidate = pathEntry :+ functionName
+      FunctionResolution.sessionNamespaceKind(candidate) match {
+        case Some(org.apache.spark.sql.catalyst.catalog.SessionCatalog.Builtin) =>
+          return true
+        case Some(org.apache.spark.sql.catalyst.catalog.SessionCatalog.Temp) =>
+          if (v1SessionCatalog.isTemporaryFunction(FunctionIdentifier(functionName))) {
+            return false
+          }
+        case None =>
+          if (persistentFunctionExists(candidate)) {
+            return false
+          }
+      }
+    }
+    false
+  }
+
+  /**
+   * Returns true when a function reference resolves to the system built-in with the requested name.
+   * This mirrors [[resolveFunction]] for special parser/analyzer rewrites that must run only for
+   * Spark's built-ins. In particular, two-part `builtin.name` is not always a system built-in:
+   * with `spark.sql.legacy.persistentCatalogFirst=true`, an existing persistent
+   * `current_catalog.builtin.name` takes precedence.
+   */
+  def functionNameResolvesToBuiltin(nameParts: Seq[String], expectedName: String): Boolean = {
+    if (!FunctionRegistry.functionSet.contains(
+          FunctionRegistry.builtinFunctionIdentifier(expectedName)) ||
+        !FunctionResolution.isUnqualifiedOrBuiltinFunctionName(nameParts, expectedName)) {
+      return false
+    }
+    nameParts.length match {
+      case 1 =>
+        unqualifiedFunctionResolvesToBuiltinBeforeAnyShadow(nameParts.head)
+      case 2 =>
+        conf.prioritizeSystemCatalog || !persistentFunctionExists(nameParts)
+      case 3 =>
+        true
+      case _ =>
+        false
+    }
+  }
+
+  private val starDisallowedJsonConstructors =
+    Set("json_array", "json_exists", "json_query", "json_value")
+
+  /** True if `nameParts` resolves to a built-in SQL/JSON constructor that forbids bare `*`. */
+  def resolvesToStarDisallowedJsonConstructor(nameParts: Seq[String]): Boolean =
+    starDisallowedJsonConstructors.exists(functionNameResolvesToBuiltin(nameParts, _))
+
+  private def persistentFunctionExists(nameParts: Seq[String]): Boolean = {
+    try {
+      // Expand through the view's frozen catalog/namespace exactly as `resolveFunctionCandidate`
+      // does, so the shadow probe queries the same catalog the real resolver would inside a view.
+      relationResolution.expandIdentifier(nameParts) match {
+        case CatalogAndIdentifier(catalog, ident) =>
+          catalog.asFunctionCatalog.functionExists(ident)
+        case _ =>
+          false
+      }
+    } catch {
+      case _: NoSuchFunctionException
+         | _: NoSuchNamespaceException
+         | _: CatalogNotFoundException =>
+        false
+      case e: AnalysisException if e.getCondition == "FORBIDDEN_OPERATION" =>
+        false
+    }
+  }
+
   /**
    * Determines the type/location of a function (builtin, temporary, persistent, etc.).
    * This is used by the LookupFunctions analyzer rule for early validation and optimization.

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

@@ -17,15 +17,14 @@
 
 package org.apache.spark.sql.catalyst.analysis.resolver
 
-import org.apache.spark.sql.catalyst.FunctionIdentifier
 import org.apache.spark.sql.catalyst.analysis.{
   FunctionResolution,
   ResolvedStar,
   Star,
   UnresolvedFunction,
   UnresolvedStar
 }
-import org.apache.spark.sql.catalyst.expressions.Literal
+import org.apache.spark.sql.catalyst.expressions.{Expression, Literal}
 import org.apache.spark.sql.errors.QueryCompilationErrors
 import org.apache.spark.sql.internal.SQLConf
 
@@ -57,12 +56,19 @@ trait FunctionResolverUtils {
    */
   protected def handleStarInArguments(
       unresolvedFunction: UnresolvedFunction): UnresolvedFunction = {
-    val functionContainsStarInArguments = unresolvedFunction.arguments.exists {
+    val functionContainsDirectStarInArguments = unresolvedFunction.arguments.exists {
       case _: Star => true
       case _ => false
     }
 
-    if (!functionContainsStarInArguments) {
+    if (functionContainsDirectStarInArguments &&
+        functionResolution.resolvesToStarDisallowedJsonConstructor(unresolvedFunction.nameParts)) {
+      // Only a bare `*` argument is rejected in a JSON constructor; a star nested in another
+      // expression (json_array(array(*))) is expanded there and count(*) is rewritten to count(1),
+      // so both stay valid arguments.
+      throw QueryCompilationErrors.invalidStarUsageError(
+        s"expression `${unresolvedFunction.prettyName}`", extractStar(unresolvedFunction.arguments))
+    } else if (!functionContainsDirectStarInArguments) {
       unresolvedFunction
     } else if (isNonDistinctCount(unresolvedFunction) &&
       hasSingleSimpleStarArgument(unresolvedFunction)) {
@@ -92,30 +98,16 @@ trait FunctionResolverUtils {
       case _ => false
     }
 
+  private def extractStar(expressions: Seq[Expression]): Seq[Star] =
+    expressions.collect { case s: Star => s }
+
   /**
    * Method used to determine whether the given function is non-distinct `count` function,
    * with optional normalization.
    */
   private def isNonDistinctCount(unresolvedFunction: UnresolvedFunction): Boolean = {
     !unresolvedFunction.isDistinct &&
-      isCount(unresolvedFunction) &&
-      !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))
-  }
-
-  private def isCount(unresolvedFunction: UnresolvedFunction): Boolean = {
-    FunctionResolution.isUnqualifiedOrBuiltinFunctionName(unresolvedFunction.nameParts, "count")
+      functionResolution.functionNameResolvesToBuiltin(unresolvedFunction.nameParts, "count")
   }
 
   /**
@@ -127,7 +119,6 @@ trait FunctionResolverUtils {
   private def normalizeCountExpression(
       unresolvedFunction: UnresolvedFunction): UnresolvedFunction = {
     unresolvedFunction.copy(
-      nameParts = Seq("count"),
       arguments = Seq(Literal(1)),
       filter = unresolvedFunction.filter
     )
@@ -141,7 +132,7 @@ trait FunctionResolverUtils {
   private def assertSingleTableStarNotInCountFunction(
       unresolvedFunction: UnresolvedFunction): Unit = {
     if (!conf.allowStarWithSingleTableIdentifierInCount &&
-      isCount(unresolvedFunction) &&
+      functionResolution.functionNameResolvesToBuiltin(unresolvedFunction.nameParts, "count") &&
       unresolvedFunction.arguments.length == 1) {
       unresolvedFunction.arguments.head match {
         case star: UnresolvedStar if scopes.current.isStarQualifiedByTable(star) =>

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

@@ -633,7 +633,8 @@ class ResolverGuard(
         true
       // JSON
       case _: GetJsonObject | _: JsonTuple | _: JsonToStructs | _: StructsToJson |
-          _: SchemaOfJson | _: JsonObjectKeys | _: LengthOfJsonArray =>
+          _: SchemaOfJson | _: JsonObjectKeys | _: LengthOfJsonArray | _: JsonArray |
+          _: JsonExists | _: JsonQuery | _: JsonValue =>
         true
       // CSV
       case _: SchemaOfCsv | _: StructsToCsv | _: CsvToStructs =>