[SPARK-54827][SQL] Add helper function TreeNode.containsTag

### What changes were proposed in this pull request?
Add helper function `TreeNode.containsTag`

### Why are the changes needed?
In many places, we don't care the tag value, we only need to check whether a tag exists.
This new function can help simplify the code a bit, e.g.

`getTagValue(Cast.BY_TABLE_INSERTION).isDefined` -> `containsTag(Cast.BY_TABLE_INSERTION)`

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

### How was this patch tested?
ci

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

Closes #53587 from zhengruifeng/containsTag.

Authored-by: Ruifeng Zheng <[email protected]>
Signed-off-by: Ruifeng Zheng <[email protected]>

Author: zhengruifeng

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

@@ -70,7 +70,7 @@ object AliasResolution {
   private def extractOnly(e: Expression): Boolean = e match {
     case _: ExtractValue => e.children.forall(extractOnly)
     case _: Literal => true
-    case attr: Attribute if attr.getTagValue(ResolverTag.SINGLE_PASS_IS_LCA).isEmpty => true
+    case attr: Attribute if !attr.containsTag(ResolverTag.SINGLE_PASS_IS_LCA) => true
     case _ => false
   }
 }

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

@@ -268,7 +268,7 @@ object ApplyDefaultCollationToStringType extends Rule[LogicalPlan] {
       newType => columnDef.copy(dataType = replaceDefaultStringType(columnDef.dataType, newType))
 
     case cast: Cast if hasDefaultStringType(cast.dataType) &&
-      cast.getTagValue(Cast.USER_SPECIFIED_CAST).isDefined =>
+      cast.containsTag(Cast.USER_SPECIFIED_CAST) =>
       newType => cast.copy(dataType = replaceDefaultStringType(cast.dataType, newType))
 
     case Literal(value, dt) if hasDefaultStringType(dt) =>

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

@@ -898,7 +898,7 @@ trait CheckAnalysis extends LookupCatalog with QueryErrorsBase with PlanToString
                 "invalidExprSqls" -> invalidExprSqls.mkString(", ")))
 
           case j @ LateralJoin(_, right, _, _)
-              if j.getTagValue(LateralJoin.BY_TABLE_ARGUMENT).isEmpty =>
+              if !j.containsTag(LateralJoin.BY_TABLE_ARGUMENT) =>
             right.plan.foreach {
               case Generate(pyudtf: PythonUDTF, _, _, _, _, _)
                   if pyudtf.evalType == PythonEvalType.SQL_ARROW_UDTF =>

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

@@ -36,7 +36,7 @@ object CollationTypeCoercion extends SQLConfHelper {
   private val COLLATION_CONTEXT_TAG = new TreeNodeTag[DataType]("collationContext")
 
   private def hasCollationContextTag(expr: Expression): Boolean = {
-    expr.getTagValue(COLLATION_CONTEXT_TAG).isDefined
+    expr.containsTag(COLLATION_CONTEXT_TAG)
   }
 
   def apply(expression: Expression): Expression = expression match {

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

@@ -140,8 +140,7 @@ trait ColumnResolutionHelper extends Logging with DataTypeErrorsBase {
           }
           matched(ordinal)
 
-        case u @ UnresolvedAttribute(nameParts)
-          if u.getTagValue(LogicalPlan.PLAN_ID_TAG).isEmpty =>
+        case u @ UnresolvedAttribute(nameParts) if !u.containsTag(LogicalPlan.PLAN_ID_TAG) =>
           // UnresolvedAttribute with PLAN_ID_TAG should be resolved in resolveDataFrameColumn
           val result = withPosition(u) {
             resolveColumnByName(nameParts)
@@ -451,7 +450,7 @@ trait ColumnResolutionHelper extends Logging with DataTypeErrorsBase {
       u: UnresolvedAttribute,
       q: LogicalPlan,
       includeLastResort: Boolean = false): Option[Expression] = {
-    assert(u.getTagValue(LogicalPlan.PLAN_ID_TAG).nonEmpty,
+    assert(u.containsTag(LogicalPlan.PLAN_ID_TAG),
       s"UnresolvedAttribute $u should have a Plan Id tag")
 
     resolveDataFrameColumn(u, q.children).map { r =>
@@ -524,7 +523,7 @@ trait ColumnResolutionHelper extends Logging with DataTypeErrorsBase {
     val planId = planIdOpt.get
     logDebug(s"Extract plan_id $planId from $u")
 
-    val isMetadataAccess = u.getTagValue(LogicalPlan.IS_METADATA_COL).nonEmpty
+    val isMetadataAccess = u.containsTag(LogicalPlan.IS_METADATA_COL)
 
     val (resolved, matched) = resolveDataFrameColumnByPlanId(
       u, planId, isMetadataAccess, q, 0)

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

@@ -36,7 +36,7 @@ class ResolveDataFrameDropColumns(val catalogManager: CatalogManager)
       //   df.drop(col("non-existing-column"))
       val dropped = d.dropList.flatMap {
         case u: UnresolvedAttribute =>
-          if (u.getTagValue(LogicalPlan.PLAN_ID_TAG).nonEmpty) {
+          if (u.containsTag(LogicalPlan.PLAN_ID_TAG)) {
             // Plan Id comes from Spark Connect,
             // Here we ignore the `UnresolvedAttribute` if its Plan Id can be found
             // but column not found.

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

@@ -92,8 +92,7 @@ object TypeCoercionValidation extends QueryErrorsBase {
       var issueFixedIfAnsiOff = true
       getAllExpressions(nonAnsiPlan).foreach(_.foreachUp {
         case e: Expression
-            if e.getTagValue(DATA_TYPE_MISMATCH_ERROR).isDefined &&
-            e.checkInputDataTypes().isFailure =>
+            if e.containsTag(DATA_TYPE_MISMATCH_ERROR) && e.checkInputDataTypes().isFailure =>
           e.checkInputDataTypes() match {
             case TypeCheckResult.TypeCheckFailure(_) | _: TypeCheckResult.DataTypeMismatch =>
               issueFixedIfAnsiOff = false

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

@@ -92,7 +92,7 @@ object DefaultCollationTypeCoercion {
    * we should change all its occurrences to [[StringType]] with default collation.
    */
   private def shouldApplyCollationToCast(cast: Cast): Boolean = {
-    cast.getTagValue(Cast.USER_SPECIFIED_CAST).isDefined &&
+    cast.containsTag(Cast.USER_SPECIFIED_CAST) &&
     hasDefaultStringType(cast.dataType)
   }
 }

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

@@ -208,7 +208,7 @@ class JoinResolver(resolver: Resolver, expressionResolver: ExpressionResolver)
       scopes.current.hiddenOutput.filter(_.qualifiedAccessOnly)
 
     val newProjectList =
-      if (unresolvedJoin.getTagValue(ResolverTag.TOP_LEVEL_OPERATOR).isEmpty) {
+      if (!unresolvedJoin.containsTag(ResolverTag.TOP_LEVEL_OPERATOR)) {
         newOutputList ++ qualifiedAccessOnlyColumnsFromHiddenOutput
       } else {
         newOutputList

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

@@ -328,7 +328,7 @@ class ResolverGuard(catalogManager: CatalogManager) extends SQLConfHelper {
 
   private def checkUnresolvedAttribute(unresolvedAttribute: UnresolvedAttribute) =
     !ResolverGuard.UNSUPPORTED_ATTRIBUTE_NAMES.contains(unresolvedAttribute.nameParts.head) &&
-    !unresolvedAttribute.getTagValue(LogicalPlan.PLAN_ID_TAG).isDefined
+    !unresolvedAttribute.containsTag(LogicalPlan.PLAN_ID_TAG)
 
   private def checkUnresolvedPredicate(unresolvedPredicate: Predicate) = unresolvedPredicate match {
     case inSubquery: InSubquery =>