Clone properties only for SQL for backward compatibility

Author: Andrew Or

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

@@ -351,8 +351,12 @@ class SparkContext(config: SparkConf) extends Logging with ExecutorAllocationCli
   protected[spark] val localProperties = new InheritableThreadLocal[Properties] {
     override protected def childValue(parent: Properties): Properties = {
       // Note: make a clone such that changes in the parent properties aren't reflected in
-      // the those of the children threads, which has confusing semantics (SPARK-10564).
-      SerializationUtils.clone(parent).asInstanceOf[Properties]
+      // the those of the children threads, which has confusing semantics (SPARK-10563).
+      if (conf.get("spark.localProperties.clone", "false").toBoolean) {
+        SerializationUtils.clone(parent).asInstanceOf[Properties]
+      } else {
+        new Properties(parent)
+      }
     }
     override protected def initialValue(): Properties = new Properties()
   }

core/src/test/scala/org/apache/spark/ThreadingSuite.scala

@@ -214,6 +214,7 @@ class ThreadingSuite extends SparkFunSuite with LocalSparkContext with Logging {
 
   test("mutation in parent local property does not affect child (SPARK-10563)") {
     sc = new SparkContext("local", "test")
+    sc.conf.set("spark.localProperties.clone", "true")
     val originalTestValue: String = "original-value"
     var threadTestValue: String = null
     sc.setLocalProperty("test", originalTestValue)

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

@@ -78,6 +78,12 @@ class SQLContext(@transient val sparkContext: SparkContext)
   sparkContext.addSparkListener(listener)
   sparkContext.ui.foreach(new SQLTab(this, _))
 
+  // Execution IDs go through SparkContext's local properties, which are not safe to use with
+  // fork join pools by default. In particular, even after a child thread is spawned, if the
+  // parent sets a property the value may be reflected in the child. This leads to undefined
+  // consequences such as SPARK-10548, so we should just clone the properties instead to be safe.
+  sparkContext.conf.set("spark.localProperties.clone", "true")
+
   /**
    * Set Spark SQL configuration properties.
    *