[SPARK-30556][SQL] Copy sparkContext.localproperties to child thread inSubqueryExec.executionContext #27267

ajithme · 2020-01-17T17:12:22Z

What changes were proposed in this pull request?

In org.apache.spark.sql.execution.SubqueryExec#relationFuture make a copy of org.apache.spark.SparkContext#localProperties and pass it to the sub-execution thread in org.apache.spark.sql.execution.SubqueryExec#executionContext

Why are the changes needed?

Local properties set via sparkContext are not available as TaskContext properties when executing jobs and threadpools have idle threads which are reused

Explanation:
When SubqueryExec, the relationFuture is evaluated via a separate thread. The threads inherit the localProperties from sparkContext as they are the child threads.
These threads are created in the executionContext (thread pools). Each Thread pool has a default keepAliveSeconds of 60 seconds for idle threads.
Scenarios where the thread pool has threads which are idle and reused for a subsequent new query, the thread local properties will not be inherited from spark context (thread properties are inherited only on thread creation) hence end up having old or no properties set. This will cause taskset properties to be missing when properties are transferred by child thread via sparkContext.runJob/submitJob

Does this PR introduce any user-facing change?

No

How was this patch tested?

Added UT

ajithme · 2020-01-17T17:14:16Z

cc @dongjoon-hyun @srowen @HyukjinKwon @gatorsmile @cloud-fan

dongjoon-hyun · 2020-01-17T17:36:10Z

ok to test

dongjoon-hyun

For WIP PR, please add [WIP] into the PR title.

How was this patch tested?

WIP

srowen · 2020-01-17T17:47:35Z

Closely related to #27266?

ajithme · 2020-01-17T17:52:36Z

Closely related to #27266?

Currently, org.apache.spark.sql.execution.SubqueryExec#executionContext has a hardcoded size of 16 threads which makes writing UT difficult so i plan to make it configurable + fix subquery thread bug hence created a new pull request compared to broadcast bug (which is independent of this fix) in #27266

SparkQA · 2020-01-17T21:35:30Z

Test build #116960 has finished for PR 27267 at commit 5042156.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

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

hvanhovell · 2020-01-20T15:56:03Z

IMO the thread pools should not be a big issue. The subquery is guaranteed to be executed on a different thread (you can even add an assert for this). You just set some unique property on the local properties (value should also be unique), construct the something that contains a broadcast join, use an accumulator that you modify using either a UDF (easy) or a dataset operation.

ajithme · 2020-01-20T16:39:02Z

IMO the thread pools should not be a big issue. The subquery is guaranteed to be executed on a different thread (you can even add an assert for this). You just set some unique property on the local properties (value should also be unique), construct the something that contains a broadcast join, use an accumulator that you modify using either a UDF (easy) or a dataset operation.

Agree. But with a pool size of 16, i will have to ensure all 16 threads are used at-least once and are alive to reproduce this issue and not to make that test flaky. Its lot easier if i can set the pool size to 1 in test to reproduce.

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

hvanhovell · 2020-01-21T09:13:54Z

@ajithme why do we need to make sure all 16 threads are used?

cloud-fan · 2020-01-21T09:31:47Z

I think this is a common problem when we run code in a thread pool. I just realized that we hit a similar issue before with SQLConf.get and we add a hack to work around it.

Shall we add a util function to easily capture the necessary thread locals and propagate them to thread pool? For example, we can add a SQLExecution.withThreadLocalCaptured

def withThreadLocalCaptured(f: => T): Future[T] = {
  val activeSession = ...
  val localProperties = ...
  // any other important thread locals?
  Future {
    // set active sessionn
    // set local properties
   f
  }
}

ajithme · 2020-01-21T09:34:27Z

util function to easily capture the necessary thread locals and propagate them to thread pool? For example, we can add a SQLExecution.withThreadLocalCaptured

sure @cloud-fan , i think similar approach was proposed by @hvanhovell , i will update my PR to introduce a utility than having a local fix.

ajithme · 2020-01-21T09:36:44Z

@ajithme why do we need to make sure all 16 threads are used?

if not, the thread executing the subquery may be a new one ( created from pool,) thus inheriting the localproperties from sparkContext . This JIRA issue is reproduced only when pool threads are reused (within keepAlive time)

ajithme · 2020-01-21T11:07:36Z

cc @cloud-fan @hvanhovell i have updated PR as per the suggestion. Please review

ajithme · 2020-01-21T11:28:33Z

This seems to be problems what i try to fix in

BroadcastExchangeExec (Refer: [SPARK-22590][SQL] Copy sparkContext.localproperties to child thread in BroadcastExchangeExec.executionContext #27266)
SubqueryBroadcastExec
SubqueryExec (Refer: [SPARK-30556][SQL] Copy sparkContext.localproperties to child thread inSubqueryExec.executionContext #27267)

so should i try to fix all in one PR or should i have a separate PR for each.? i previously raised them separately so that i can complete them with individual UTs.

Please suggest @cloud-fan @srowen @dongjoon-hyun @hvanhovell, i am kinda neutral if to make separate PRs or single PR

cloud-fan · 2020-01-21T12:21:44Z

I think we can fix one place in this PR, and send more PRs to fix more places later, with tests.

SparkQA · 2020-01-21T13:24:50Z

Test build #117171 has finished for PR 27267 at commit fea9160.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2020-01-21T13:26:04Z

Test build #117173 has finished for PR 27267 at commit 577904f.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

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

ajithme · 2020-01-22T19:32:16Z

@dongjoon-hyun Updated with fix

SparkQA · 2020-01-22T23:44:19Z

Test build #117252 has finished for PR 27267 at commit f1cac4d.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

dongjoon-hyun

+1, LGTM. Merged to master.

dongjoon-hyun · 2020-01-23T02:23:11Z

Could you make a backporting PR against branch-2.4, @ajithme ?

ajithme · 2020-01-23T02:33:52Z

Thanks @dongjoon-hyun @cloud-fan @hvanhovell

@dongjoon-hyun sure, i will make a backport PR for this over 2.4

hvanhovell · 2020-01-23T12:29:06Z

sql/catalyst/src/main/scala/org/apache/spark/sql/internal/StaticSQLConf.scala

+      .checkValue(thres => thres > 0 && thres <= 128, "The threshold must be in (0,128].")
      .createWithDefault(128)

+  val SUBQUERY_MAX_THREAD_THRESHOLD =


This is actually a different change right?

It should also be a static configuration since we can only change it at startup.

This is just opening as a configuration to make the change testable. DO you want me to raise separate PR just to make this configuration change seperate.?

This is part of StaticSQLConf which is defined at startup, is there any other mechanism to define static conf.?

If its a static conf, then why isn't your unite test failing? Moreover, if its static then setting it in your test probably does not have any effect because we use the same JVM/SparkContext to run most tests, the chances are pretty high that it has been set before.

i see. as per documentation in

spark/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/StaticSQLConf.scala

Line 26 in 705fc6a

* Static SQL configuration is a cross-session, immutable Spark configuration. External users can

its should not modified. I followed the same way BroadcastExchangeExec creates executionContext. My initial guess is executionContext is not created till first subquery hence it work for UT. i will further investigate and get back with analysis.

hvanhovell · 2020-01-23T12:33:40Z

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

+        }
+
+        // set local configuration and assert
+        val confValue1 = "e"


IMO it is better to use something unique here to avoid a fluke. How about using UUID.randomUUID().toString()?

I am neutral about both approaches. Just wanted to have a fixed input value to get predictable output value from testcase. I can update this and raise followup if you insist

hvanhovell · 2020-01-23T12:34:34Z

@ajithme can you address my comments in a follow up?

gatorsmile · 2020-02-09T05:58:31Z

@ajithme Can you submit a PR to address the comments?

gatorsmile · 2020-02-09T06:02:06Z

@xuanyuanking Could you submit a PR to address the comments?

gatorsmile · 2020-02-09T06:04:25Z

https://github.com/apache/spark/pull/27267/files#r370089158 is the major comment we need to address.

gatorsmile · 2020-02-09T06:06:01Z

sql/catalyst/src/main/scala/org/apache/spark/sql/internal/StaticSQLConf.scala

+      .checkValue(thres => thres > 0 && thres <= 128, "The threshold must be in (0,128].")
      .createWithDefault(128)

+  val SUBQUERY_MAX_THREAD_THRESHOLD =


I think we should submit a separate PR for this change.

I think we could keep this in StaticSQLConf.

I locally test the static config takes effect, it can only be set while startup.

The UT can pass because it was used in lazy val SubqueryExec.relationFuture on the executor side, so the withSQLConf in UT could set the config before executor start.

cc @hvanhovell

ajithme · 2020-02-10T03:23:05Z

@ajithme Can you submit a PR to address the comments?

@gatorsmile @xuanyuanking Sure, i will submit a follow up PR shortly.

xuanyuanking · 2020-02-10T04:06:56Z

@ajithme Sorry, I nearly submit a follow-up PR, could you help on reviewing? Thanks.

….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]>

….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]> (cherry picked from commit a6b91d2) Signed-off-by: herman <[email protected]>

dbaliafroozeh · 2020-02-18T14:32:13Z

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

+    val localProps = Utils.cloneProperties(sc.getLocalProperties)
+    Future {
+      SparkSession.setActiveSession(activeSession)
+      sc.setLocalProperties(localProps)


@hvanhovell two questions:

Shouldn't we clonelocalProps here? in the sense that what if a concurrent thread modify them?

Does the order of setting localProps and activeSession matter?

the localProps is already a clone: https://github.com/apache/spark/pull/27267/files#diff-ab49028253e599e6e74cc4f4dcb2e3a8R178

And I think the order doesn't matter.

….withThreadLocalCaptured Follow up for apache#27267, reset the status changed in SQLExecution.withThreadLocalCaptured. For code safety. No. Existing UT. Closes apache#27516 from xuanyuanking/SPARK-30556-follow. (cherry picked from commit a6b91d2)

…tion 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. (cherry picked from commit a6b91d2) Closes #27633 from xuanyuanking/SPARK-30556-backport. Authored-by: Yuanjian Li <[email protected]> Signed-off-by: Wenchen Fan <[email protected]>

….withThreadLocalCaptured ### What changes were proposed in this pull request? Follow up for apache#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 apache#27516 from xuanyuanking/SPARK-30556-follow. Authored-by: Yuanjian Li <[email protected]> Signed-off-by: herman <[email protected]>

Copy localproperties to child thread of subquery execution

5042156

ajithme force-pushed the subquerylocalprop branch from a229ecb to 5042156 Compare January 17, 2020 17:15

dongjoon-hyun changed the title ~~[SPARK-30556] Copy sparkContext.localproperties to child thread inSubqueryExec.executionContext~~ [SPARK-30556][SQL] Copy sparkContext.localproperties to child thread inSubqueryExec.executionContext Jan 17, 2020

dongjoon-hyun reviewed Jan 17, 2020

View reviewed changes

dongjoon-hyun changed the title ~~[SPARK-30556][SQL] Copy sparkContext.localproperties to child thread inSubqueryExec.executionContext~~ [WIP][SPARK-30556][SQL] Copy sparkContext.localproperties to child thread inSubqueryExec.executionContext Jan 17, 2020

srowen mentioned this pull request Jan 17, 2020

[SPARK-22590][SQL] Copy sparkContext.localproperties to child thread in BroadcastExchangeExec.executionContext #27266

Closed

hvanhovell reviewed Jan 20, 2020

View reviewed changes

sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala Outdated Show resolved Hide resolved

cloud-fan reviewed Jan 21, 2020

View reviewed changes

sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala Outdated Show resolved Hide resolved

ajithme changed the title ~~[WIP][SPARK-30556][SQL] Copy sparkContext.localproperties to child thread inSubqueryExec.executionContext~~ [SPARK-30556][SQL] Copy sparkContext.localproperties to child thread inSubqueryExec.executionContext Jan 21, 2020

Update with UT

577904f

ajithme force-pushed the subquerylocalprop branch from fea9160 to 577904f Compare January 21, 2020 09:04

ajithme added 2 commits January 21, 2020 16:31

Updated as per review comments

b6b5bae

Update format

c2b157a

Update to accept active session also

1ca7981

dongjoon-hyun reviewed Jan 22, 2020

View reviewed changes

sql/core/src/test/scala/org/apache/spark/sql/internal/ExecutorSideSQLConfSuite.scala Outdated Show resolved Hide resolved

dongjoon-hyun reviewed Jan 22, 2020

View reviewed changes

sql/core/src/test/scala/org/apache/spark/sql/internal/ExecutorSideSQLConfSuite.scala Outdated Show resolved Hide resolved

dongjoon-hyun reviewed Jan 22, 2020

View reviewed changes

sql/core/src/test/scala/org/apache/spark/sql/internal/ExecutorSideSQLConfSuite.scala Outdated Show resolved Hide resolved

Review

f1cac4d

dongjoon-hyun approved these changes Jan 23, 2020

View reviewed changes

dongjoon-hyun closed this in bbab2bb Jan 23, 2020

ajithme mentioned this pull request Jan 23, 2020

[SPARK-30556][SQL][2.4] Copy sparkContext.localproperties to child thread inSubqueryExec.executionContext #27340

Closed

hvanhovell reviewed Jan 23, 2020

View reviewed changes

gatorsmile reviewed Feb 9, 2020

View reviewed changes

xuanyuanking mentioned this pull request Feb 10, 2020

[SPARK-30556][SQL][FOLLOWUP] Reset the status changed in SQLExecution.withThreadLocalCaptured #27516

Closed

dbaliafroozeh reviewed Feb 18, 2020

View reviewed changes

xuanyuanking mentioned this pull request Feb 19, 2020

[SPARK-30556][SQL][BACKPORT-2.4] Reset the status changed in SQLExecution withThreadLocalCaptured #27633

Closed

[SPARK-30556][SQL] Copy sparkContext.localproperties to child thread inSubqueryExec.executionContext #27267

[SPARK-30556][SQL] Copy sparkContext.localproperties to child thread inSubqueryExec.executionContext #27267

Uh oh!

Conversation

ajithme commented Jan 17, 2020 • edited by dongjoon-hyun Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

Uh oh!

ajithme commented Jan 17, 2020

Uh oh!

dongjoon-hyun commented Jan 17, 2020

Uh oh!

dongjoon-hyun left a comment

Choose a reason for hiding this comment

How was this patch tested?

Uh oh!

srowen commented Jan 17, 2020

Uh oh!

ajithme commented Jan 17, 2020

Uh oh!

SparkQA commented Jan 17, 2020

Uh oh!

Uh oh!

hvanhovell commented Jan 20, 2020

Uh oh!

ajithme commented Jan 20, 2020

Uh oh!

Uh oh!

hvanhovell commented Jan 21, 2020

Uh oh!

cloud-fan commented Jan 21, 2020

Uh oh!

ajithme commented Jan 21, 2020

Uh oh!

ajithme commented Jan 21, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

ajithme commented Jan 21, 2020

Uh oh!

ajithme commented Jan 21, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

cloud-fan commented Jan 21, 2020

Uh oh!

SparkQA commented Jan 21, 2020

Uh oh!

SparkQA commented Jan 21, 2020

Uh oh!

Uh oh!

Uh oh!

Uh oh!

ajithme commented Jan 22, 2020

Uh oh!

SparkQA commented Jan 22, 2020

Uh oh!

dongjoon-hyun left a comment

Choose a reason for hiding this comment

Uh oh!

dongjoon-hyun commented Jan 23, 2020

Uh oh!

ajithme commented Jan 23, 2020

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ajithme Jan 23, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

ajithme commented Jan 17, 2020 •

edited by dongjoon-hyun

Loading

ajithme commented Jan 21, 2020 •

edited

Loading

ajithme commented Jan 21, 2020 •

edited

Loading

ajithme Jan 23, 2020 •

edited

Loading