remove internal tags

core/src/main/scala/org/apache/spark/SparkContext.scala

@@ -912,18 +912,11 @@ class SparkContext(config: SparkConf) extends Logging {
    */
   def addJobTag(tag: String): Unit = {
     SparkContext.throwIfInvalidTag(tag)
-    val existingTags = getJobTags() ++ getInternalJobTags()
+    val existingTags = getJobTags()
     val newTags = (existingTags + tag).mkString(SparkContext.SPARK_JOB_TAGS_SEP)
     setLocalProperty(SparkContext.SPARK_JOB_TAGS, newTags)
   }
 
-  /**
-   * Add a tag to be assigned to all the jobs started by this thread. The tag will be prefixed with
-   * an internal prefix to avoid conflicts with user tags.
-   */
-  private[spark] def addInternalJobTag(tag: String): Unit =
-    addJobTag(s"${SparkContext.SPARK_JOB_TAGS_INTERNAL_PREFIX}$tag")
-
   /**
    * Remove a tag previously added to be assigned to all the jobs started by this thread.
    * Noop if such a tag was not added earlier.
@@ -934,7 +927,7 @@ class SparkContext(config: SparkConf) extends Logging {
    */
   def removeJobTag(tag: String): Unit = {
     SparkContext.throwIfInvalidTag(tag)
-    val existingTags = getJobTags() ++ getInternalJobTags()
+    val existingTags = getJobTags()
     val newTags = (existingTags - tag).mkString(SparkContext.SPARK_JOB_TAGS_SEP)
     if (newTags.isEmpty) {
       clearJobTags()
@@ -943,22 +936,6 @@ class SparkContext(config: SparkConf) extends Logging {
     }
   }
 
-  /**
-   * Remove an internal tag previously added to be assigned to all the jobs started by this thread.
-   */
-  private[spark] def removeInternalJobTag(tag: String): Unit =
-    removeJobTag(s"${SparkContext.SPARK_JOB_TAGS_INTERNAL_PREFIX}$tag")
-
-  /**
-   * Get the tags that are currently set to be assigned to all the jobs started by this thread.
-   */
-  private[spark] def getInternalJobTags(): Set[String] = {
-    Option(getLocalProperty(SparkContext.SPARK_JOB_TAGS))
-      .map(_.split(SparkContext.SPARK_JOB_TAGS_SEP).toSet)
-      .getOrElse(Set())
-      .filter(_.startsWith(SparkContext.SPARK_JOB_TAGS_INTERNAL_PREFIX)) // only internal tags
-  }
-
   /**
    * Get the tags that are currently set to be assigned to all the jobs started by this thread.
    *
@@ -968,8 +945,7 @@ class SparkContext(config: SparkConf) extends Logging {
     Option(getLocalProperty(SparkContext.SPARK_JOB_TAGS))
       .map(_.split(SparkContext.SPARK_JOB_TAGS_SEP).toSet)
       .getOrElse(Set())
-      .filterNot(_.startsWith(SparkContext.SPARK_JOB_TAGS_INTERNAL_PREFIX)) // exclude internal tags
-      .filter(_.nonEmpty) // empty string tag should not happen, but be defensive
+      .filter(!_.isEmpty) // empty string tag should not happen, but be defensive
   }
 
   /**
@@ -978,14 +954,7 @@ class SparkContext(config: SparkConf) extends Logging {
    * @since 3.5.0
    */
   def clearJobTags(): Unit = {
-    val internalTags = getInternalJobTags() // exclude internal tags
-    if (internalTags.isEmpty) {
-      setLocalProperty(SparkContext.SPARK_JOB_TAGS, null)
-    } else {
-      setLocalProperty(
-        SparkContext.SPARK_JOB_TAGS,
-        internalTags.mkString(SparkContext.SPARK_JOB_TAGS_SEP))
-    }
+    setLocalProperty(SparkContext.SPARK_JOB_TAGS, null)
   }
 
   /**
@@ -3148,9 +3117,6 @@ object SparkContext extends Logging {
   /** Separator of tags in SPARK_JOB_TAGS property */
   private[spark] val SPARK_JOB_TAGS_SEP = ","
 
-  /** Prefix to mark a tag to be visible internally, not by users */
-  private[spark] val SPARK_JOB_TAGS_INTERNAL_PREFIX = "~~spark~internal~tag~~"
-
   // Same rules apply to Spark Connect execution tags, see ExecuteHolder.throwIfInvalidTag
   private[spark] def throwIfInvalidTag(tag: String) = {
     if (tag == null) {

sql/core/src/main/scala/org/apache/spark/sql/SparkSession.scala

@@ -885,7 +885,7 @@ class SparkSession private(
    * @since 4.0.0
    */
   def interruptAll(): Seq[String] =
-    doInterruptTag(sessionJobTag, "as part of cancellation of all jobs", tagIsInternal = true)
+    doInterruptTag(sessionJobTag, "as part of cancellation of all jobs")
 
   /**
    * Request to interrupt all currently running operations of this session with the given job tag.
@@ -897,16 +897,12 @@ class SparkSession private(
   def interruptTag(tag: String): Seq[String] = {
     val realTag = managedJobTags.get(tag)
     if (realTag == null) return Seq.empty
-    doInterruptTag(realTag, s"part of cancelled job tags $tag", tagIsInternal = false)
+    doInterruptTag(realTag, s"part of cancelled job tags $tag")
   }
 
-  private def doInterruptTag(
-      tag: String,
-      reason: String,
-      tagIsInternal: Boolean): Seq[String] = {
-    val realTag = if (tagIsInternal) s"${SparkContext.SPARK_JOB_TAGS_INTERNAL_PREFIX}$tag" else tag
+  private def doInterruptTag(tag: String, reason: String): Seq[String] = {
     val cancelledTags =
-      sparkContext.cancelJobsWithTagWithFuture(realTag, reason)
+      sparkContext.cancelJobsWithTagWithFuture(tag, reason)
 
     ThreadUtils.awaitResult(cancelledTags, 60.seconds)
       .flatMap(job => Option(job.properties.getProperty(SQLExecution.EXECUTION_ROOT_ID_KEY)))
@@ -926,7 +922,7 @@ class SparkSession private(
     scala.util.Try(executionId.toLong).toOption match {
       case Some(executionIdToBeCancelled) =>
         val tagToBeCancelled = SQLExecution.executionIdJobTag(this, executionIdToBeCancelled)
-        doInterruptTag(tagToBeCancelled, reason = "", tagIsInternal = true)
+        doInterruptTag(tagToBeCancelled, reason = "")
       case None =>
         throw new IllegalArgumentException("executionId must be a number in string form.")
     }

sql/core/src/main/scala/org/apache/spark/sql/execution/SQLExecution.scala

@@ -85,7 +85,7 @@ object SQLExecution extends Logging {
     // And for the root execution, rootExecutionId == executionId.
     if (sc.getLocalProperty(EXECUTION_ROOT_ID_KEY) == null) {
       sc.setLocalProperty(EXECUTION_ROOT_ID_KEY, executionId.toString)
-      sc.addInternalJobTag(executionIdJobTag(sparkSession, executionId))
+      sc.addJobTag(executionIdJobTag(sparkSession, executionId))
     }
     val rootExecutionId = sc.getLocalProperty(EXECUTION_ROOT_ID_KEY).toLong
     executionIdToQueryExecution.put(executionId, queryExecution)
@@ -133,7 +133,7 @@ object SQLExecution extends Logging {
             sparkPlanInfo = SparkPlanInfo.EMPTY,
             time = System.currentTimeMillis(),
             modifiedConfigs = redactedConfigs,
-            jobTags = sc.getJobTags() ++ sc.getInternalJobTags(),
+            jobTags = sc.getJobTags(),
             jobGroupId = Option(sc.getLocalProperty(SparkContext.SPARK_JOB_GROUP_ID))
           )
           try {
@@ -217,7 +217,7 @@ object SQLExecution extends Logging {
       // The current execution is the root execution if rootExecutionId == executionId.
       if (sc.getLocalProperty(EXECUTION_ROOT_ID_KEY) == executionId.toString) {
         sc.setLocalProperty(EXECUTION_ROOT_ID_KEY, null)
-        sc.removeInternalJobTag(executionIdJobTag(sparkSession, executionId))
+        sc.removeJobTag(executionIdJobTag(sparkSession, executionId))
       }
       sc.setLocalProperty(SPARK_JOB_INTERRUPT_ON_CANCEL, originalInterruptOnCancel)
     }
@@ -257,14 +257,14 @@ object SQLExecution extends Logging {
   }
 
   private[sql] def withSessionTagsApplied[T](sparkSession: SparkSession)(block: => T): T = {
-    sparkSession.sparkContext.addInternalJobTag(sparkSession.sessionJobTag)
+    sparkSession.sparkContext.addJobTag(sparkSession.sessionJobTag)
     val userTags = sparkSession.managedJobTags.values().asScala.toSeq
     userTags.foreach(sparkSession.sparkContext.addJobTag)
 
     try {
       block
     } finally {
-      sparkSession.sparkContext.removeInternalJobTag(sparkSession.sessionJobTag)
+      sparkSession.sparkContext.removeJobTag(sparkSession.sessionJobTag)
       userTags.foreach(sparkSession.sparkContext.removeJobTag)
     }
   }

sql/core/src/test/scala/org/apache/spark/sql/SparkSessionJobTaggingAndCancellationSuite.scala

@@ -171,20 +171,18 @@ class SparkSessionJobTaggingAndCancellationSuite
         val tags = job.head.properties.get(SparkContext.SPARK_JOB_TAGS).asInstanceOf[String]
           .split(SparkContext.SPARK_JOB_TAGS_SEP)
 
-        val sessionTag = s"${SparkContext.SPARK_JOB_TAGS_INTERNAL_PREFIX}${ss.sessionJobTag}"
-        val executionRootIdTag = SparkContext.SPARK_JOB_TAGS_INTERNAL_PREFIX +
-          SQLExecution.executionIdJobTag(
+        val executionRootIdTag = SQLExecution.executionIdJobTag(
             ss,
             job.head.properties.get(SQLExecution.EXECUTION_ROOT_ID_KEY).asInstanceOf[String].toLong)
         val userTagsPrefix = s"spark-session-${ss.sessionUUID}-"
 
         ss match {
           case s if s == sessionA => assert(tags.toSet == Set(
-            sessionTag, executionRootIdTag, s"${userTagsPrefix}one"))
+            s.sessionJobTag, executionRootIdTag, s"${userTagsPrefix}one"))
           case s if s == sessionB => assert(tags.toSet == Set(
-            sessionTag, executionRootIdTag, s"${userTagsPrefix}one", s"${userTagsPrefix}two"))
+            s.sessionJobTag, executionRootIdTag, s"${userTagsPrefix}one", s"${userTagsPrefix}two"))
           case s if s == sessionC => assert(tags.toSet == Set(
-            sessionTag, executionRootIdTag, s"${userTagsPrefix}boo"))
+            s.sessionJobTag, executionRootIdTag, s"${userTagsPrefix}boo"))
         }
       }