Skip to content

Commit a6b91d2

Browse files
xuanyuankinghvanhovell
authored andcommitted
[SPARK-30556][SQL][FOLLOWUP] Reset the status changed in SQLExecution.withThreadLocalCaptured
### What changes were proposed in this pull request? Follow up for #27267, reset the status changed in SQLExecution.withThreadLocalCaptured. ### Why are the changes needed? For code safety. ### Does this PR introduce any user-facing change? No. ### How was this patch tested? Existing UT. Closes #27516 from xuanyuanking/SPARK-30556-follow. Authored-by: Yuanjian Li <[email protected]> Signed-off-by: herman <[email protected]>
1 parent 3c1c9b4 commit a6b91d2

File tree

2 files changed

+17
-5
lines changed

2 files changed

+17
-5
lines changed

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

Lines changed: 11 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -177,9 +177,19 @@ object SQLExecution {
177177
val sc = sparkSession.sparkContext
178178
val localProps = Utils.cloneProperties(sc.getLocalProperties)
179179
Future {
180+
val originalSession = SparkSession.getActiveSession
181+
val originalLocalProps = sc.getLocalProperties
180182
SparkSession.setActiveSession(activeSession)
181183
sc.setLocalProperties(localProps)
182-
body
184+
val res = body
185+
// reset active session and local props.
186+
sc.setLocalProperties(originalLocalProps)
187+
if (originalSession.nonEmpty) {
188+
SparkSession.setActiveSession(originalSession.get)
189+
} else {
190+
SparkSession.clearActiveSession()
191+
}
192+
res
183193
}(exec)
184194
}
185195
}

sql/core/src/test/scala/org/apache/spark/sql/internal/ExecutorSideSQLConfSuite.scala

Lines changed: 6 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,8 @@
1717

1818
package org.apache.spark.sql.internal
1919

20+
import java.util.UUID
21+
2022
import org.scalatest.Assertions._
2123

2224
import org.apache.spark.{SparkException, SparkFunSuite, TaskContext}
@@ -144,16 +146,16 @@ class ExecutorSideSQLConfSuite extends SparkFunSuite with SQLTestUtils {
144146
}
145147

146148
// set local configuration and assert
147-
val confValue1 = "e"
149+
val confValue1 = UUID.randomUUID().toString()
148150
createDataframe(confKey, confValue1).createOrReplaceTempView("m")
149151
spark.sparkContext.setLocalProperty(confKey, confValue1)
150-
assert(sql("SELECT * FROM l WHERE EXISTS (SELECT * FROM m)").collect.size == 1)
152+
assert(sql("SELECT * FROM l WHERE EXISTS (SELECT * FROM m)").collect().length == 1)
151153

152154
// change the conf value and assert again
153-
val confValue2 = "f"
155+
val confValue2 = UUID.randomUUID().toString()
154156
createDataframe(confKey, confValue2).createOrReplaceTempView("n")
155157
spark.sparkContext.setLocalProperty(confKey, confValue2)
156-
assert(sql("SELECT * FROM l WHERE EXISTS (SELECT * FROM n)").collect().size == 1)
158+
assert(sql("SELECT * FROM l WHERE EXISTS (SELECT * FROM n)").collect().length == 1)
157159
}
158160
}
159161
}

0 commit comments

Comments
 (0)