[SPARK-59685][SQL] Keep clause-free SQL/JSON constructor canonical SQL bound to the built-in

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

@@ -687,6 +687,20 @@ object JsonValueBehavior {
  * }}}
  */
 // scalastyle:on line.size.limit
+
+/**
+ * A SQL/JSON built-in whose clause-free call can route through function resolution, letting a
+ * same-named routine on the SQL PATH shadow it. So when `sql` renders an otherwise clause-free
+ * form, it appends the default clause to keep the reparse bound to this built-in. `sqlString`'s
+ * `forceBuiltinOwnership` gates that clause (true for `sql`, false for the never-reparsed pretty
+ * form) and `renderChild` renders each child.
+ */
+trait RoutedSqlJsonExpression extends Expression {
+  private[sql] def sqlString(
+      forceBuiltinOwnership: Boolean,
+      renderChild: Expression => String): String
+}
+
 case class JsonValue(
     child: Expression,
     path: String,
@@ -701,7 +715,8 @@ case class JsonValue(
   with TimeZoneAwareExpression
   with CodegenFallback
   with ExpectsInputTypes
-  with QueryErrorsBase {
+  with QueryErrorsBase
+  with RoutedSqlJsonExpression {
 
   override def nullable: Boolean = true
 
@@ -825,7 +840,12 @@ case class JsonValue(
 
   override def prettyName: String = "json_value"
 
-  override def sql: String = {
+  override def sql: String = sqlString(forceBuiltinOwnership = true, _.sql)
+
+  // See [[RoutedSqlJsonExpression]] for `forceBuiltinOwnership` and `renderChild`.
+  private[sql] def sqlString(
+      forceBuiltinOwnership: Boolean,
+      renderChild: Expression => String): String = {
     // Reference identity, not value equality: an explicitly collated `RETURNING STRING COLLATE
     // UTF8_BINARY` is a distinct instance that compares `==` to the companion, so `==` would drop
     // it. Only the companion (by reference) renders nothing -- reached by omission and by a plain
@@ -835,7 +855,7 @@ case class JsonValue(
     def behaviorSQL(b: JsonValueBehavior, default: Option[Expression]): String = b match {
       case JsonValueBehavior.Null => "NULL"
       case JsonValueBehavior.Error => "ERROR"
-      case JsonValueBehavior.Default => s"DEFAULT ${default.get.sql}"
+      case JsonValueBehavior.Default => s"DEFAULT ${renderChild(default.get)}"
     }
     val emptySQL = if (onEmpty == JsonValueBehavior.Null) ""
       else s" ${behaviorSQL(onEmpty, emptyDefault)} ON EMPTY"
@@ -844,7 +864,12 @@ case class JsonValue(
     // Render the path as a properly escaped string literal so bracket-quoted paths such as
     // `$['a']` (and any path containing a quote or backslash) round-trip as valid SQL.
     val pathSQL = Literal(UTF8String.fromString(path), StringType).sql
-    s"JSON_VALUE(${child.sql}, $pathSQL$returningSQL$emptySQL$errorSQL)"
+    // Append the default `RETURNING STRING` when the render is otherwise clause-free.
+    val ownershipClause =
+      if (forceBuiltinOwnership && returningSQL.isEmpty && emptySQL.isEmpty && errorSQL.isEmpty) {
+        " RETURNING STRING"
+      } else ""
+    s"JSON_VALUE(${renderChild(child)}, $pathSQL$returningSQL$emptySQL$errorSQL$ownershipClause)"
   }
 
   override protected def withNewChildrenInternal(
@@ -918,7 +943,8 @@ case class JsonExists(
   extends UnaryExpression
   with CodegenFallback
   with ExpectsInputTypes
-  with QueryErrorsBase {
+  with QueryErrorsBase
+  with RoutedSqlJsonExpression {
 
   // The result is NULL only when the input is SQL NULL, or when `UNKNOWN ON ERROR` turns malformed
   // input into a BOOLEAN NULL. With a non-nullable input and any other ON ERROR behavior the result
@@ -979,14 +1005,21 @@ case class JsonExists(
 
   override def prettyName: String = "json_exists"
 
-  override def sql: String = {
+  override def sql: String = sqlString(forceBuiltinOwnership = true, _.sql)
+
+  // See [[RoutedSqlJsonExpression]] for `forceBuiltinOwnership` and `renderChild`.
+  private[sql] def sqlString(
+      forceBuiltinOwnership: Boolean,
+      renderChild: Expression => String): String = {
     val errorSQL = onError match {
       case JsonExistsBehavior.False => "" // the default
       case JsonExistsBehavior.True => " TRUE ON ERROR"
       case JsonExistsBehavior.Unknown => " UNKNOWN ON ERROR"
       case JsonExistsBehavior.Error => " ERROR ON ERROR"
     }
-    s"JSON_EXISTS(${child.sql}, ${toSQLValue(path)}$errorSQL)"
+    // Append the default `FALSE ON ERROR` when the render is otherwise clause-free.
+    val ownershipClause = if (forceBuiltinOwnership && errorSQL.isEmpty) " FALSE ON ERROR" else ""
+    s"JSON_EXISTS(${renderChild(child)}, ${toSQLValue(path)}$errorSQL$ownershipClause)"
   }
 
   override protected def withNewChildInternal(newChild: Expression): JsonExists =
@@ -1069,7 +1102,8 @@ case class JsonQuery(
   with CodegenFallback
   with ExpectsInputTypes
   with QueryErrorsBase
-  with ImplicitlyFormattedAsJson {
+  with ImplicitlyFormattedAsJson
+  with RoutedSqlJsonExpression {
 
   override def nullable: Boolean = true
 
@@ -1163,7 +1197,12 @@ case class JsonQuery(
 
   override def prettyName: String = "json_query"
 
-  override def sql: String = {
+  override def sql: String = sqlString(forceBuiltinOwnership = true, _.sql)
+
+  // See [[RoutedSqlJsonExpression]] for `forceBuiltinOwnership` and `renderChild`.
+  private[sql] def sqlString(
+      forceBuiltinOwnership: Boolean,
+      renderChild: Expression => String): String = {
     // Reference identity, not value equality, so an explicitly collated `RETURNING STRING COLLATE
     // UTF8_BINARY` (a distinct instance that compares `==` to the companion) is still rendered.
     // Only the companion (by reference) renders nothing -- reached by omission and by a plain
@@ -1191,7 +1230,16 @@ case class JsonQuery(
       if (onError == JsonQueryBehavior.Null) "" else s" ${behaviorSQL(onError)} ON ERROR"
     // Render the path as a properly escaped string literal so bracket-quoted paths round-trip.
     val pathSQL = Literal(UTF8String.fromString(path), StringType).sql
-    s"JSON_QUERY(${child.sql}, $pathSQL$returningSQL$wrapperSQL$quotesSQL$emptySQL$errorSQL)"
+    // Append the default `RETURNING STRING` when the render is otherwise clause-free.
+    val ownershipClause =
+      if (forceBuiltinOwnership && returningSQL.isEmpty && wrapperSQL.isEmpty &&
+          quotesSQL.isEmpty && emptySQL.isEmpty && errorSQL.isEmpty) {
+        " RETURNING STRING"
+      } else {
+        ""
+      }
+    s"JSON_QUERY(${renderChild(child)}, " +
+      s"$pathSQL$returningSQL$wrapperSQL$quotesSQL$emptySQL$errorSQL$ownershipClause)"
   }
 
   override protected def withNewChildInternal(newChild: Expression): JsonQuery =
@@ -1300,7 +1348,8 @@ case class JsonArray(
   with ExpectsInputTypes
   with QueryErrorsBase
   with DefaultStringProducingExpression
-  with ImplicitlyFormattedAsJson {
+  with ImplicitlyFormattedAsJson
+  with RoutedSqlJsonExpression {
 
   // `formatJson(i)` freezes whether element `i` is spliced raw (vs quoted); `needsValidation(i)`
   // freezes whether its raw text is arbitrary user input that must be JSON-validated at eval (an
@@ -1553,7 +1602,12 @@ case class JsonArray(
 
   override def prettyName: String = "json_array"
 
-  override def sql: String = {
+  override def sql: String = sqlString(forceBuiltinOwnership = true, _.sql)
+
+  // See [[RoutedSqlJsonExpression]] for `forceBuiltinOwnership` and `renderChild`.
+  private[sql] def sqlString(
+      forceBuiltinOwnership: Boolean,
+      renderChild: Expression => String): String = {
     val valuesSQL = values.zip(formatJson).map { case (v, isJson) =>
       // Emit SQL that reparses to the same splice/quote decision as the frozen `formatJson` flag.
       // The parser splices a value iff it is an explicit `FORMAT JSON` or a lexically-nested JSON
@@ -1577,11 +1631,11 @@ case class JsonArray(
       }
       val transitivelyImplicit = JsonArray.isImplicitlyJson(v)
       if (isJson && !directlyImplicit) {
-        s"${v.sql} FORMAT JSON"
+        s"${renderChild(v)} FORMAT JSON"
       } else if (!isJson && transitivelyImplicit) {
-        s"CAST(${v.sql} AS STRING)"
+        s"CAST(${renderChild(v)} AS STRING)"
       } else {
-        v.sql
+        renderChild(v)
       }
     }.mkString(", ")
     // Reference identity, not value equality: an explicitly collated `RETURNING STRING COLLATE
@@ -1594,7 +1648,15 @@ case class JsonArray(
       case JsonConstructorNullBehavior.Null => " NULL ON NULL"
       case JsonConstructorNullBehavior.Absent => ""
     }
-    s"JSON_ARRAY($valuesSQL$nullSQL$returningSQL)"
+    // Append the default `RETURNING STRING` when the render is otherwise clause-free. A spliced
+    // element (an explicit `FORMAT JSON`, or a nested constructor that reparses as implicit JSON)
+    // already forces the direct grammar branch, so it needs no clause.
+    val ownershipClause =
+      if (forceBuiltinOwnership && !formatJson.exists(identity) && nullSQL.isEmpty &&
+          returningSQL.isEmpty) {
+        " RETURNING STRING"
+      } else ""
+    s"JSON_ARRAY($valuesSQL$nullSQL$returningSQL$ownershipClause)"
   }
 
   override protected def withNewChildrenInternal(

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/package.scala

@@ -129,6 +129,16 @@ package object util extends Logging {
       )
     case c: Cast if !c.containsTag(Cast.USER_SPECIFIED_CAST) =>
       PrettyAttribute(usePrettyExpression(c.child, shouldTrimTempResolvedColumn).sql, c.dataType)
+    case j: RoutedSqlJsonExpression =>
+      // Column names are for display and never reparsed, so render the clean clause-free form, not
+      // the canonical `sql` that appends a round-trip-only default clause. Render children in place
+      // rather than substituting the node's children, so the raw-vs-quoted splice decisions (which
+      // key off each child's identity) are unchanged and no synthetic `FORMAT JSON` leaks in.
+      PrettyAttribute(
+        j.sqlString(
+          forceBuiltinOwnership = false,
+          usePrettyExpression(_, shouldTrimTempResolvedColumn).sql),
+        j.dataType)
     case p: PythonFuncExpression => PrettyPythonUDF(p.name, p.dataType, p.children)
     // Present a transpiled UDF exactly like the UDF it wraps, so auto-generated
     // column names stay `f(a)` whether or not transpilation engages (the node