[SPARK-52908][CORE] Prevent for iterator variable name clashing with names of labels in the path to the root of AST

### What changes were proposed in this pull request?

Proposed change is to explicitly prohibit the interaction of iterator variable hiding the scoped variable if the label of scope and the iterator variable names are the same.

### Why are the changes needed?

For iterator variable hides scoped variables if the label of the scope and iterator variable name are the same.

This interaction leads to undesirable behavior:

- Column of the iterator variable and a variable in scope having the same name will result in the column of the iterator variable hiding the variable in scope;
- Trying to access the variable in scope that does not clash with the column of the iterator variable will result in the compiler not being able to resolve the variable in scope.

### Does this PR introduce _any_ user-facing change?

Yes, it does. Changes are:

- Error LABEL_ALREADY_DEFINED was renamed to LABEL_OR_FOR_VARIABLE_ALREADY_DEFINED;
- Error LABEL_NAME_FORBIDDEN was renamed to LABEL_OR_FOR_VARIABLE_NAME_FORBIDDEN.

Old behavior:
![2C77E459-18FA-4B0E-8BDF-AA5D3E9B412F](https://github.com/user-attachments/assets/007ec190-e27f-4d70-b5a8-fb0272e05057)

<img width="1198" height="214" alt="image" src="https://github.com/user-attachments/assets/b2586697-49e9-4cbc-b57e-53b6c91700bc" />

New behavior:

<img width="1618" height="162" alt="image" src="https://github.com/user-attachments/assets/d023715c-08a1-47a2-9db1-3a19758140d6" />

<img width="1393" height="110" alt="image" src="https://github.com/user-attachments/assets/855d80c1-3fbd-42a6-ab1e-f664c4b4b47e" />

### How was this patch tested?

New tests in SqlScriptingExecutionSuite and existing tests.

Instead of printing a variable resolution exception, exception printed is stating the prohibition of such interactions.

Old behavior:
<img width="1335" height="380" alt="467960247-895da398-3ace-4334-b597-1be4a400acf4" src="https://github.com/user-attachments/assets/2d9412e1-5896-4286-b230-b728315a0fc6" />

New behavior:
<img width="1651" height="263" alt="467961070-92a0cbb7-bb93-410a-8266-3ef2591350f8" src="https://github.com/user-attachments/assets/5a44b505-6be2-4c1a-9f05-7fa563eab441" />

### Was this patch authored or co-authored using generative AI tooling?

No.

Closes #51595 from TeodorDjelic/prevent-for-iterator-variable-name-clashing-with-names-of-labels-in-the-path-to-the-root-of-ast.

Authored-by: Teodor Djelic <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>

Author: TeodorDjelic

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

@@ -4051,15 +4051,15 @@
     ],
     "sqlState" : "42K0L"
   },
-  "LABEL_ALREADY_EXISTS" : {
+  "LABEL_OR_FOR_VARIABLE_ALREADY_EXISTS" : {
     "message" : [
-      "The label <label> already exists. Choose another name or rename the existing label."
+      "The label or FOR variable <label> already exists. Choose another name or rename the existing one."
     ],
     "sqlState" : "42K0L"
   },
-  "LABEL_NAME_FORBIDDEN" : {
+  "LABEL_OR_FOR_VARIABLE_NAME_FORBIDDEN" : {
     "message" : [
-      "The label name <label> is forbidden."
+      "The label or FOR variable name <label> is forbidden."
     ],
     "sqlState" : "42K0L"
   },

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

@@ -26,7 +26,7 @@ import org.apache.spark.sql.AnalysisException
 import org.apache.spark.sql.catalyst.SqlScriptingContextManager
 import org.apache.spark.sql.catalyst.expressions._
 import org.apache.spark.sql.catalyst.expressions.SubExprUtils.wrapOuterReference
-import org.apache.spark.sql.catalyst.parser.SqlScriptingLabelContext.isForbiddenLabelName
+import org.apache.spark.sql.catalyst.parser.SqlScriptingLabelContext.isForbiddenLabelOrForVariableName
 import org.apache.spark.sql.catalyst.plans.logical._
 import org.apache.spark.sql.catalyst.trees.CurrentOrigin.withOrigin
 import org.apache.spark.sql.catalyst.trees.TreePattern._
@@ -269,7 +269,8 @@ trait ColumnResolutionHelper extends Logging with DataTypeErrorsBase {
       .filterNot(_ => AnalysisContext.get.isExecuteImmediate)
       // If variable name is qualified with session.<varName> treat it as a session variable.
       .filterNot(_ =>
-        nameParts.length > 2 || (nameParts.length == 2 && isForbiddenLabelName(nameParts.head)))
+        nameParts.length > 2
+          || (nameParts.length == 2 && isForbiddenLabelOrForVariableName(nameParts.head)))
       .flatMap(_.get(namePartsCaseAdjusted))
       .map { varDef =>
         VariableReference(

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala

@@ -557,13 +557,15 @@ class AstBuilder extends DataTypeAstBuilder
     val query = withOrigin(queryCtx) {
       SingleStatement(visitQuery(queryCtx))
     }
+    parsingCtx.labelContext.enterForScope(Option(ctx.multipartIdentifier()))
     val varName = Option(ctx.multipartIdentifier()).map(_.getText)
     val body = visitCompoundBodyImpl(
       ctx.compoundBody(),
       None,
       parsingCtx,
       isScope = false
     )
+    parsingCtx.labelContext.exitForScope(Option(ctx.multipartIdentifier()))
     parsingCtx.labelContext.exitLabeledScope(Option(ctx.beginLabel()))
 
     ForStatement(query, varName, body, Some(labelText))

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/ParserUtils.scala

@@ -28,7 +28,7 @@ import org.antlr.v4.runtime.tree.{ParseTree, TerminalNodeImpl}
 
 import org.apache.spark.SparkException
 import org.apache.spark.sql.catalyst.analysis.UnresolvedIdentifier
-import org.apache.spark.sql.catalyst.parser.SqlBaseParser.{BeginLabelContext, EndLabelContext}
+import org.apache.spark.sql.catalyst.parser.SqlBaseParser.{BeginLabelContext, EndLabelContext, MultipartIdentifierContext}
 import org.apache.spark.sql.catalyst.plans.logical.{CreateVariable, ErrorCondition}
 import org.apache.spark.sql.catalyst.trees.CurrentOrigin
 import org.apache.spark.sql.catalyst.util.SparkParserUtils
@@ -316,6 +316,23 @@ class SqlScriptingLabelContext {
     beginLabelCtx.map(_.multipartIdentifier().getText).isDefined
   }
 
+  /**
+   * Assert the identifier is not contained within seenLabels.
+   * If the identifier is contained within seenLabels, raise an exception.
+   */
+  private def assertIdentifierNotInSeenLabels(
+      identifierCtx: Option[MultipartIdentifierContext]): Unit = {
+    identifierCtx.foreach { ctx =>
+      val identifierName = ctx.getText
+      if (seenLabels.contains(identifierName.toLowerCase(Locale.ROOT))) {
+        withOrigin(ctx) {
+          throw SqlScriptingErrors
+            .duplicateLabels(CurrentOrigin.get, identifierName.toLowerCase(Locale.ROOT))
+        }
+      }
+    }
+  }
+
   /**
    * Enter a labeled scope and return the label text.
    * If the label is defined, it will be returned and added to seenLabels.
@@ -342,9 +359,9 @@ class SqlScriptingLabelContext {
       // Do not add the label to the seenLabels set if it is not defined.
       java.util.UUID.randomUUID.toString.toLowerCase(Locale.ROOT)
     }
-    if (SqlScriptingLabelContext.isForbiddenLabelName(labelText)) {
+    if (SqlScriptingLabelContext.isForbiddenLabelOrForVariableName(labelText)) {
       withOrigin(beginLabelCtx.get) {
-        throw SqlScriptingErrors.labelNameForbidden(CurrentOrigin.get, labelText)
+        throw SqlScriptingErrors.labelOrForVariableNameForbidden(CurrentOrigin.get, labelText)
       }
     }
     labelText
@@ -359,13 +376,46 @@ class SqlScriptingLabelContext {
       seenLabels.remove(beginLabelCtx.get.multipartIdentifier().getText.toLowerCase(Locale.ROOT))
     }
   }
+
+  /**
+   * Enter a for loop scope.
+   * If the for loop variable is defined, it will be asserted to not be inside seenLabels;
+   * Then, if the for loop variable is defined, it will be added to seenLabels.
+   */
+  def enterForScope(identifierCtx: Option[MultipartIdentifierContext]): Unit = {
+    identifierCtx.foreach { ctx =>
+      val identifierName = ctx.getText
+      assertIdentifierNotInSeenLabels(identifierCtx)
+      seenLabels.add(identifierName.toLowerCase(Locale.ROOT))
+
+      if (SqlScriptingLabelContext.isForbiddenLabelOrForVariableName(identifierName)) {
+        withOrigin(ctx) {
+          throw SqlScriptingErrors.labelOrForVariableNameForbidden(
+            CurrentOrigin.get,
+            identifierName.toLowerCase(Locale.ROOT))
+        }
+      }
+    }
+  }
+
+  /**
+   * Exit a for loop scope.
+   * If the for loop variable is defined, it will be removed from seenLabels.
+   */
+  def exitForScope(identifierCtx: Option[MultipartIdentifierContext]): Unit = {
+    identifierCtx.foreach { ctx =>
+      val identifierName = ctx.getText
+      seenLabels.remove(identifierName.toLowerCase(Locale.ROOT))
+    }
+  }
+
 }
 
 object SqlScriptingLabelContext {
   private val forbiddenLabelNames: immutable.Set[Regex] =
     immutable.Set("builtin".r, "session".r, "sys.*".r)
 
-  def isForbiddenLabelName(labelName: String): Boolean = {
+  def isForbiddenLabelOrForVariableName(labelName: String): Boolean = {
     forbiddenLabelNames.exists(_.matches(labelName.toLowerCase(Locale.ROOT)))
   }
 }

sql/catalyst/src/main/scala/org/apache/spark/sql/errors/SqlScriptingErrors.scala

@@ -33,7 +33,7 @@ private[sql] object SqlScriptingErrors {
   def duplicateLabels(origin: Origin, label: String): Throwable = {
     new SqlScriptingException(
       origin = origin,
-      errorClass = "LABEL_ALREADY_EXISTS",
+      errorClass = "LABEL_OR_FOR_VARIABLE_ALREADY_EXISTS",
       cause = null,
       messageParameters = Map("label" -> toSQLId(label)))
   }
@@ -54,10 +54,10 @@ private[sql] object SqlScriptingErrors {
       messageParameters = Map("endLabel" -> toSQLId(endLabel)))
   }
 
-  def labelNameForbidden(origin: Origin, label: String): Throwable = {
+  def labelOrForVariableNameForbidden(origin: Origin, label: String): Throwable = {
     new SqlScriptingException(
       origin = origin,
-      errorClass = "LABEL_NAME_FORBIDDEN",
+      errorClass = "LABEL_OR_FOR_VARIABLE_NAME_FORBIDDEN",
       cause = null,
       messageParameters = Map("label" -> toSQLId(label))
     )