[SPARK-57262][SQL][WEBUI] Job description derived from a query should respect spark.sql.redaction.string.regex

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

@@ -170,9 +170,7 @@ object SQLExecution extends Logging {
       val desc = Option(sc.getLocalProperty(SPARK_JOB_DESCRIPTION))
         .filter(_ => truncateLength > 0)
         .map { sqlStr =>
-          val redactedStr = Utils
-            .redact(sparkSession.sessionState.conf.stringRedactionPattern, sqlStr)
-          redactedStr.substring(0, Math.min(truncateLength, redactedStr.length))
+          sqlStr.substring(0, Math.min(truncateLength, sqlStr.length))
         }.getOrElse(callSite.shortForm)
 
       val globalConfigs = sparkSession.sharedState.conf.getAll.toMap

sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkSQLDriver.scala

@@ -66,7 +66,9 @@ private[hive] class SparkSQLDriver(val sparkSession: SparkSession = SparkSQLEnv.
       val substitutorCommand = SQLConf.withExistingConf(sparkSession.sessionState.conf) {
         new VariableSubstitution().substitute(command)
       }
-      sparkSession.sparkContext.setJobDescription(substitutorCommand)
+      val redactedCommand =
+        Utils.redact(sparkSession.sessionState.conf.stringRedactionPattern, substitutorCommand)
+      sparkSession.sparkContext.setJobDescription(redactedCommand)
 
       // Parse with an empty parameter context to enable pre-parsing phase that scans for
       // parameter markers. If any parameter markers (:name or ?) are found in the SQL,

sql/hive-thriftserver/src/test/scala/org/apache/spark/sql/hive/thriftserver/SparkSQLDriverSuite.scala

@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.hive.thriftserver
+
+import org.apache.spark.SparkContext
+import org.apache.spark.scheduler.{SparkListener, SparkListenerJobStart}
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.test.SharedSparkSession
+import org.apache.spark.util.Utils.REDACTION_REPLACEMENT_TEXT
+
+class SparkSQLDriverSuite extends SharedSparkSession {
+
+  test("SPARK-57262: job description should be redacted by spark.sql.redaction.string.regex") {
+    withSQLConf(SQLConf.SQL_STRING_REDACTION_PATTERN.key -> "password=([^\\s]+)") {
+      var jobDescription: String = null
+      spark.sparkContext.addSparkListener(new SparkListener {
+        override def onJobStart(jobStart: SparkListenerJobStart): Unit = {
+          jobDescription =
+            jobStart.properties.getProperty(SparkContext.SPARK_JOB_DESCRIPTION)
+        }
+      })
+
+      val driver = new SparkSQLDriver(spark)
+      try {
+        driver.run("SELECT 'password=secret123'")
+      } finally {
+        driver.close()
+      }
+
+      spark.sparkContext.listenerBus.waitUntilEmpty()
+      assert(jobDescription != null)
+      assert(!jobDescription.contains("secret123"))
+      assert(jobDescription.contains(REDACTION_REPLACEMENT_TEXT))
+    }
+  }
+}