[SPARK-32991][SQL] Use conf in shared state as the original configuraion for RESET #30045

yaooqinn · 2020-10-15T03:49:25Z

What changes were proposed in this pull request?

case

the case here covers the static and dynamic SQL configs behavior in sharedState and sessionState, and the specially handled config spark.sql.warehouse.dir
the case can be found here - https://github.com/yaooqinn/sugar/blob/master/src/main/scala/com/netease/mammut/spark/training/sql/WarehouseSCBeforeSS.scala

import java.lang.reflect.Field

import org.apache.spark.sql.SparkSession
import org.apache.spark.{SparkConf, SparkContext}

object WarehouseSCBeforeSS extends App {

  val wh = "spark.sql.warehouse.dir"
  val td = "spark.sql.globalTempDatabase"
  val custom = "spark.sql.custom"

  val conf = new SparkConf()
    .setMaster("local")
    .setAppName("SPARK-32991")
    .set(wh, "./data1")
    .set(td, "bob")

  val sc = new SparkContext(conf)

  val spark = SparkSession.builder()
    .config(wh, "./data2")
    .config(td, "alice")
    .config(custom, "kyao")
    .getOrCreate()

  val confField: Field = spark.sharedState.getClass.getDeclaredField("conf")
  confField.setAccessible(true)
  private val shared: SparkConf = confField.get(spark.sharedState).asInstanceOf[SparkConf]
  println()
  println(s"=====> SharedState: $wh=${shared.get(wh)}")
  println(s"=====> SharedState: $td=${shared.get(td)}")
  println(s"=====> SharedState: $custom=${shared.get(custom, "")}")

  println(s"=====> SessionState: $wh=${spark.conf.get(wh)}")
  println(s"=====> SessionState: $td=${spark.conf.get(td)}")
  println(s"=====> SessionState: $custom=${spark.conf.get(custom, "")}")

  val spark2 = SparkSession.builder().config(td, "fred").getOrCreate()

  println(s"=====> SessionState 2: $wh=${spark2.conf.get(wh)}")
  println(s"=====> SessionState 2: $td=${spark2.conf.get(td)}")
  println(s"=====> SessionState 2: $custom=${spark2.conf.get(custom, "")}")

  SparkSession.setActiveSession(spark)
  spark.sql("RESET")

  println(s"=====> SessionState RESET: $wh=${spark.conf.get(wh)}")
  println(s"=====> SessionState RESET: $td=${spark.conf.get(td)}")
  println(s"=====> SessionState RESET: $custom=${spark.conf.get(custom, "")}")

  val spark3 = SparkSession.builder().getOrCreate()

  println(s"=====> SessionState 3: $wh=${spark2.conf.get(wh)}")
  println(s"=====> SessionState 3: $td=${spark2.conf.get(td)}")
  println(s"=====> SessionState 3: $custom=${spark2.conf.get(custom, "")}")
}

outputs and analysis

// 1. Make the cloned spark conf in shared state respect the warehouse dir from the 1st SparkSession
//=====> SharedState: spark.sql.warehouse.dir=./data1
// 2. ⏬
//=====> SharedState: spark.sql.globalTempDatabase=alice
//=====> SharedState: spark.sql.custom=kyao
//=====> SessionState: spark.sql.warehouse.dir=./data2
//=====> SessionState: spark.sql.globalTempDatabase=alice
//=====> SessionState: spark.sql.custom=kyao
//=====> SessionState 2: spark.sql.warehouse.dir=./data2
//=====> SessionState 2: spark.sql.globalTempDatabase=alice
//=====> SessionState 2: spark.sql.custom=kyao
// 2'.🔼 OK until here
// 3. Make the below 3 ones respect the cloned spark conf in shared state with issue 1 fixed
//=====> SessionState RESET: spark.sql.warehouse.dir=./data1
//=====> SessionState RESET: spark.sql.globalTempDatabase=bob
//=====> SessionState RESET: spark.sql.custom=
// 4. Then the SparkSessions created after RESET will be corrected.
//=====> SessionState 3: spark.sql.warehouse.dir=./data1
//=====> SessionState 3: spark.sql.globalTempDatabase=bob
//=====> SessionState 3: spark.sql.custom=

In this PR, we gather all valid config to the cloned conf of sharedState during being constructed, well, actually only spark.sql.warehouse.dir is missing. Then we use this conf as defaults for RESET Command.

SparkSession.clearActiveSession/clearDefaultSession will make the shared state invisible and unsharable. They will be internal only soon (confirmed with Wenchen), so cases with them called will not be a problem.

Why are the changes needed?

bugfix for programming API to call RESET while users creating SparkContext first and config SparkSession later.

Does this PR introduce any user-facing change?

yes, before this change when you use programming API and call RESET, all configs will be reset to SparkContext.conf, now they go to SparkSession.sharedState.conf

How was this patch tested?

new tests

…tion for RESET

SparkQA · 2020-10-15T05:28:56Z

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/34400/

SparkQA · 2020-10-15T05:46:29Z

Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/34400/

SparkQA · 2020-10-15T07:05:01Z

Test build #129792 has finished for PR 30045 at commit f253fad.

This patch fails due to an unknown error code, -9.
This patch merges cleanly.
This patch adds no public classes.

yaooqinn · 2020-10-15T07:06:20Z

cc @cloud-fan @maropu @gatorsmile thanks~

SparkQA · 2020-10-15T08:31:00Z

Test build #129815 has finished for PR 30045 at commit a52c86a.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2020-10-15T08:36:27Z

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/34420/

SparkQA · 2020-10-15T09:01:29Z

Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/34420/

SparkQA · 2020-10-15T10:38:43Z

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/34431/

SparkQA · 2020-10-15T10:55:23Z

Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/34431/

SparkQA · 2020-10-15T14:52:07Z

Test build #129825 has finished for PR 30045 at commit ccbf0ae.

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

yaooqinn · 2020-10-16T07:30:47Z

cc @hvanhovell too

SparkQA · 2020-10-16T08:18:29Z

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/34488/

SparkQA · 2020-10-16T08:36:08Z

Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/34488/

sql/core/src/main/scala/org/apache/spark/sql/execution/command/SetCommand.scala

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

sql/core/src/main/scala/org/apache/spark/sql/internal/SharedState.scala

SparkQA · 2020-10-16T11:17:10Z

Test build #129883 has finished for PR 30045 at commit 6848b2f.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
public static class Bitmaps
public static class BitmapArrays
class BlockPushErrorHandler implements ErrorHandler
public class MergedBlockMeta
public class OneForOneBlockPusher
public class FinalizeShuffleMerge extends BlockTransferMessage
public class MergeStatuses extends BlockTransferMessage
public class PushBlockStream extends BlockTransferMessage

SparkQA · 2020-10-16T16:10:39Z

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/34506/

SparkQA · 2020-10-16T16:29:42Z

Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/34506/

SparkQA · 2020-10-16T19:37:18Z

Test build #129900 has finished for PR 30045 at commit 91c2e91.

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

HyukjinKwon · 2020-10-18T08:54:45Z

From a cursory look, I think this is correct. But probably it's best to have @hvanhovell's look.

cloud-fan · 2020-10-22T09:23:18Z

sql/core/src/main/scala/org/apache/spark/sql/internal/SharedState.scala

      // When neither spark.sql.warehouse.dir nor hive.metastore.warehouse.dir is set
      // we will set hive.metastore.warehouse.dir to the default value of spark.sql.warehouse.dir.
-      val sparkWarehouseDir = sparkConf.get(WAREHOUSE_PATH)
+      val sparkWarehouseDir = sparkWarehouse.getOrElse(sparkConf.get(WAREHOUSE_PATH))


sparkWarehouse already has .orElse(sparkConf.getOption(WAREHOUSE_PATH.key)). Do you mean sparkWarehouse.getOrElse(WAREHOUSE_PATH.defaultValueString)?

Yes, here will finally touch the default value string if never matched before

can we change to sparkWarehouse.getOrElse(WAREHOUSE_PATH.defaultValueString) so that it's more explicit?

cloud-fan · 2020-10-22T09:26:14Z

sql/hive/src/test/scala/org/apache/spark/sql/hive/HiveSharedStateSuite.scala

    val metastorePath = Utils.createTempDir()
    val tmpDb = "tmp_db"

    // The initial configs used to generate SharedState, none of these should affect the global


this is wrong now, as warehouse conf is an exception.

SparkQA · 2020-10-22T11:05:56Z

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/34757/

SparkQA · 2020-10-22T11:28:00Z

Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/34757/

SparkQA · 2020-10-22T11:55:46Z

Test build #130150 has finished for PR 30045 at commit 0cbb4bc.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2020-10-22T16:34:16Z

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/34773/

SparkQA · 2020-10-22T16:56:52Z

Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/34773/

SparkQA · 2020-10-22T19:44:30Z

Test build #130166 has finished for PR 30045 at commit d194796.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

yaooqinn · 2020-10-23T02:16:11Z

retest this please

SparkQA · 2020-10-23T02:59:12Z

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/34786/

SparkQA · 2020-10-23T03:27:44Z

Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/34786/

cloud-fan · 2020-10-23T05:52:37Z

GA passed, merging to master, thanks!

SparkQA · 2020-10-23T06:36:29Z

Test build #130184 has finished for PR 30045 at commit d194796.

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

hvanhovell · 2020-11-13T13:32:12Z

@cloud-fan @yaooqinn I have a question about the semantics. We currently reset to the initial state of the first session. You can have many different sessions (e.g. in thrift server) with different initial settings, IMO it would be more sane to reset to the initial state of the current session. WDYT?

HyukjinKwon · 2020-11-13T14:46:54Z

^^ makes sense to me.

cloud-fan · 2020-11-16T08:35:30Z

makes sense. @yaooqinn what do you think?

yaooqinn · 2020-11-16T09:13:20Z

make sense to me too, I will raise a followup later.

yaooqinn · 2020-12-06T17:26:49Z

I have a question about the semantics. We currently reset to the initial state of the first session. You can have many different sessions (e.g. in thrift server) with different initial settings, IMO it would be more sane to reset to the initial state of the current session. WDYT?

I did some research:

w/o calling clearDefaultSession and clearActiveSession(there was a PR to make these 2 internal but reverted for some reasons), the existing APIs for creating new SparkSession cannot lead users to such a situation.

Let's assume the initial configs we talk about here is those going with the instantiating process of a SparkSession instance, not those being the first-time set

w/ the SparkSession.newSession() API, there are no parameters provided to set initial configs
w the SparkSession.Builder.getOrCreate, we are just referencing the original one only but w/o creating a new SparkSession instance.

So do we have a way to actually create a new session with initial configs when there is an existing active one? The answer is NO.

This situation only happens when these 2 APIs called, but the current approach actually meets our goal here. We do keep the session configs per session in such a use case.

The actual problem here that revealed in the following case is that the GLOBAL SharedState is not being shared after those clear-like APIs being called

bin/spark-shell \
  --conf spark.sql.warehouse.dir=./warehouse \
  --conf spark.sql.globalTempDatabase=mytemp \
  --conf spark.sql.custom=abc

scala> org.apache.spark.sql.SparkSession.clearDefaultSession()

scala> org.apache.spark.sql.SparkSession.clearActiveSession()

scala> val nes = org.apache.spark.sql.SparkSession.builder.config("spark.sql.warehouse.dir", "w2").config("spark.sql.globalTempDatabase", "m2").config("spark.sql.custom", "xyz").getOrCreate
20/12/07 00:59:35 WARN SparkContext: Using an existing SparkContext; some configuration may not take effect.
nes: org.apache.spark.sql.SparkSession = org.apache.spark.sql.SparkSession@175f1ff5

scala> nes.conf.get("spark.sql.warehouse.dir")
20/12/07 01:00:06 WARN SharedState: Not allowing to set spark.sql.warehouse.dir or hive.metastore.warehouse.dir in SparkSession's options, it should be set statically for cross-session usages
res2: String = w2

scala> nes.conf.get("spark.sql.globalTempDatabase")
res3: String = m2

scala> nes.conf.get("spark.sql.custom")
res4: String = xyz

scala> nes.sql("reset")
res5: org.apache.spark.sql.DataFrame = []

scala> nes.conf.get("spark.sql.globalTempDatabase")
res6: String = m2

scala> nes.conf.get("spark.sql.custom")
res7: String = xyz

cc @gatorsmile @cloud-fan

cloud-fan · 2020-12-07T04:37:20Z

@yaooqinn active session is a thread local. What if we create a new session in another thread?

yaooqinn · 2020-12-07T06:24:14Z

@yaooqinn active session is a thread local. What if we create a new session in another thread?

w/o SparkSession.clearDefaultSession(), we will not create a new one

w/ SparkSession.clearDefaultSession(), the shared state is also not able to reach

cloud-fan · 2020-12-07T07:35:21Z

I think you are right about reality. But it seems safer to handle the per-session initial configs as it's a property of the SparkSession. It's kind of a potential bug and we'd better fix it earlier. @yaooqinn what do you think?

yaooqinn · 2020-12-07T08:40:39Z

I have created #30642 to fix this

…s first ### What changes were proposed in this pull request? As a follow-up of #30045, we modify the RESET command here to respect the session initial configs per session first then fall back to the `SharedState` conf, which makes each session could maintain a different copy of initial configs for resetting. ### Why are the changes needed? to make reset command saner. ### Does this PR introduce _any_ user-facing change? yes, RESET will respect session initials first not always go to the system defaults ### How was this patch tested? add new tests Closes #30642 from yaooqinn/SPARK-32991-F. Authored-by: Kent Yao <[email protected]> Signed-off-by: Wenchen Fan <[email protected]>

…s first ### What changes were proposed in this pull request? As a follow-up of #30045, we modify the RESET command here to respect the session initial configs per session first then fall back to the `SharedState` conf, which makes each session could maintain a different copy of initial configs for resetting. ### Why are the changes needed? to make reset command saner. ### Does this PR introduce _any_ user-facing change? yes, RESET will respect session initials first not always go to the system defaults ### How was this patch tested? add new tests Closes #30642 from yaooqinn/SPARK-32991-F. Authored-by: Kent Yao <[email protected]> Signed-off-by: Wenchen Fan <[email protected]> (cherry picked from commit 205d8e4) Signed-off-by: Wenchen Fan <[email protected]>

[SPARK-32991][SQL] Use conf in shared state as the original configura…

f253fad

…tion for RESET

fix test

a52c86a

fix test

ccbf0ae

dongjoon-hyun added the SQL label Oct 15, 2020

Merge branch 'master' into SPARK-32991

6848b2f

cloud-fan reviewed Oct 16, 2020

View reviewed changes

sql/core/src/main/scala/org/apache/spark/sql/execution/command/SetCommand.scala Show resolved Hide resolved

cloud-fan reviewed Oct 16, 2020

View reviewed changes

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

cloud-fan reviewed Oct 16, 2020

View reviewed changes

sql/core/src/main/scala/org/apache/spark/sql/internal/SharedState.scala Show resolved Hide resolved

update comments and fix test

91c2e91

cloud-fan reviewed Oct 22, 2020

View reviewed changes

cloud-fan approved these changes Oct 22, 2020

View reviewed changes

update comment

0cbb4bc

dft val str

d194796

cloud-fan closed this in e21bb71 Oct 23, 2020

yaooqinn mentioned this pull request Dec 7, 2020

[SPARK-32991][SQL] [FOLLOWUP] Reset command relies on session initials first #30642

Closed

[SPARK-32991][SQL] Use conf in shared state as the original configuraion for RESET #30045

[SPARK-32991][SQL] Use conf in shared state as the original configuraion for RESET #30045

Uh oh!

Conversation

yaooqinn commented Oct 15, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What changes were proposed in this pull request?

case

outputs and analysis

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

Uh oh!

SparkQA commented Oct 15, 2020

Uh oh!

SparkQA commented Oct 15, 2020

Uh oh!

SparkQA commented Oct 15, 2020

Uh oh!

yaooqinn commented Oct 15, 2020

Uh oh!

SparkQA commented Oct 15, 2020

Uh oh!

SparkQA commented Oct 15, 2020

Uh oh!

SparkQA commented Oct 15, 2020

Uh oh!

SparkQA commented Oct 15, 2020

Uh oh!

SparkQA commented Oct 15, 2020

Uh oh!

SparkQA commented Oct 15, 2020

Uh oh!

yaooqinn commented Oct 16, 2020

Uh oh!

SparkQA commented Oct 16, 2020

Uh oh!

SparkQA commented Oct 16, 2020

Uh oh!

Uh oh!

Uh oh!

Uh oh!

SparkQA commented Oct 16, 2020

Uh oh!

SparkQA commented Oct 16, 2020

Uh oh!

SparkQA commented Oct 16, 2020

Uh oh!

SparkQA commented Oct 16, 2020

Uh oh!

HyukjinKwon commented Oct 18, 2020

Uh oh!

cloud-fan Oct 22, 2020

Choose a reason for hiding this comment

Uh oh!

yaooqinn Oct 22, 2020

Choose a reason for hiding this comment

Uh oh!

cloud-fan Oct 22, 2020

Choose a reason for hiding this comment

Uh oh!

yaooqinn Oct 22, 2020

Choose a reason for hiding this comment

Uh oh!

cloud-fan Oct 22, 2020

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Oct 22, 2020

Uh oh!

SparkQA commented Oct 22, 2020

Uh oh!

SparkQA commented Oct 22, 2020

Uh oh!

SparkQA commented Oct 22, 2020

Uh oh!

SparkQA commented Oct 22, 2020

Uh oh!

SparkQA commented Oct 22, 2020

Uh oh!

yaooqinn commented Oct 23, 2020

yaooqinn commented Oct 15, 2020 •

edited

Loading

yaooqinn commented Dec 6, 2020 •

edited

Loading