You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "juliuszsompolski (via GitHub)" <gi...@apache.org> on 2024/09/09 16:08:11 UTC
[PR] [SPARK-48195] Save and reuse RDD/Broadcast created by SparkPlan [spark]
juliuszsompolski opened a new pull request, #48037:
URL: https://github.com/apache/spark/pull/48037
### What changes were proposed in this pull request?
Save the RDD created by doExecute, instead of creating a new one in execute each time.
Currently, many types of SparkPlans already save the RDD they create. For example, shuffle just save `lazy val inputRDD: RDD[InternalRow] = child.execute()`. It creates inconsistencies when an action (e.g. repeated `df.collect()`) is executed on Dataframe twice:
* The SparkPlan will be reused, since the same `df.queryExecution.executedPlan` will be used.
* Any not-result stage will be reused, as the shuffle operators will just have their `inputRDD` reused.
* However, for result stage, `execute()` will call `doExecute()` again, and the logic of generating the actual execution RDD will be reexecuted for the result stage.
This means that for example for the result stage, WSCG code gen will generate and compile new code, create a new RDD out of it. Generation of execution RDDs is also often influenced by config: for example, staying with WSCG, various configs like `spark.sql.codegen.hugeMethodLimit` or `spark.sql.codegen.methodSplitThreshold`. The fact that upon re-execution this will be evaluated anew for the result stage, but not for earlier stages creates inconsistencies in what config changes are visible.
By saving the result of `doExecute` and reusing the RDD in `execute` we make sure that work in creating that RDD is not duplicated, and it is more consistent that all RDDs of the plan are reused, same as with the `executedPlan`.
Note, that while the results of earlier shuffle stages are also reused, the result stage still does get executed again, as the result of it are not saved and available for Reuse in BlockManager.
We also add a `Lazy` utility instead of using `lazy val` to deal with shortcomings of scala lazy val.
### Why are the changes needed?
Resolved subtle inconsistencies coming from object reuse vs. recreating objects from scratch.
### Does this PR introduce _any_ user-facing change?
Subtle changes caused by the RDD being reused, e.g. when a config change might be picked up. However, it makes things more consistent. Spark 4.0.0 might be a good candidate for making such a change.
### How was this patch tested?
Existing SQL execution tests validate that the change in SparkPlan works.
Tests were added for the new Lazy utility.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Github Copilot
(trivial code completion suggestions)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48195] Save and reuse RDD/Broadcast created by SparkPlan [spark]
Posted by "juliuszsompolski (via GitHub)" <gi...@apache.org>.
juliuszsompolski commented on PR #48037:
URL: https://github.com/apache/spark/pull/48037#issuecomment-2338534250
@JoshRosen @cloud-fan
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48195] Save and reuse RDD/Broadcast created by SparkPlan [spark]
Posted by "juliuszsompolski (via GitHub)" <gi...@apache.org>.
juliuszsompolski commented on code in PR #48037:
URL: https://github.com/apache/spark/pull/48037#discussion_r1752078651
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/SparkPlan.scala:
##########
@@ -182,6 +182,10 @@ abstract class SparkPlan extends QueryPlan[SparkPlan] with Logging with Serializ
/** Specifies sort order for each partition requirements on the input data for this operator. */
def requiredChildOrdering: Seq[Seq[SortOrder]] = Seq.fill(children.size)(Nil)
+ private val executeRDD = LazyTry {
Review Comment:
Currently it's already cached at shuffle boundaries https://github.com/apache/spark/blob/master/sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/ShuffleExchangeExec.scala#L257
and we do want that for when the plans gets reused or also for when fragments of plan are moved around by AQE.
We don't really need to cache it for plans that are not roots of some execution pipeline (Stage).
I get your worry of if this would make us needlessly hold references to some throwaway intermediate RDDs that are just accidentally triggered... Hmm...
For anything that is then actually used in the Stages, references to it would be kept by the pipeline roots anyway... The question is whether there are some accidental calls to `execute` that will now keep these references even though it's not used at all... E.g. an `execute` call to a non-WSCG operator inside WSCG, creating and saving the non-WSCG RDD, while actual execution uses WSCG...
I don't think so, though such things happen via accident / non-intended side effect.
I think it's a bigger issue that Dataset holding references to QueryExecution, holding references to SparkPlans, holding references to other stuff results in a lot of references that impedes garbage collection.
E.g. https://github.com/apache/spark/blob/master/sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/ShuffleExchangeExec.scala#L257 leads to shuffle data not being cleaned up from disk while the user's program holds a live reference to Dataset.
Running in Spark Connect mode actually makes a lot of such issues go away, as all objects created during connect execution have lifetime that's finished after connect run finishes.
Here I'm trying to solve a different problem, where it's inconsistent what objects are reused vs. recreated from scratch, resulting in duplication of side effects.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48195] Save and reuse RDD/Broadcast created by SparkPlan [spark]
Posted by "juliuszsompolski (via GitHub)" <gi...@apache.org>.
juliuszsompolski commented on PR #48037:
URL: https://github.com/apache/spark/pull/48037#issuecomment-2358257233
Hm, it does seem that there is something wrong with `pyspark.ml.torch.tests.test_data_loader` and `pyspark.ml.tests.connect.test_parity_torch_data_loader`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48195] Save and reuse RDD/Broadcast created by SparkPlan [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on PR #48037:
URL: https://github.com/apache/spark/pull/48037#issuecomment-2360546886
@HyukjinKwon is there a way to increase timeout for these tests?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48195][CORE] Save and reuse RDD/Broadcast created by SparkPlan [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on PR #48037:
URL: https://github.com/apache/spark/pull/48037#issuecomment-2478376297
hmmm there are 4 suppressed exceptions for this simple query?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48195] Save and reuse RDD/Broadcast created by SparkPlan [spark]
Posted by "juliuszsompolski (via GitHub)" <gi...@apache.org>.
juliuszsompolski commented on PR #48037:
URL: https://github.com/apache/spark/pull/48037#issuecomment-2360714776
@cloud-fan this test seems to be genuinely hanging for 1+ h until the job timeouts, so there is something indeed going on; unfortunately I can't seem to reproduce it locally, as for me the test just fails with a PyTorch error... so most likely something is not set up in my environment...
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48195] Save and reuse RDD/Broadcast created by SparkPlan [spark]
Posted by "juliuszsompolski (via GitHub)" <gi...@apache.org>.
juliuszsompolski commented on PR #48037:
URL: https://github.com/apache/spark/pull/48037#issuecomment-2360953704
I merged in latest master. Thanks @HyukjinKwon !
As for my environment, I have a venv with python3.11, because that's what the CI seems to be using, I installed all dependencies with `pip install -r dev/requirements.txt`, and I am running the test with
```
python/run-tests --parallelism 1 --python-executables python3.11 --testnames pyspark.ml.torch.tests.test_data_loader
```
after doing `build/sbt package`.
For me the test fails after a moment with
```
File "/opt/homebrew/Cellar/python@3.11/3.11.9_1/Frameworks/Python.framework/Versions/3.11/lib/python3.11/multiprocessing/reduction.py", line 60, in dump
ForkingPickler(file, protocol).dump(obj)
AttributeError: Can't pickle local object '_SparkPartitionTorchDataset._get_field_converter.<locals>.converter'
...
torch.distributed.elastic.multiprocessing.errors.ChildFailedError:
============================================================
/Users/julek/dev/apache-spark/python/target/5fb69f79-c617-4c31-ae50-50ccd9b3f722/tmp0aee6vws/train.py FAILED
------------------------------------------------------------
Failures:
<NO_OTHER_FAILURES>
------------------------------------------------------------
Root Cause (first observed failure):
[0]:
time : 2024-09-19_15:09:01
host : QWKM9379T2
rank : 0 (local_rank: 0)
exitcode : 1 (pid: 53022)
error_file: <N/A>
traceback : To enable traceback see: https://pytorch.org/docs/stable/elastic/errors.html
============================================================
ERROR (44.113s)
...
py4j.protocol.Py4JJavaError: An error occurred while calling o75.collectToPython.
: org.apache.spark.SparkException: Job aborted due to stage failure: Could not recover from a failed barrier ResultStage. Most recent failure reason: Stage failed because barrier task ResultTask(0, 0) finished unsuccessfully.
...
RuntimeError: TorchDistributor failed during training.View stdout logs for detailed error message.
```
which doesn't seem to be related to this PR. But on this PR CI it doesn't fail like that and instead the test hangs...
Is there something that I'm missing in my environment?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48195][CORE] Save and reuse RDD/Broadcast created by SparkPlan [spark]
Posted by "juliuszsompolski (via GitHub)" <gi...@apache.org>.
juliuszsompolski commented on PR #48037:
URL: https://github.com/apache/spark/pull/48037#issuecomment-2368352989
If I see correctly, this has passed post-merge in https://github.com/apache/spark/actions/runs/10987281689/job/30501927292
```
Starting test(python3.11): pyspark.ml.tests.connect.test_parity_torch_data_loader (temp output: /__w/spark/spark/python/target/fde4d4f6-0ff0-4488-898e-7a1e5550a6c8/python3.11__pyspark.ml.tests.connect.test_parity_torch_data_loader__d41gd_ni.log)
Finished test(python3.11): pyspark.ml.tests.connect.test_parity_torch_data_loader (117s)
...
Starting test(python3.11): pyspark.ml.torch.tests.test_data_loader (temp output: /__w/spark/spark/python/target/5c48d675-3471-469c-85a7-b700118da4ec/python3.11__pyspark.ml.torch.tests.test_data_loader__mln4j8og.log)
Finished test(python3.11): pyspark.ml.torch.tests.test_data_loader (62s)
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48195] Save and reuse RDD/Broadcast created by SparkPlan [spark]
Posted by "juliuszsompolski (via GitHub)" <gi...@apache.org>.
juliuszsompolski commented on code in PR #48037:
URL: https://github.com/apache/spark/pull/48037#discussion_r1752051998
##########
core/src/test/scala/org/apache/spark/util/UtilsSuite.scala:
##########
@@ -1523,6 +1523,116 @@ class UtilsSuite extends SparkFunSuite with ResetSystemProperties {
conf.set(SERIALIZER, "org.apache.spark.serializer.JavaSerializer")
assert(Utils.isPushBasedShuffleEnabled(conf, isDriver = true) === false)
}
+
+
+ private def throwException(): String = {
+ throw new Exception("test")
+ }
+
+ private def callDoTry(): Try[String] = {
+ Utils.doTryWithCallerStacktrace {
+ throwException()
+ }
+ }
+
+ private def callGetTry(t: Try[String]): String = {
+ Utils.getTryWithCallerStacktrace(t)
+ }
+
+ private def callGetTryAgain(t: Try[String]): String = {
+ Utils.getTryWithCallerStacktrace(t)
+ }
+
+ test("doTryWithCallerStacktrace and getTryWithCallerStacktrace") {
+ val t = callDoTry()
+
+ val e1 = intercept[Exception] {
+ callGetTry(t)
+ }
+ // Uncomment for manual inspection
+ // e.printStackTrace()
+ // Example:
+ // java.lang.Exception: test
+ // at org.apache.spark.util.UtilsSuite.throwException(UtilsSuite.scala:1640)
+ // at org.apache.spark.util.UtilsSuite.$anonfun$callDoTry$1(UtilsSuite.scala:1645)
Review Comment:
`assert(!st1.exists(_.getMethodName == "callDoTry"))` -> callDoTry shouldn't be on the stack trace -> !st1.exists
`assert(st1.exists(_.getMethodName == "callGetTry"))` -> but callGetTry should be -> st1.exists
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48195] Save and reuse RDD/Broadcast created by SparkPlan [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #48037:
URL: https://github.com/apache/spark/pull/48037#discussion_r1752042209
##########
core/src/test/scala/org/apache/spark/util/UtilsSuite.scala:
##########
@@ -1523,6 +1523,116 @@ class UtilsSuite extends SparkFunSuite with ResetSystemProperties {
conf.set(SERIALIZER, "org.apache.spark.serializer.JavaSerializer")
assert(Utils.isPushBasedShuffleEnabled(conf, isDriver = true) === false)
}
+
+
+ private def throwException(): String = {
+ throw new Exception("test")
+ }
+
+ private def callDoTry(): Try[String] = {
+ Utils.doTryWithCallerStacktrace {
+ throwException()
+ }
+ }
+
+ private def callGetTry(t: Try[String]): String = {
+ Utils.getTryWithCallerStacktrace(t)
+ }
+
+ private def callGetTryAgain(t: Try[String]): String = {
+ Utils.getTryWithCallerStacktrace(t)
+ }
+
+ test("doTryWithCallerStacktrace and getTryWithCallerStacktrace") {
+ val t = callDoTry()
+
+ val e1 = intercept[Exception] {
+ callGetTry(t)
+ }
+ // Uncomment for manual inspection
+ // e.printStackTrace()
+ // Example:
+ // java.lang.Exception: test
+ // at org.apache.spark.util.UtilsSuite.throwException(UtilsSuite.scala:1640)
+ // at org.apache.spark.util.UtilsSuite.$anonfun$callDoTry$1(UtilsSuite.scala:1645)
Review Comment:
Sorry I'm confused, but the test says
```
// callDoTry shouldn't be on the stack trace, but callGetTry should be.
assert(!st1.exists(_.getMethodName == "callDoTry"))
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48195] Save and reuse RDD/Broadcast created by SparkPlan [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #48037:
URL: https://github.com/apache/spark/pull/48037#discussion_r1752046485
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/SparkPlan.scala:
##########
@@ -182,6 +182,10 @@ abstract class SparkPlan extends QueryPlan[SparkPlan] with Logging with Serializ
/** Specifies sort order for each partition requirements on the input data for this operator. */
def requiredChildOrdering: Seq[Seq[SortOrder]] = Seq.fill(children.size)(Nil)
+ private val executeRDD = LazyTry {
Review Comment:
ideally we only need to cache it for the root plan node?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48195] Save and reuse RDD/Broadcast created by SparkPlan [spark]
Posted by "juliuszsompolski (via GitHub)" <gi...@apache.org>.
juliuszsompolski commented on PR #48037:
URL: https://github.com/apache/spark/pull/48037#issuecomment-2358329661
Or actually I don't know... I don't know how to run the python environment correctly and it fails for me on master as well...
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48195] Save and reuse RDD/Broadcast created by SparkPlan [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #48037:
URL: https://github.com/apache/spark/pull/48037#discussion_r1752037079
##########
core/src/test/scala/org/apache/spark/util/LazyTrySuite.scala:
##########
@@ -0,0 +1,151 @@
+/*
+ * 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.util
+
+import org.apache.spark.SparkFunSuite
+
+class LazyTrySuite extends SparkFunSuite{
+ test("LazyTry should initialize only once") {
+ var count = 0
+ val lazyVal = LazyTry {
+ count += 1
+ count
+ }
+ assert(count == 0)
+ assert(lazyVal.get == 1)
+ assert(count == 1)
+ assert(lazyVal.get == 1)
+ assert(count == 1)
+ }
+
+ test("LazyTry should re-throw exceptions") {
+ val lazyVal = LazyTry {
+ throw new RuntimeException("test")
+ }
+ intercept[RuntimeException] {
+ lazyVal.get
+ }
+ intercept[RuntimeException] {
+ lazyVal.get
+ }
+ }
+
+ test("LazyTry should re-throw exceptions with current caller stack-trace") {
+ val fileName = Thread.currentThread().getStackTrace()(1).getFileName
+ val lineNo = Thread.currentThread().getStackTrace()(1).getLineNumber
+ val lazyVal = LazyTry {
+ throw new RuntimeException("test")
+ }
+
+ val e1 = intercept[RuntimeException] {
+ lazyVal.get // lineNo + 6
+ }
+ assert(e1.getStackTrace
+ .exists(elem => elem.getFileName == fileName && elem.getLineNumber == lineNo + 6))
+
+ val e2 = intercept[RuntimeException] {
+ lazyVal.get // lineNo + 12
+ }
+ assert(e2.getStackTrace
+ .exists(elem => elem.getFileName == fileName && elem.getLineNumber == lineNo + 12))
+ }
+
+ test("LazyTry does not lock containing object") {
+ class LazyContainer() {
+ @volatile var aSet = 0
+
+ val a: LazyTry[Int] = LazyTry {
+ aSet = 1
+ aSet
+ }
+
+ val b: LazyTry[Int] = LazyTry {
+ val t = new Thread(new Runnable {
+ override def run(): Unit = {
+ assert(a.get == 1)
+ }
+ })
+ t.start()
+ t.join()
+ aSet
+ }
+ }
+ val container = new LazyContainer()
+ // Nothing is lazy initialized yet
+ assert(container.aSet == 0)
+ // This will not deadlock, thread t will initialize a, and update aSet
+ assert(container.b.get == 1)
+ assert(container.aSet == 1)
+ }
+
+ // Scala lazy val tests are added to test for potential changes in the semantics of scala lazy val
Review Comment:
Do we really need them? If scala changes the lazy val semantic in the future, we will still stick with our own `LazyTry`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48195] Save and reuse RDD/Broadcast created by SparkPlan [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #48037:
URL: https://github.com/apache/spark/pull/48037#discussion_r1752038999
##########
core/src/test/scala/org/apache/spark/util/UtilsSuite.scala:
##########
@@ -1523,6 +1523,116 @@ class UtilsSuite extends SparkFunSuite with ResetSystemProperties {
conf.set(SERIALIZER, "org.apache.spark.serializer.JavaSerializer")
assert(Utils.isPushBasedShuffleEnabled(conf, isDriver = true) === false)
}
+
+
+ private def throwException(): String = {
+ throw new Exception("test")
+ }
+
+ private def callDoTry(): Try[String] = {
+ Utils.doTryWithCallerStacktrace {
+ throwException()
+ }
+ }
+
+ private def callGetTry(t: Try[String]): String = {
+ Utils.getTryWithCallerStacktrace(t)
+ }
+
+ private def callGetTryAgain(t: Try[String]): String = {
+ Utils.getTryWithCallerStacktrace(t)
+ }
+
+ test("doTryWithCallerStacktrace and getTryWithCallerStacktrace") {
+ val t = callDoTry()
+
+ val e1 = intercept[Exception] {
+ callGetTry(t)
+ }
+ // Uncomment for manual inspection
+ // e.printStackTrace()
Review Comment:
```suggestion
// e1.printStackTrace()
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48195][CORE] Save and reuse RDD/Broadcast created by SparkPlan [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on PR #48037:
URL: https://github.com/apache/spark/pull/48037#issuecomment-2367165081
yea let's merge and see
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48195][CORE] Save and reuse RDD/Broadcast created by SparkPlan [spark]
Posted by "pavel0fadeev (via GitHub)" <gi...@apache.org>.
pavel0fadeev commented on PR #48037:
URL: https://github.com/apache/spark/pull/48037#issuecomment-2366981669
jfi: I tried this branch in my GH account and the workflow completed [successful](https://github.com/pavel0fadeev/spark/actions/runs/10982688435/job/30492618334)
++ These tests work fine for this branch in my local environment.
Maybe there was something wrong with the GH runners when you tried to run it?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48195][CORE] Save and reuse RDD/Broadcast created by SparkPlan [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon closed pull request #48037: [SPARK-48195][CORE] Save and reuse RDD/Broadcast created by SparkPlan
URL: https://github.com/apache/spark/pull/48037
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48195] Save and reuse RDD/Broadcast created by SparkPlan [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #48037:
URL: https://github.com/apache/spark/pull/48037#discussion_r1752023916
##########
core/src/main/scala/org/apache/spark/util/LazyTry.scala:
##########
@@ -0,0 +1,70 @@
+/*
+ * 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.util
+
+import scala.util.Try
+
+/**
+ * Wrapper utility for a lazy val, with two differences compared to scala behavior:
+ *
+ * 1. Non-retrying in case of failure. This wrapper stores the exception in a Try, and will re-throw
+ * it on the access to `get`.
+ * In scala, when a `lazy val` field initialization throws an exception, the field remains
+ * uninitialized, and initialization will be re-attempted on the next access. This also can lead
+ * to performance issues, needlessly computing something towards a failure, and also can lead to
+ * duplicated side effects.
+ *
+ * 2. Resolving locking issues.
+ * In scala, when a `lazy val` field is initialized, it grabs the synchronized lock on the
+ * enclosing object instance. This can lead both to performance issues, and deadlocks.
+ * For example:
+ * a) Thread 1 entered a synchronized method, grabbing a coarse lock on the parent object.
+ * b) Thread 2 get spawned off, and tries to initialize a lazy value on the same parent object
+ * This causes scala to also try to grab a lock on the parent object.
+ * c) If thread 1 waits for thread 2 to join, a deadlock occurs.
+ * This wrapper will only grab a lock on the wrapper itself, and not the parent object.
+ *
+ * @param initialize The block of code to initialize the lazy value.
+ * @tparam T type of the lazy value.
+ */
+private[spark] class LazyTry[T](initialize: => T) extends Serializable {
+ private lazy val tryT: Try[T] = Utils.doTryWithCallerStacktrace { initialize }
+
+ /**
+ * Get the lazy value. If the initialization block threw an exception, it will be re-thrown here.
+ * The exception will be re-thrown with the current caller's stacktrace.
+ * An exception with stack trace from when the exception was first thrown can be accessed with
+ * ```
+ * ex.getSuppressed.find { e =>
+ * e.getMessage == org.apache.spark.util.Utils.TRY_WITH_CALLER_STACKTRACE_FULL_STACKTRACE
+ * }
+ * ```
+ */
+ def get: T = Utils.getTryWithCallerStacktrace(tryT)
+}
+
+private[spark] object LazyTry {
+ /**
+ * Create a new Lazy instance.
Review Comment:
```suggestion
* Create a new LazyTry instance.
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48195] Save and reuse RDD/Broadcast created by SparkPlan [spark]
Posted by "juliuszsompolski (via GitHub)" <gi...@apache.org>.
juliuszsompolski commented on code in PR #48037:
URL: https://github.com/apache/spark/pull/48037#discussion_r1752047279
##########
core/src/test/scala/org/apache/spark/util/LazyTrySuite.scala:
##########
@@ -0,0 +1,151 @@
+/*
+ * 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.util
+
+import org.apache.spark.SparkFunSuite
+
+class LazyTrySuite extends SparkFunSuite{
+ test("LazyTry should initialize only once") {
+ var count = 0
+ val lazyVal = LazyTry {
+ count += 1
+ count
+ }
+ assert(count == 0)
+ assert(lazyVal.get == 1)
+ assert(count == 1)
+ assert(lazyVal.get == 1)
+ assert(count == 1)
+ }
+
+ test("LazyTry should re-throw exceptions") {
+ val lazyVal = LazyTry {
+ throw new RuntimeException("test")
+ }
+ intercept[RuntimeException] {
+ lazyVal.get
+ }
+ intercept[RuntimeException] {
+ lazyVal.get
+ }
+ }
+
+ test("LazyTry should re-throw exceptions with current caller stack-trace") {
+ val fileName = Thread.currentThread().getStackTrace()(1).getFileName
+ val lineNo = Thread.currentThread().getStackTrace()(1).getLineNumber
+ val lazyVal = LazyTry {
+ throw new RuntimeException("test")
+ }
+
+ val e1 = intercept[RuntimeException] {
+ lazyVal.get // lineNo + 6
+ }
+ assert(e1.getStackTrace
+ .exists(elem => elem.getFileName == fileName && elem.getLineNumber == lineNo + 6))
+
+ val e2 = intercept[RuntimeException] {
+ lazyVal.get // lineNo + 12
+ }
+ assert(e2.getStackTrace
+ .exists(elem => elem.getFileName == fileName && elem.getLineNumber == lineNo + 12))
+ }
+
+ test("LazyTry does not lock containing object") {
+ class LazyContainer() {
+ @volatile var aSet = 0
+
+ val a: LazyTry[Int] = LazyTry {
+ aSet = 1
+ aSet
+ }
+
+ val b: LazyTry[Int] = LazyTry {
+ val t = new Thread(new Runnable {
+ override def run(): Unit = {
+ assert(a.get == 1)
+ }
+ })
+ t.start()
+ t.join()
+ aSet
+ }
+ }
+ val container = new LazyContainer()
+ // Nothing is lazy initialized yet
+ assert(container.aSet == 0)
+ // This will not deadlock, thread t will initialize a, and update aSet
+ assert(container.b.get == 1)
+ assert(container.aSet == 1)
+ }
+
+ // Scala lazy val tests are added to test for potential changes in the semantics of scala lazy val
Review Comment:
There are other places that use plain `lazy val` in Spark. Maybe if it changes (as e.g. if Spark updates to scala 3, it leads to some hard to trace issues. E.g. I remember last year I had to spend some time debugging a strange issue because of behavior change in a minor grpc upgrade: https://github.com/apache/spark/pull/43962
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48195] Save and reuse RDD/Broadcast created by SparkPlan [spark]
Posted by "pavel0fadeev (via GitHub)" <gi...@apache.org>.
pavel0fadeev commented on PR #48037:
URL: https://github.com/apache/spark/pull/48037#issuecomment-2347163058
Can the cache at https://github.com/apache/spark/blob/master/sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/ShuffleExchangeExec.scala#L257 be removed if there is the cache on SparkPlan level?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48195] Save and reuse RDD/Broadcast created by SparkPlan [spark]
Posted by "juliuszsompolski (via GitHub)" <gi...@apache.org>.
juliuszsompolski commented on PR #48037:
URL: https://github.com/apache/spark/pull/48037#issuecomment-2356374821
@cloud-fan [Run / Build modules: pyspark-mllib, pyspark-ml, pyspark-ml-connect](https://github.com/juliuszsompolski/apache-spark/actions/runs/10903670029/job/30258373853#logs) timed out again after 3 hours, though it doesn't seem stuck, just making progress until timeout... I'll trigger yet again
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48195] Save and reuse RDD/Broadcast created by SparkPlan [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #48037:
URL: https://github.com/apache/spark/pull/48037#discussion_r1752024515
##########
core/src/main/scala/org/apache/spark/util/LazyTry.scala:
##########
@@ -0,0 +1,70 @@
+/*
+ * 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.util
+
+import scala.util.Try
+
+/**
+ * Wrapper utility for a lazy val, with two differences compared to scala behavior:
+ *
+ * 1. Non-retrying in case of failure. This wrapper stores the exception in a Try, and will re-throw
+ * it on the access to `get`.
+ * In scala, when a `lazy val` field initialization throws an exception, the field remains
+ * uninitialized, and initialization will be re-attempted on the next access. This also can lead
+ * to performance issues, needlessly computing something towards a failure, and also can lead to
+ * duplicated side effects.
+ *
+ * 2. Resolving locking issues.
+ * In scala, when a `lazy val` field is initialized, it grabs the synchronized lock on the
+ * enclosing object instance. This can lead both to performance issues, and deadlocks.
+ * For example:
+ * a) Thread 1 entered a synchronized method, grabbing a coarse lock on the parent object.
+ * b) Thread 2 get spawned off, and tries to initialize a lazy value on the same parent object
+ * This causes scala to also try to grab a lock on the parent object.
+ * c) If thread 1 waits for thread 2 to join, a deadlock occurs.
+ * This wrapper will only grab a lock on the wrapper itself, and not the parent object.
+ *
+ * @param initialize The block of code to initialize the lazy value.
+ * @tparam T type of the lazy value.
+ */
+private[spark] class LazyTry[T](initialize: => T) extends Serializable {
+ private lazy val tryT: Try[T] = Utils.doTryWithCallerStacktrace { initialize }
+
+ /**
+ * Get the lazy value. If the initialization block threw an exception, it will be re-thrown here.
+ * The exception will be re-thrown with the current caller's stacktrace.
+ * An exception with stack trace from when the exception was first thrown can be accessed with
+ * ```
+ * ex.getSuppressed.find { e =>
+ * e.getMessage == org.apache.spark.util.Utils.TRY_WITH_CALLER_STACKTRACE_FULL_STACKTRACE
+ * }
+ * ```
+ */
+ def get: T = Utils.getTryWithCallerStacktrace(tryT)
+}
+
+private[spark] object LazyTry {
+ /**
+ * Create a new Lazy instance.
+ *
+ * @param initialize The block of code to initialize the lazy value.
+ * @tparam T type of the lazy value.
+ * @return a new Lazy instance.
Review Comment:
```suggestion
* @return a new LazyTry instance.
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48195] Save and reuse RDD/Broadcast created by SparkPlan [spark]
Posted by "juliuszsompolski (via GitHub)" <gi...@apache.org>.
juliuszsompolski commented on PR #48037:
URL: https://github.com/apache/spark/pull/48037#issuecomment-2353240267
I changed it in ShuffleExchangeExec.
InMemoryTableScanExec is the only other place, I would have to take a closer look there.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48195][CORE] Save and reuse RDD/Broadcast created by SparkPlan [spark]
Posted by "yaooqinn (via GitHub)" <gi...@apache.org>.
yaooqinn commented on PR #48037:
URL: https://github.com/apache/spark/pull/48037#issuecomment-2478002391
Instead of adding them to the suppressed field repeatedly, how about enrolling them in the stack traces of the current exception?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48195][CORE] Save and reuse RDD/Broadcast created by SparkPlan [spark]
Posted by "yaooqinn (via GitHub)" <gi...@apache.org>.
yaooqinn commented on PR #48037:
URL: https://github.com/apache/spark/pull/48037#issuecomment-2478403269
It looks like the number of suppressed can be calculated by `2 * calls of Utils.doTryWithCallerStacktrace`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48195][CORE] Save and reuse RDD/Broadcast created by SparkPlan [spark]
Posted by "juliuszsompolski (via GitHub)" <gi...@apache.org>.
juliuszsompolski commented on PR #48037:
URL: https://github.com/apache/spark/pull/48037#issuecomment-2367628341
Thank you @pavel0fadeev @cloud-fan @HyukjinKwon !
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48195][CORE] Save and reuse RDD/Broadcast created by SparkPlan [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on PR #48037:
URL: https://github.com/apache/spark/pull/48037#issuecomment-2367176952
Merged to master.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48195] Save and reuse RDD/Broadcast created by SparkPlan [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on PR #48037:
URL: https://github.com/apache/spark/pull/48037#issuecomment-2360760186
@juliuszsompolski mind syncing to the latest master brunch? I can take a look tmr
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48195] Save and reuse RDD/Broadcast created by SparkPlan [spark]
Posted by "juliuszsompolski (via GitHub)" <gi...@apache.org>.
juliuszsompolski commented on PR #48037:
URL: https://github.com/apache/spark/pull/48037#issuecomment-2355720934
https://github.com/juliuszsompolski/apache-spark/actions/runs/10887100133/job/30209339909 timed out. Retrigger
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48195] Save and reuse RDD/Broadcast created by SparkPlan [spark]
Posted by "juliuszsompolski (via GitHub)" <gi...@apache.org>.
juliuszsompolski commented on code in PR #48037:
URL: https://github.com/apache/spark/pull/48037#discussion_r1750544399
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/CollectMetricsExec.scala:
##########
@@ -60,6 +60,11 @@ case class CollectMetricsExec(
override def outputOrdering: Seq[SortOrder] = child.outputOrdering
+ override def resetMetrics(): Unit = {
+ accumulator.reset()
+ super.resetMetrics()
+ }
Review Comment:
because RDD returned by doExecute is reused, accumulator needs to be reset here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48195] Save and reuse RDD/Broadcast created by SparkPlan [spark]
Posted by "JoshRosen (via GitHub)" <gi...@apache.org>.
JoshRosen commented on code in PR #48037:
URL: https://github.com/apache/spark/pull/48037#discussion_r1751032754
##########
core/src/main/scala/org/apache/spark/util/Lazy.scala:
##########
@@ -0,0 +1,61 @@
+/*
+ * 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.util
+
+import scala.util.Try
+
+/**
+ * Wrapper utility for lazy val, with two differences compared to scala behavir:
+ *
+ * 1. In scala, when a `lazy val` field is initialized, it grabs the synchronized lock on the
+ * enclosing object instance. This can lead both to performance issues, and deadlocks.
+ * For example:
+ * a) Thread 1 entered a synchronized method, grabbing a coarse lock on the parent object.
+ * b) Thread 2 get spawned off, and tries to initialize a lazy value on the same parent object
+ * This causes scala to also try to grab a lock on the parent object.
+ * c) If thread 1 waits for thread 2 to join, a deadlock occurs.
+ * This Lazy wrapper will only grab a lock on the wrapper itself, and not the parent object.
+ *
+ * 2. In scala, when a `lazy val` field initialization throws an exception, the field remains
+ * uninitialized, and initialization will be re-attempted on the next access. This also can lead
+ * to performance issues, needlessly computing something towards a failure, and also can lead to
+ * duplicated side effects.
+ * This Lazy wrapper stores the exception in a Try, and will re-throw it on the next access.
+ *
+ * @param initialize The block of code to initialize the lazy value.
+ * @tparam T type of the lazy value.
+ */
+private[spark] class Lazy[T](initialize: => T) extends Serializable {
+ private lazy val tryT: Try[T] = Utils.doTryWithCallerStacktrace { initialize }
Review Comment:
I am slightly concerned that having this somewhat implicitly be a `Try` (implicit in the sense that it's not obvious that it's a try just from looking at a usage like `val x: Lazy[Foo] ...`) might lead to the introduction of unintentional "fate sharing" or "negative caching" bugs related to query cancellation: if one thread is computing a lazy value and is interrupted, the InterruptedException might be cached and doom all future calls.
Whether or not that's actually a problem is going to be context-of-use dependent.
Since this semantic differs from other similar `Lazy` wrapper classes that I've seen in other codebases, I think it might make sense to make the failure caching semantic more explicit, e.g. by naming this something like `LazyNonRetrying` or `NonRetryingLazy`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48195] Save and reuse RDD/Broadcast created by SparkPlan [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on PR #48037:
URL: https://github.com/apache/spark/pull/48037#issuecomment-2351463500
Yea, I think all physical nodes no longer need to cache RDDs explicitly as the framework does it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48195][CORE] Save and reuse RDD/Broadcast created by SparkPlan [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on PR #48037:
URL: https://github.com/apache/spark/pull/48037#issuecomment-2362532403
btw I think it's not just ser/de problem. ser/de issue can be worked around via 959a89f2811b7508a385e75d4e8e3ee893197caa but the issue still persists
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48195][CORE] Save and reuse RDD/Broadcast created by SparkPlan [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #48037:
URL: https://github.com/apache/spark/pull/48037#discussion_r1768620829
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/SparkPlan.scala:
##########
@@ -182,6 +182,10 @@ abstract class SparkPlan extends QueryPlan[SparkPlan] with Logging with Serializ
/** Specifies sort order for each partition requirements on the input data for this operator. */
def requiredChildOrdering: Seq[Seq[SortOrder]] = Seq.fill(children.size)(Nil)
+ private val executeRDD = LazyTry {
Review Comment:
```suggestion
@transient
private val executeRDD = LazyTry {
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48195][CORE] Save and reuse RDD/Broadcast created by SparkPlan [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on PR #48037:
URL: https://github.com/apache/spark/pull/48037#issuecomment-2367017195
Let me get this in and see if it passes. Otherwise, I will revert it back.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48195][CORE] Save and reuse RDD/Broadcast created by SparkPlan [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on PR #48037:
URL: https://github.com/apache/spark/pull/48037#issuecomment-2367017333
Does it LG to you, @cloud-fan @JoshRosen ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48195][CORE] Save and reuse RDD/Broadcast created by SparkPlan [spark]
Posted by "juliuszsompolski (via GitHub)" <gi...@apache.org>.
juliuszsompolski commented on PR #48037:
URL: https://github.com/apache/spark/pull/48037#issuecomment-2483100505
@yaooqinn @cloud-fan
My intention here was that for debuggability, the user may need both the stack trace of the error where it originally happened (as stored in the Try and what Try normally returns), and the current stack trace when the error is "retriggered" by accessing the Try again.
The two suppressed exceptions are needed for that: `TRY_WITH_CALLER_STACKTRACE_FULL_STACKTRACE` is the full original stacktrace when the stuff inside the Try was actually executed. `TRY_WITH_CALLER_STACKTRACE_TRY_STACKTRACE` is a prefix of the stack trace that is under `doTryWithCallerStacktrace` of the caller, which is needed so that `getTryWithCallerStacktrace` could stitch together the stack trace of the new caller with the original caller. I initially made it find `doTryWithCallerStacktrace` in there, but that doesn't work in recursive situations like is actually the case here.
`TRY_WITH_CALLER_STACKTRACE_TRY_STACKTRACE` does not need to be visible and could be hidden...
Maybe the `val fullEx = new Exception(TRY_WITH_CALLER_STACKTRACE_FULL_STACKTRACE)` instead of being a generic Exception, could be a special exception type that just holds the length of that prefix. Let me do that.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48195][CORE] Save and reuse RDD/Broadcast created by SparkPlan [spark]
Posted by "juliuszsompolski (via GitHub)" <gi...@apache.org>.
juliuszsompolski commented on PR #48037:
URL: https://github.com/apache/spark/pull/48037#issuecomment-2483356362
> Instead of adding them to the suppressed field repeatedly, how about enrolling them in the stack traces of the current exception?
The gist is that we want two stack traces:
- the one from the original execution when the exception happened
- the one from the current execution, that retrieved the exception.
The first one is in suppressed exception. The second one is already stitched into the stack trace of the rethrown exception. The third one (with TRY_WITH_CALLER_STACKTRACE_TRY_STACKTRACE) was only needed internally to do the stitching, and I remove it in https://github.com/apache/spark/pull/48882
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48195] Save and reuse RDD/Broadcast created by SparkPlan [spark]
Posted by "juliuszsompolski (via GitHub)" <gi...@apache.org>.
juliuszsompolski commented on PR #48037:
URL: https://github.com/apache/spark/pull/48037#issuecomment-2358463856
In my local environment I get `RuntimeError: TorchDistributor failed during training.View stdout logs for detailed error message.` on these tests both on master and in this PR branch; I can't seem to find more details, I use python3.11 and have all the dependencies from dev/requirements.txt installed... On this PR, this test seems to be hanging in CI.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48195][CORE] Save and reuse RDD/Broadcast created by SparkPlan [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on PR #48037:
URL: https://github.com/apache/spark/pull/48037#issuecomment-2362523667
@WeichenXu123 mind taking a look when you find some time?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48195] Save and reuse RDD/Broadcast created by SparkPlan [spark]
Posted by "pavel0fadeev (via GitHub)" <gi...@apache.org>.
pavel0fadeev commented on PR #48037:
URL: https://github.com/apache/spark/pull/48037#issuecomment-2362343583
@juliuszsompolski Regarding the issue with your environment in **master** branch.
I checked and it looks like I have the same problem when I run `"pyspark.ml.torch.tests.test_data_loader"` tests on my Mac with Python 3.11.
For me it seems that `"AttributeError: Can't pickle local object '_SparkPartitionTorchDataset._get_field_converter.<locals>.converter'"` is the main error message here and there is an article about this https://medium.com/devopss-hole/python-multiprocessing-pickle-issue-e2d35ccf96a9. In short, in some circumstances there is a problem in the **multiprocessing** library when using "spawn" multiprocessing_context, which is the default for Mac and Windows. All Unix OS except Mac have "fork" as the default multiprocessing context and this may explain why we see different behaviour for this test on some local environments and on Github runners.
So, **torch** library uses **multiprocessing** library in DataLoader and I tried to change multiprocessing_context to "fork" which helped me get over the initial exception but I got another one. Then I tried to set num_workers=0 for DataLoader to get rid of multiprocessing at all in this test:
https://github.com/apache/spark/blob/04455797bfb3631b13b41cfa5d2604db3bf8acc2/python/pyspark/ml/torch/tests/test_data_loader.py#L70
I changed it to
`data_loader = _get_spark_partition_data_loader(num_samples, batch_size, num_workers=0)`
after that the test completed successfully.
But I don't know how to change the environment to get rid of this issue without changing the code.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48195] Save and reuse RDD/Broadcast created by SparkPlan [spark]
Posted by "juliuszsompolski (via GitHub)" <gi...@apache.org>.
juliuszsompolski commented on PR #48037:
URL: https://github.com/apache/spark/pull/48037#issuecomment-2353261392
I changed it in InMemoryTableScanExec as well.
The underlying InMemoryRelation keeps the underlying buffers in _cachedColumnBuffers, but the conversions to RDD[InternalRow] or RDD[ColumnarBatch] can be cached by the framework.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48195][CORE] Save and reuse RDD/Broadcast created by SparkPlan [spark]
Posted by "yaooqinn (via GitHub)" <gi...@apache.org>.
yaooqinn commented on PR #48037:
URL: https://github.com/apache/spark/pull/48037#issuecomment-2477909517
Hi, can we improve a bit of the stacktrace? It's looking very noisy now
```
spark-sql (default)> SELECT make_timestamp(2021, 07, 11, 6, 30, 60.007);
[INVALID_FRACTION_OF_SECOND] Valid range for seconds is [0, 60] (inclusive), but the provided value is 60.007000. To avoid this error, use `try_make_timestamp`, which returns NULL on error.
If you do not want to use the session default timestamp version of this function, use `try_make_timestamp_ntz` or `try_make_timestamp_ltz`. SQLSTATE: 22023
org.apache.spark.SparkDateTimeException: [INVALID_FRACTION_OF_SECOND] Valid range for seconds is [0, 60] (inclusive), but the provided value is 60.007000. To avoid this error, use `try_make_timestamp`, which returns NULL on error.
If you do not want to use the session default timestamp version of this function, use `try_make_timestamp_ntz` or `try_make_timestamp_ltz`. SQLSTATE: 22023
at org.apache.spark.sql.errors.QueryExecutionErrors$.invalidFractionOfSecondError(QueryExecutionErrors.scala:267)
at org.apache.spark.sql.catalyst.expressions.MakeTimestamp.toMicros(datetimeExpressions.scala:2811)
at org.apache.spark.sql.catalyst.expressions.MakeTimestamp.nullSafeEval(datetimeExpressions.scala:2847)
at org.apache.spark.sql.catalyst.expressions.SeptenaryExpression.eval(Expression.scala:1184)
at org.apache.spark.sql.catalyst.optimizer.ConstantFolding$.org$apache$spark$sql$catalyst$optimizer$ConstantFolding$$constantFolding(expressions.scala:82)
at org.apache.spark.sql.catalyst.optimizer.ConstantFolding$.$anonfun$constantFolding$4(expressions.scala:102)
at org.apache.spark.sql.catalyst.trees.UnaryLike.mapChildren(TreeNode.scala:1231)
at org.apache.spark.sql.catalyst.trees.UnaryLike.mapChildren$(TreeNode.scala:1230)
at org.apache.spark.sql.catalyst.expressions.UnaryExpression.mapChildren(Expression.scala:553)
at org.apache.spark.sql.catalyst.optimizer.ConstantFolding$.org$apache$spark$sql$catalyst$optimizer$ConstantFolding$$constantFolding(expressions.scala:102)
at org.apache.spark.sql.catalyst.optimizer.ConstantFolding$$anonfun$apply$1.$anonfun$applyOrElse$1(expressions.scala:106)
at org.apache.spark.sql.catalyst.plans.QueryPlan.$anonfun$mapExpressions$1(QueryPlan.scala:223)
at org.apache.spark.sql.catalyst.trees.CurrentOrigin$.withOrigin(origin.scala:86)
at org.apache.spark.sql.catalyst.plans.QueryPlan.transformExpression$1(QueryPlan.scala:223)
at org.apache.spark.sql.catalyst.plans.QueryPlan.recursiveTransform$1(QueryPlan.scala:235)
at org.apache.spark.sql.catalyst.plans.QueryPlan.$anonfun$mapExpressions$4(QueryPlan.scala:241)
at scala.collection.immutable.List.map(List.scala:247)
at scala.collection.immutable.List.map(List.scala:79)
at org.apache.spark.sql.catalyst.plans.QueryPlan.recursiveTransform$1(QueryPlan.scala:241)
at org.apache.spark.sql.catalyst.plans.QueryPlan.$anonfun$mapExpressions$5(QueryPlan.scala:246)
at org.apache.spark.sql.catalyst.trees.TreeNode.mapProductIterator(TreeNode.scala:312)
at org.apache.spark.sql.catalyst.plans.QueryPlan.mapExpressions(QueryPlan.scala:246)
at org.apache.spark.sql.catalyst.optimizer.ConstantFolding$$anonfun$apply$1.applyOrElse(expressions.scala:106)
at org.apache.spark.sql.catalyst.optimizer.ConstantFolding$$anonfun$apply$1.applyOrElse(expressions.scala:105)
at org.apache.spark.sql.catalyst.trees.TreeNode.$anonfun$transformDownWithPruning$1(TreeNode.scala:470)
at org.apache.spark.sql.catalyst.trees.CurrentOrigin$.withOrigin(origin.scala:86)
at org.apache.spark.sql.catalyst.trees.TreeNode.transformDownWithPruning(TreeNode.scala:470)
at org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.org$apache$spark$sql$catalyst$plans$logical$AnalysisHelper$$super$transformDownWithPruning(LogicalPlan.scala:37)
at org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper.transformDownWithPruning(AnalysisHelper.scala:330)
at org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper.transformDownWithPruning$(AnalysisHelper.scala:326)
at org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.transformDownWithPruning(LogicalPlan.scala:37)
at org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.transformDownWithPruning(LogicalPlan.scala:37)
at org.apache.spark.sql.catalyst.trees.TreeNode.transformWithPruning(TreeNode.scala:436)
at org.apache.spark.sql.catalyst.optimizer.ConstantFolding$.apply(expressions.scala:105)
at org.apache.spark.sql.catalyst.optimizer.ConstantFolding$.apply(expressions.scala:48)
at org.apache.spark.sql.catalyst.rules.RuleExecutor.$anonfun$execute$2(RuleExecutor.scala:226)
at scala.collection.LinearSeqOps.foldLeft(LinearSeq.scala:183)
at scala.collection.LinearSeqOps.foldLeft$(LinearSeq.scala:179)
at scala.collection.immutable.List.foldLeft(List.scala:79)
at org.apache.spark.sql.catalyst.rules.RuleExecutor.$anonfun$execute$1(RuleExecutor.scala:223)
at org.apache.spark.sql.catalyst.rules.RuleExecutor.$anonfun$execute$1$adapted(RuleExecutor.scala:215)
at scala.collection.immutable.List.foreach(List.scala:334)
at org.apache.spark.sql.catalyst.rules.RuleExecutor.execute(RuleExecutor.scala:215)
at org.apache.spark.sql.catalyst.rules.RuleExecutor.$anonfun$executeAndTrack$1(RuleExecutor.scala:186)
at org.apache.spark.sql.catalyst.QueryPlanningTracker$.withTracker(QueryPlanningTracker.scala:89)
at org.apache.spark.sql.catalyst.rules.RuleExecutor.executeAndTrack(RuleExecutor.scala:186)
at org.apache.spark.sql.execution.QueryExecution.$anonfun$lazyOptimizedPlan$2(QueryExecution.scala:179)
at org.apache.spark.sql.catalyst.QueryPlanningTracker.measurePhase(QueryPlanningTracker.scala:138)
at org.apache.spark.sql.execution.QueryExecution.$anonfun$executePhase$2(QueryExecution.scala:255)
at org.apache.spark.sql.execution.QueryExecution$.withInternalError(QueryExecution.scala:631)
at org.apache.spark.sql.execution.QueryExecution.$anonfun$executePhase$1(QueryExecution.scala:255)
at org.apache.spark.sql.SparkSession.withActive(SparkSession.scala:752)
at org.apache.spark.sql.execution.QueryExecution.executePhase(QueryExecution.scala:254)
at org.apache.spark.sql.execution.QueryExecution.$anonfun$lazyOptimizedPlan$1(QueryExecution.scala:175)
at scala.util.Try$.apply(Try.scala:217)
at org.apache.spark.util.Utils$.doTryWithCallerStacktrace(Utils.scala:1375)
at org.apache.spark.util.Utils$.getTryWithCallerStacktrace(Utils.scala:1429)
at org.apache.spark.util.LazyTry.get(LazyTry.scala:58)
at org.apache.spark.sql.execution.QueryExecution.executedPlan(QueryExecution.scala:225)
at org.apache.spark.sql.execution.QueryExecution.simpleString(QueryExecution.scala:274)
at org.apache.spark.sql.execution.QueryExecution.org$apache$spark$sql$execution$QueryExecution$$explainString(QueryExecution.scala:321)
at org.apache.spark.sql.execution.QueryExecution.explainString(QueryExecution.scala:289)
at org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId0$7(SQLExecution.scala:147)
at org.apache.spark.sql.execution.SQLExecution$.withSessionTagsApplied(SQLExecution.scala:264)
at org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId0$6(SQLExecution.scala:123)
at org.apache.spark.sql.execution.SQLExecution$.withSQLConfPropagated(SQLExecution.scala:287)
at org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId0$1(SQLExecution.scala:123)
at org.apache.spark.sql.SparkSession.withActive(SparkSession.scala:752)
at org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId0(SQLExecution.scala:77)
at org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:229)
at org.apache.spark.sql.hive.thriftserver.SparkSQLDriver.run(SparkSQLDriver.scala:77)
at org.apache.spark.sql.hive.thriftserver.SparkSQLCLIDriver.processCmd(SparkSQLCLIDriver.scala:461)
at org.apache.spark.sql.hive.thriftserver.SparkSQLCLIDriver.$anonfun$processLine$1(SparkSQLCLIDriver.scala:579)
at org.apache.spark.sql.hive.thriftserver.SparkSQLCLIDriver.$anonfun$processLine$1$adapted(SparkSQLCLIDriver.scala:573)
at scala.collection.IterableOnceOps.foreach(IterableOnce.scala:619)
at scala.collection.IterableOnceOps.foreach$(IterableOnce.scala:617)
at scala.collection.AbstractIterable.foreach(Iterable.scala:935)
at org.apache.spark.sql.hive.thriftserver.SparkSQLCLIDriver.processLine(SparkSQLCLIDriver.scala:573)
at org.apache.spark.sql.hive.thriftserver.SparkSQLCLIDriver$.main(SparkSQLCLIDriver.scala:288)
at org.apache.spark.sql.hive.thriftserver.SparkSQLCLIDriver.main(SparkSQLCLIDriver.scala)
at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:77)
at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.base/java.lang.reflect.Method.invoke(Method.java:569)
at org.apache.spark.deploy.JavaMainApplication.start(SparkApplication.scala:52)
at org.apache.spark.deploy.SparkSubmit.org$apache$spark$deploy$SparkSubmit$$runMain(SparkSubmit.scala:1032)
at org.apache.spark.deploy.SparkSubmit.doRunMain$1(SparkSubmit.scala:204)
at org.apache.spark.deploy.SparkSubmit.submit(SparkSubmit.scala:227)
at org.apache.spark.deploy.SparkSubmit.doSubmit(SparkSubmit.scala:96)
at org.apache.spark.deploy.SparkSubmit$$anon$2.doSubmit(SparkSubmit.scala:1137)
at org.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:1146)
at org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala)
Suppressed: java.lang.Exception: Stacktrace under doTryWithCallerStacktrace
at org.apache.spark.sql.errors.QueryExecutionErrors$.invalidFractionOfSecondError(QueryExecutionErrors.scala:267)
at org.apache.spark.sql.catalyst.expressions.MakeTimestamp.toMicros(datetimeExpressions.scala:2811)
at org.apache.spark.sql.catalyst.expressions.MakeTimestamp.nullSafeEval(datetimeExpressions.scala:2847)
at org.apache.spark.sql.catalyst.expressions.SeptenaryExpression.eval(Expression.scala:1184)
at org.apache.spark.sql.catalyst.optimizer.ConstantFolding$.org$apache$spark$sql$catalyst$optimizer$ConstantFolding$$constantFolding(expressions.scala:82)
at org.apache.spark.sql.catalyst.optimizer.ConstantFolding$.$anonfun$constantFolding$4(expressions.scala:102)
at org.apache.spark.sql.catalyst.trees.UnaryLike.mapChildren(TreeNode.scala:1231)
at org.apache.spark.sql.catalyst.trees.UnaryLike.mapChildren$(TreeNode.scala:1230)
at org.apache.spark.sql.catalyst.expressions.UnaryExpression.mapChildren(Expression.scala:553)
at org.apache.spark.sql.catalyst.optimizer.ConstantFolding$.org$apache$spark$sql$catalyst$optimizer$ConstantFolding$$constantFolding(expressions.scala:102)
at org.apache.spark.sql.catalyst.optimizer.ConstantFolding$$anonfun$apply$1.$anonfun$applyOrElse$1(expressions.scala:106)
at org.apache.spark.sql.catalyst.plans.QueryPlan.$anonfun$mapExpressions$1(QueryPlan.scala:223)
at org.apache.spark.sql.catalyst.trees.CurrentOrigin$.withOrigin(origin.scala:86)
at org.apache.spark.sql.catalyst.plans.QueryPlan.transformExpression$1(QueryPlan.scala:223)
at org.apache.spark.sql.catalyst.plans.QueryPlan.recursiveTransform$1(QueryPlan.scala:235)
at org.apache.spark.sql.catalyst.plans.QueryPlan.$anonfun$mapExpressions$4(QueryPlan.scala:241)
at scala.collection.immutable.List.map(List.scala:247)
at scala.collection.immutable.List.map(List.scala:79)
at org.apache.spark.sql.catalyst.plans.QueryPlan.recursiveTransform$1(QueryPlan.scala:241)
at org.apache.spark.sql.catalyst.plans.QueryPlan.$anonfun$mapExpressions$5(QueryPlan.scala:246)
at org.apache.spark.sql.catalyst.trees.TreeNode.mapProductIterator(TreeNode.scala:312)
at org.apache.spark.sql.catalyst.plans.QueryPlan.mapExpressions(QueryPlan.scala:246)
at org.apache.spark.sql.catalyst.optimizer.ConstantFolding$$anonfun$apply$1.applyOrElse(expressions.scala:106)
at org.apache.spark.sql.catalyst.optimizer.ConstantFolding$$anonfun$apply$1.applyOrElse(expressions.scala:105)
at org.apache.spark.sql.catalyst.trees.TreeNode.$anonfun$transformDownWithPruning$1(TreeNode.scala:470)
at org.apache.spark.sql.catalyst.trees.CurrentOrigin$.withOrigin(origin.scala:86)
at org.apache.spark.sql.catalyst.trees.TreeNode.transformDownWithPruning(TreeNode.scala:470)
at org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.org$apache$spark$sql$catalyst$plans$logical$AnalysisHelper$$super$transformDownWithPruning(LogicalPlan.scala:37)
at org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper.transformDownWithPruning(AnalysisHelper.scala:330)
at org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper.transformDownWithPruning$(AnalysisHelper.scala:326)
at org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.transformDownWithPruning(LogicalPlan.scala:37)
at org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.transformDownWithPruning(LogicalPlan.scala:37)
at org.apache.spark.sql.catalyst.trees.TreeNode.transformWithPruning(TreeNode.scala:436)
at org.apache.spark.sql.catalyst.optimizer.ConstantFolding$.apply(expressions.scala:105)
at org.apache.spark.sql.catalyst.optimizer.ConstantFolding$.apply(expressions.scala:48)
at org.apache.spark.sql.catalyst.rules.RuleExecutor.$anonfun$execute$2(RuleExecutor.scala:226)
at scala.collection.LinearSeqOps.foldLeft(LinearSeq.scala:183)
at scala.collection.LinearSeqOps.foldLeft$(LinearSeq.scala:179)
at scala.collection.immutable.List.foldLeft(List.scala:79)
at org.apache.spark.sql.catalyst.rules.RuleExecutor.$anonfun$execute$1(RuleExecutor.scala:223)
at org.apache.spark.sql.catalyst.rules.RuleExecutor.$anonfun$execute$1$adapted(RuleExecutor.scala:215)
at scala.collection.immutable.List.foreach(List.scala:334)
at org.apache.spark.sql.catalyst.rules.RuleExecutor.execute(RuleExecutor.scala:215)
at org.apache.spark.sql.catalyst.rules.RuleExecutor.$anonfun$executeAndTrack$1(RuleExecutor.scala:186)
at org.apache.spark.sql.catalyst.QueryPlanningTracker$.withTracker(QueryPlanningTracker.scala:89)
at org.apache.spark.sql.catalyst.rules.RuleExecutor.executeAndTrack(RuleExecutor.scala:186)
at org.apache.spark.sql.execution.QueryExecution.$anonfun$lazyOptimizedPlan$2(QueryExecution.scala:179)
at org.apache.spark.sql.catalyst.QueryPlanningTracker.measurePhase(QueryPlanningTracker.scala:138)
at org.apache.spark.sql.execution.QueryExecution.$anonfun$executePhase$2(QueryExecution.scala:255)
at org.apache.spark.sql.execution.QueryExecution$.withInternalError(QueryExecution.scala:631)
at org.apache.spark.sql.execution.QueryExecution.$anonfun$executePhase$1(QueryExecution.scala:255)
at org.apache.spark.sql.SparkSession.withActive(SparkSession.scala:752)
at org.apache.spark.sql.execution.QueryExecution.executePhase(QueryExecution.scala:254)
at org.apache.spark.sql.execution.QueryExecution.$anonfun$lazyOptimizedPlan$1(QueryExecution.scala:175)
at scala.util.Try$.apply(Try.scala:217)
at org.apache.spark.util.Utils$.doTryWithCallerStacktrace(Utils.scala:1375)
Suppressed: java.lang.Exception: Full stacktrace of original doTryWithCallerStacktrace caller
at org.apache.spark.sql.errors.QueryExecutionErrors$.invalidFractionOfSecondError(QueryExecutionErrors.scala:267)
at org.apache.spark.sql.catalyst.expressions.MakeTimestamp.toMicros(datetimeExpressions.scala:2811)
at org.apache.spark.sql.catalyst.expressions.MakeTimestamp.nullSafeEval(datetimeExpressions.scala:2847)
at org.apache.spark.sql.catalyst.expressions.SeptenaryExpression.eval(Expression.scala:1184)
at org.apache.spark.sql.catalyst.optimizer.ConstantFolding$.org$apache$spark$sql$catalyst$optimizer$ConstantFolding$$constantFolding(expressions.scala:82)
at org.apache.spark.sql.catalyst.optimizer.ConstantFolding$.$anonfun$constantFolding$4(expressions.scala:102)
at org.apache.spark.sql.catalyst.trees.UnaryLike.mapChildren(TreeNode.scala:1231)
at org.apache.spark.sql.catalyst.trees.UnaryLike.mapChildren$(TreeNode.scala:1230)
at org.apache.spark.sql.catalyst.expressions.UnaryExpression.mapChildren(Expression.scala:553)
at org.apache.spark.sql.catalyst.optimizer.ConstantFolding$.org$apache$spark$sql$catalyst$optimizer$ConstantFolding$$constantFolding(expressions.scala:102)
at org.apache.spark.sql.catalyst.optimizer.ConstantFolding$$anonfun$apply$1.$anonfun$applyOrElse$1(expressions.scala:106)
at org.apache.spark.sql.catalyst.plans.QueryPlan.$anonfun$mapExpressions$1(QueryPlan.scala:223)
at org.apache.spark.sql.catalyst.trees.CurrentOrigin$.withOrigin(origin.scala:86)
at org.apache.spark.sql.catalyst.plans.QueryPlan.transformExpression$1(QueryPlan.scala:223)
at org.apache.spark.sql.catalyst.plans.QueryPlan.recursiveTransform$1(QueryPlan.scala:235)
at org.apache.spark.sql.catalyst.plans.QueryPlan.$anonfun$mapExpressions$4(QueryPlan.scala:241)
at scala.collection.immutable.List.map(List.scala:247)
at scala.collection.immutable.List.map(List.scala:79)
at org.apache.spark.sql.catalyst.plans.QueryPlan.recursiveTransform$1(QueryPlan.scala:241)
at org.apache.spark.sql.catalyst.plans.QueryPlan.$anonfun$mapExpressions$5(QueryPlan.scala:246)
at org.apache.spark.sql.catalyst.trees.TreeNode.mapProductIterator(TreeNode.scala:312)
at org.apache.spark.sql.catalyst.plans.QueryPlan.mapExpressions(QueryPlan.scala:246)
at org.apache.spark.sql.catalyst.optimizer.ConstantFolding$$anonfun$apply$1.applyOrElse(expressions.scala:106)
at org.apache.spark.sql.catalyst.optimizer.ConstantFolding$$anonfun$apply$1.applyOrElse(expressions.scala:105)
at org.apache.spark.sql.catalyst.trees.TreeNode.$anonfun$transformDownWithPruning$1(TreeNode.scala:470)
at org.apache.spark.sql.catalyst.trees.CurrentOrigin$.withOrigin(origin.scala:86)
at org.apache.spark.sql.catalyst.trees.TreeNode.transformDownWithPruning(TreeNode.scala:470)
at org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.org$apache$spark$sql$catalyst$plans$logical$AnalysisHelper$$super$transformDownWithPruning(LogicalPlan.scala:37)
at org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper.transformDownWithPruning(AnalysisHelper.scala:330)
at org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper.transformDownWithPruning$(AnalysisHelper.scala:326)
at org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.transformDownWithPruning(LogicalPlan.scala:37)
at org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.transformDownWithPruning(LogicalPlan.scala:37)
at org.apache.spark.sql.catalyst.trees.TreeNode.transformWithPruning(TreeNode.scala:436)
at org.apache.spark.sql.catalyst.optimizer.ConstantFolding$.apply(expressions.scala:105)
at org.apache.spark.sql.catalyst.optimizer.ConstantFolding$.apply(expressions.scala:48)
at org.apache.spark.sql.catalyst.rules.RuleExecutor.$anonfun$execute$2(RuleExecutor.scala:226)
at scala.collection.LinearSeqOps.foldLeft(LinearSeq.scala:183)
at scala.collection.LinearSeqOps.foldLeft$(LinearSeq.scala:179)
at scala.collection.immutable.List.foldLeft(List.scala:79)
at org.apache.spark.sql.catalyst.rules.RuleExecutor.$anonfun$execute$1(RuleExecutor.scala:223)
at org.apache.spark.sql.catalyst.rules.RuleExecutor.$anonfun$execute$1$adapted(RuleExecutor.scala:215)
at scala.collection.immutable.List.foreach(List.scala:334)
at org.apache.spark.sql.catalyst.rules.RuleExecutor.execute(RuleExecutor.scala:215)
at org.apache.spark.sql.catalyst.rules.RuleExecutor.$anonfun$executeAndTrack$1(RuleExecutor.scala:186)
at org.apache.spark.sql.catalyst.QueryPlanningTracker$.withTracker(QueryPlanningTracker.scala:89)
at org.apache.spark.sql.catalyst.rules.RuleExecutor.executeAndTrack(RuleExecutor.scala:186)
at org.apache.spark.sql.execution.QueryExecution.$anonfun$lazyOptimizedPlan$2(QueryExecution.scala:179)
at org.apache.spark.sql.catalyst.QueryPlanningTracker.measurePhase(QueryPlanningTracker.scala:138)
at org.apache.spark.sql.execution.QueryExecution.$anonfun$executePhase$2(QueryExecution.scala:255)
at org.apache.spark.sql.execution.QueryExecution$.withInternalError(QueryExecution.scala:631)
at org.apache.spark.sql.execution.QueryExecution.$anonfun$executePhase$1(QueryExecution.scala:255)
at org.apache.spark.sql.SparkSession.withActive(SparkSession.scala:752)
at org.apache.spark.sql.execution.QueryExecution.executePhase(QueryExecution.scala:254)
at org.apache.spark.sql.execution.QueryExecution.$anonfun$lazyOptimizedPlan$1(QueryExecution.scala:175)
at scala.util.Try$.apply(Try.scala:217)
at org.apache.spark.util.Utils$.doTryWithCallerStacktrace(Utils.scala:1375)
at org.apache.spark.util.LazyTry.tryT$lzycompute(LazyTry.scala:46)
at org.apache.spark.util.LazyTry.tryT(LazyTry.scala:46)
at org.apache.spark.util.LazyTry.get(LazyTry.scala:58)
at org.apache.spark.sql.execution.QueryExecution.optimizedPlan(QueryExecution.scala:189)
at org.apache.spark.sql.execution.QueryExecution.assertOptimized(QueryExecution.scala:191)
at org.apache.spark.sql.execution.QueryExecution.$anonfun$lazyExecutedPlan$1(QueryExecution.scala:211)
at scala.util.Try$.apply(Try.scala:217)
at org.apache.spark.util.Utils$.doTryWithCallerStacktrace(Utils.scala:1375)
at org.apache.spark.util.LazyTry.tryT$lzycompute(LazyTry.scala:46)
at org.apache.spark.util.LazyTry.tryT(LazyTry.scala:46)
... 35 more
Suppressed: java.lang.Exception: Stacktrace under doTryWithCallerStacktrace
at org.apache.spark.sql.errors.QueryExecutionErrors$.invalidFractionOfSecondError(QueryExecutionErrors.scala:267)
at org.apache.spark.sql.catalyst.expressions.MakeTimestamp.toMicros(datetimeExpressions.scala:2811)
at org.apache.spark.sql.catalyst.expressions.MakeTimestamp.nullSafeEval(datetimeExpressions.scala:2847)
at org.apache.spark.sql.catalyst.expressions.SeptenaryExpression.eval(Expression.scala:1184)
at org.apache.spark.sql.catalyst.optimizer.ConstantFolding$.org$apache$spark$sql$catalyst$optimizer$ConstantFolding$$constantFolding(expressions.scala:82)
at org.apache.spark.sql.catalyst.optimizer.ConstantFolding$.$anonfun$constantFolding$4(expressions.scala:102)
at org.apache.spark.sql.catalyst.trees.UnaryLike.mapChildren(TreeNode.scala:1231)
at org.apache.spark.sql.catalyst.trees.UnaryLike.mapChildren$(TreeNode.scala:1230)
at org.apache.spark.sql.catalyst.expressions.UnaryExpression.mapChildren(Expression.scala:553)
at org.apache.spark.sql.catalyst.optimizer.ConstantFolding$.org$apache$spark$sql$catalyst$optimizer$ConstantFolding$$constantFolding(expressions.scala:102)
at org.apache.spark.sql.catalyst.optimizer.ConstantFolding$$anonfun$apply$1.$anonfun$applyOrElse$1(expressions.scala:106)
at org.apache.spark.sql.catalyst.plans.QueryPlan.$anonfun$mapExpressions$1(QueryPlan.scala:223)
at org.apache.spark.sql.catalyst.trees.CurrentOrigin$.withOrigin(origin.scala:86)
at org.apache.spark.sql.catalyst.plans.QueryPlan.transformExpression$1(QueryPlan.scala:223)
at org.apache.spark.sql.catalyst.plans.QueryPlan.recursiveTransform$1(QueryPlan.scala:235)
at org.apache.spark.sql.catalyst.plans.QueryPlan.$anonfun$mapExpressions$4(QueryPlan.scala:241)
at scala.collection.immutable.List.map(List.scala:247)
at scala.collection.immutable.List.map(List.scala:79)
at org.apache.spark.sql.catalyst.plans.QueryPlan.recursiveTransform$1(QueryPlan.scala:241)
at org.apache.spark.sql.catalyst.plans.QueryPlan.$anonfun$mapExpressions$5(QueryPlan.scala:246)
at org.apache.spark.sql.catalyst.trees.TreeNode.mapProductIterator(TreeNode.scala:312)
at org.apache.spark.sql.catalyst.plans.QueryPlan.mapExpressions(QueryPlan.scala:246)
at org.apache.spark.sql.catalyst.optimizer.ConstantFolding$$anonfun$apply$1.applyOrElse(expressions.scala:106)
at org.apache.spark.sql.catalyst.optimizer.ConstantFolding$$anonfun$apply$1.applyOrElse(expressions.scala:105)
at org.apache.spark.sql.catalyst.trees.TreeNode.$anonfun$transformDownWithPruning$1(TreeNode.scala:470)
at org.apache.spark.sql.catalyst.trees.CurrentOrigin$.withOrigin(origin.scala:86)
at org.apache.spark.sql.catalyst.trees.TreeNode.transformDownWithPruning(TreeNode.scala:470)
at org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.org$apache$spark$sql$catalyst$plans$logical$AnalysisHelper$$super$transformDownWithPruning(LogicalPlan.scala:37)
at org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper.transformDownWithPruning(AnalysisHelper.scala:330)
at org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper.transformDownWithPruning$(AnalysisHelper.scala:326)
at org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.transformDownWithPruning(LogicalPlan.scala:37)
at org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.transformDownWithPruning(LogicalPlan.scala:37)
at org.apache.spark.sql.catalyst.trees.TreeNode.transformWithPruning(TreeNode.scala:436)
at org.apache.spark.sql.catalyst.optimizer.ConstantFolding$.apply(expressions.scala:105)
at org.apache.spark.sql.catalyst.optimizer.ConstantFolding$.apply(expressions.scala:48)
at org.apache.spark.sql.catalyst.rules.RuleExecutor.$anonfun$execute$2(RuleExecutor.scala:226)
at scala.collection.LinearSeqOps.foldLeft(LinearSeq.scala:183)
at scala.collection.LinearSeqOps.foldLeft$(LinearSeq.scala:179)
at scala.collection.immutable.List.foldLeft(List.scala:79)
at org.apache.spark.sql.catalyst.rules.RuleExecutor.$anonfun$execute$1(RuleExecutor.scala:223)
at org.apache.spark.sql.catalyst.rules.RuleExecutor.$anonfun$execute$1$adapted(RuleExecutor.scala:215)
at scala.collection.immutable.List.foreach(List.scala:334)
at org.apache.spark.sql.catalyst.rules.RuleExecutor.execute(RuleExecutor.scala:215)
at org.apache.spark.sql.catalyst.rules.RuleExecutor.$anonfun$executeAndTrack$1(RuleExecutor.scala:186)
at org.apache.spark.sql.catalyst.QueryPlanningTracker$.withTracker(QueryPlanningTracker.scala:89)
at org.apache.spark.sql.catalyst.rules.RuleExecutor.executeAndTrack(RuleExecutor.scala:186)
at org.apache.spark.sql.execution.QueryExecution.$anonfun$lazyOptimizedPlan$2(QueryExecution.scala:179)
at org.apache.spark.sql.catalyst.QueryPlanningTracker.measurePhase(QueryPlanningTracker.scala:138)
at org.apache.spark.sql.execution.QueryExecution.$anonfun$executePhase$2(QueryExecution.scala:255)
at org.apache.spark.sql.execution.QueryExecution$.withInternalError(QueryExecution.scala:631)
at org.apache.spark.sql.execution.QueryExecution.$anonfun$executePhase$1(QueryExecution.scala:255)
at org.apache.spark.sql.SparkSession.withActive(SparkSession.scala:752)
at org.apache.spark.sql.execution.QueryExecution.executePhase(QueryExecution.scala:254)
at org.apache.spark.sql.execution.QueryExecution.$anonfun$lazyOptimizedPlan$1(QueryExecution.scala:175)
at scala.util.Try$.apply(Try.scala:217)
at org.apache.spark.util.Utils$.doTryWithCallerStacktrace(Utils.scala:1375)
at org.apache.spark.util.Utils$.getTryWithCallerStacktrace(Utils.scala:1429)
at org.apache.spark.util.LazyTry.get(LazyTry.scala:58)
at org.apache.spark.sql.execution.QueryExecution.optimizedPlan(QueryExecution.scala:189)
at org.apache.spark.sql.execution.QueryExecution.assertOptimized(QueryExecution.scala:191)
at org.apache.spark.sql.execution.QueryExecution.$anonfun$lazyExecutedPlan$1(QueryExecution.scala:211)
at scala.util.Try$.apply(Try.scala:217)
at org.apache.spark.util.Utils$.doTryWithCallerStacktrace(Utils.scala:1375)
Suppressed: java.lang.Exception: Full stacktrace of original doTryWithCallerStacktrace caller
at org.apache.spark.sql.errors.QueryExecutionErrors$.invalidFractionOfSecondError(QueryExecutionErrors.scala:267)
at org.apache.spark.sql.catalyst.expressions.MakeTimestamp.toMicros(datetimeExpressions.scala:2811)
at org.apache.spark.sql.catalyst.expressions.MakeTimestamp.nullSafeEval(datetimeExpressions.scala:2847)
at org.apache.spark.sql.catalyst.expressions.SeptenaryExpression.eval(Expression.scala:1184)
at org.apache.spark.sql.catalyst.optimizer.ConstantFolding$.org$apache$spark$sql$catalyst$optimizer$ConstantFolding$$constantFolding(expressions.scala:82)
at org.apache.spark.sql.catalyst.optimizer.ConstantFolding$.$anonfun$constantFolding$4(expressions.scala:102)
at org.apache.spark.sql.catalyst.trees.UnaryLike.mapChildren(TreeNode.scala:1231)
at org.apache.spark.sql.catalyst.trees.UnaryLike.mapChildren$(TreeNode.scala:1230)
at org.apache.spark.sql.catalyst.expressions.UnaryExpression.mapChildren(Expression.scala:553)
at org.apache.spark.sql.catalyst.optimizer.ConstantFolding$.org$apache$spark$sql$catalyst$optimizer$ConstantFolding$$constantFolding(expressions.scala:102)
at org.apache.spark.sql.catalyst.optimizer.ConstantFolding$$anonfun$apply$1.$anonfun$applyOrElse$1(expressions.scala:106)
at org.apache.spark.sql.catalyst.plans.QueryPlan.$anonfun$mapExpressions$1(QueryPlan.scala:223)
at org.apache.spark.sql.catalyst.trees.CurrentOrigin$.withOrigin(origin.scala:86)
at org.apache.spark.sql.catalyst.plans.QueryPlan.transformExpression$1(QueryPlan.scala:223)
at org.apache.spark.sql.catalyst.plans.QueryPlan.recursiveTransform$1(QueryPlan.scala:235)
at org.apache.spark.sql.catalyst.plans.QueryPlan.$anonfun$mapExpressions$4(QueryPlan.scala:241)
at scala.collection.immutable.List.map(List.scala:247)
at scala.collection.immutable.List.map(List.scala:79)
at org.apache.spark.sql.catalyst.plans.QueryPlan.recursiveTransform$1(QueryPlan.scala:241)
at org.apache.spark.sql.catalyst.plans.QueryPlan.$anonfun$mapExpressions$5(QueryPlan.scala:246)
at org.apache.spark.sql.catalyst.trees.TreeNode.mapProductIterator(TreeNode.scala:312)
at org.apache.spark.sql.catalyst.plans.QueryPlan.mapExpressions(QueryPlan.scala:246)
at org.apache.spark.sql.catalyst.optimizer.ConstantFolding$$anonfun$apply$1.applyOrElse(expressions.scala:106)
at org.apache.spark.sql.catalyst.optimizer.ConstantFolding$$anonfun$apply$1.applyOrElse(expressions.scala:105)
at org.apache.spark.sql.catalyst.trees.TreeNode.$anonfun$transformDownWithPruning$1(TreeNode.scala:470)
at org.apache.spark.sql.catalyst.trees.CurrentOrigin$.withOrigin(origin.scala:86)
at org.apache.spark.sql.catalyst.trees.TreeNode.transformDownWithPruning(TreeNode.scala:470)
at org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.org$apache$spark$sql$catalyst$plans$logical$AnalysisHelper$$super$transformDownWithPruning(LogicalPlan.scala:37)
at org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper.transformDownWithPruning(AnalysisHelper.scala:330)
at org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper.transformDownWithPruning$(AnalysisHelper.scala:326)
at org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.transformDownWithPruning(LogicalPlan.scala:37)
at org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.transformDownWithPruning(LogicalPlan.scala:37)
at org.apache.spark.sql.catalyst.trees.TreeNode.transformWithPruning(TreeNode.scala:436)
at org.apache.spark.sql.catalyst.optimizer.ConstantFolding$.apply(expressions.scala:105)
at org.apache.spark.sql.catalyst.optimizer.ConstantFolding$.apply(expressions.scala:48)
at org.apache.spark.sql.catalyst.rules.RuleExecutor.$anonfun$execute$2(RuleExecutor.scala:226)
at scala.collection.LinearSeqOps.foldLeft(LinearSeq.scala:183)
at scala.collection.LinearSeqOps.foldLeft$(LinearSeq.scala:179)
at scala.collection.immutable.List.foldLeft(List.scala:79)
at org.apache.spark.sql.catalyst.rules.RuleExecutor.$anonfun$execute$1(RuleExecutor.scala:223)
at org.apache.spark.sql.catalyst.rules.RuleExecutor.$anonfun$execute$1$adapted(RuleExecutor.scala:215)
at scala.collection.immutable.List.foreach(List.scala:334)
at org.apache.spark.sql.catalyst.rules.RuleExecutor.execute(RuleExecutor.scala:215)
at org.apache.spark.sql.catalyst.rules.RuleExecutor.$anonfun$executeAndTrack$1(RuleExecutor.scala:186)
at org.apache.spark.sql.catalyst.QueryPlanningTracker$.withTracker(QueryPlanningTracker.scala:89)
at org.apache.spark.sql.catalyst.rules.RuleExecutor.executeAndTrack(RuleExecutor.scala:186)
at org.apache.spark.sql.execution.QueryExecution.$anonfun$lazyOptimizedPlan$2(QueryExecution.scala:179)
at org.apache.spark.sql.catalyst.QueryPlanningTracker.measurePhase(QueryPlanningTracker.scala:138)
at org.apache.spark.sql.execution.QueryExecution.$anonfun$executePhase$2(QueryExecution.scala:255)
at org.apache.spark.sql.execution.QueryExecution$.withInternalError(QueryExecution.scala:631)
at org.apache.spark.sql.execution.QueryExecution.$anonfun$executePhase$1(QueryExecution.scala:255)
at org.apache.spark.sql.SparkSession.withActive(SparkSession.scala:752)
at org.apache.spark.sql.execution.QueryExecution.executePhase(QueryExecution.scala:254)
at org.apache.spark.sql.execution.QueryExecution.$anonfun$lazyOptimizedPlan$1(QueryExecution.scala:175)
at scala.util.Try$.apply(Try.scala:217)
at org.apache.spark.util.Utils$.doTryWithCallerStacktrace(Utils.scala:1375)
at org.apache.spark.util.Utils$.getTryWithCallerStacktrace(Utils.scala:1429)
at org.apache.spark.util.LazyTry.get(LazyTry.scala:58)
at org.apache.spark.sql.execution.QueryExecution.optimizedPlan(QueryExecution.scala:189)
at org.apache.spark.sql.execution.QueryExecution.assertOptimized(QueryExecution.scala:191)
at org.apache.spark.sql.execution.QueryExecution.$anonfun$lazyExecutedPlan$1(QueryExecution.scala:211)
at scala.util.Try$.apply(Try.scala:217)
at org.apache.spark.util.Utils$.doTryWithCallerStacktrace(Utils.scala:1375)
at org.apache.spark.util.LazyTry.tryT$lzycompute(LazyTry.scala:46)
at org.apache.spark.util.LazyTry.tryT(LazyTry.scala:46)
... 35 more
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48195] Save and reuse RDD/Broadcast created by SparkPlan [spark]
Posted by "juliuszsompolski (via GitHub)" <gi...@apache.org>.
juliuszsompolski commented on code in PR #48037:
URL: https://github.com/apache/spark/pull/48037#discussion_r1751855958
##########
core/src/main/scala/org/apache/spark/util/Lazy.scala:
##########
@@ -0,0 +1,61 @@
+/*
+ * 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.util
+
+import scala.util.Try
+
+/**
+ * Wrapper utility for lazy val, with two differences compared to scala behavir:
+ *
+ * 1. In scala, when a `lazy val` field is initialized, it grabs the synchronized lock on the
+ * enclosing object instance. This can lead both to performance issues, and deadlocks.
+ * For example:
+ * a) Thread 1 entered a synchronized method, grabbing a coarse lock on the parent object.
+ * b) Thread 2 get spawned off, and tries to initialize a lazy value on the same parent object
+ * This causes scala to also try to grab a lock on the parent object.
+ * c) If thread 1 waits for thread 2 to join, a deadlock occurs.
+ * This Lazy wrapper will only grab a lock on the wrapper itself, and not the parent object.
+ *
+ * 2. In scala, when a `lazy val` field initialization throws an exception, the field remains
+ * uninitialized, and initialization will be re-attempted on the next access. This also can lead
+ * to performance issues, needlessly computing something towards a failure, and also can lead to
+ * duplicated side effects.
+ * This Lazy wrapper stores the exception in a Try, and will re-throw it on the next access.
+ *
+ * @param initialize The block of code to initialize the lazy value.
+ * @tparam T type of the lazy value.
+ */
+private[spark] class Lazy[T](initialize: => T) extends Serializable {
+ private lazy val tryT: Try[T] = Utils.doTryWithCallerStacktrace { initialize }
Review Comment:
I renamed it to `LazyTry`, good point. The non-retrying behavior was the primary point of what I wanted to get rid of strange duplicated side effects behaviors.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48195] Save and reuse RDD/Broadcast created by SparkPlan [spark]
Posted by "juliuszsompolski (via GitHub)" <gi...@apache.org>.
juliuszsompolski commented on code in PR #48037:
URL: https://github.com/apache/spark/pull/48037#discussion_r1750547413
##########
core/src/test/scala/org/apache/spark/util/UtilsSuite.scala:
##########
@@ -1523,6 +1523,116 @@ class UtilsSuite extends SparkFunSuite with ResetSystemProperties {
conf.set(SERIALIZER, "org.apache.spark.serializer.JavaSerializer")
assert(Utils.isPushBasedShuffleEnabled(conf, isDriver = true) === false)
}
+
+
+ private def throwException(): String = {
+ throw new Exception("test")
+ }
+
+ private def callDoTry(): Try[String] = {
+ Utils.doTryWithCallerStacktrace {
+ throwException()
+ }
+ }
+
+ private def callGetTry(t: Try[String]): String = {
+ Utils.getTryWithCallerStacktrace(t)
+ }
+
+ private def callGetTryAgain(t: Try[String]): String = {
+ Utils.getTryWithCallerStacktrace(t)
+ }
+
+ test("doTryWithCallerStacktrace and getTryWithCallerStacktrace") {
+ val t = callDoTry()
+
+ val e1 = intercept[Exception] {
+ callGetTry(t)
+ }
+ // Uncomment for manual inspection
+ // e.printStackTrace()
+ // Example:
+ // java.lang.Exception: test
+ // at org.apache.spark.util.UtilsSuite.throwException(UtilsSuite.scala:1640)
+ // at org.apache.spark.util.UtilsSuite.$anonfun$callDoTry$1(UtilsSuite.scala:1645)
+ // at scala.util.Try$.apply(Try.scala:213)
+ // at org.apache.spark.util.Utils$.doTryWithCallerStacktrace(Utils.scala:1586)
+ // at org.apache.spark.util.Utils$.getTryWithCallerStacktrace(Utils.scala:1639)
+ // at org.apache.spark.util.UtilsSuite.callGetTry(UtilsSuite.scala:1650)
+ // at org.apache.spark.util.UtilsSuite.$anonfun$new$165(UtilsSuite.scala:1661)
+ // <- callGetTry is seen as calling getTryWithCallerStacktrace
+
+ val st1 = e1.getStackTrace
+ // throwException should be on the stack trace
+ assert(st1.exists(_.getMethodName == "throwException"))
+ // callDoTry shouldn't be on the stack trace, but callGetTry should be.
+ assert(!st1.exists(_.getMethodName == "callDoTry"))
+ assert(st1.exists(_.getMethodName == "callGetTry"))
+
+ // The original stack trace with callDoTry should be in the suppressed exceptions.
+ // Example:
+ // scalastyle:off line.size.limit
+ // Suppressed: java.lang.Exception: Full stacktrace of original doTryWithCallerStacktrace caller
+ // at org.apache.spark.util.UtilsSuite.throwException(UtilsSuite.scala:1640)
+ // at org.apache.spark.util.UtilsSuite.$anonfun$callDoTry$1(UtilsSuite.scala:1645)
+ // at scala.util.Try$.apply(Try.scala:213)
+ // at org.apache.spark.util.Utils$.doTryWithCallerStacktrace(Utils.scala:1586)
+ // at org.apache.spark.util.UtilsSuite.callDoTry(UtilsSuite.scala:1645)
+ // at org.apache.spark.util.UtilsSuite.$anonfun$new$165(UtilsSuite.scala:1658)
+ // ... 56 more
+ // scalastyle:on line.size.limit
+ val origSt = e1.getSuppressed.find(
+ _.getMessage == Utils.TRY_WITH_CALLER_STACKTRACE_FULL_STACKTRACE)
+ assert(origSt.isDefined)
+ assert(origSt.get.getStackTrace.exists(_.getMethodName == "throwException"))
+ assert(origSt.get.getStackTrace.exists(_.getMethodName == "callDoTry"))
+
+ // The stack trace under Try should be in the suppressed exceptions.
+ // Example:
+ // Suppressed: java.lang.Exception: Stacktrace under doTryWithCallerStacktrace
+ // at org.apache.spark.util.UtilsSuite.throwException(UtilsSuite.scala: 1640)
+ // at org.apache.spark.util.UtilsSuite.$anonfun$callDoTry$1(UtilsSuite.scala: 1645)
+ // at scala.util.Try$.apply(Try.scala: 213)
+ // at org.apache.spark.util.Utils$.doTryWithCallerStacktrace(Utils.scala: 1586)
+ val trySt = e1.getSuppressed.find(
+ _.getMessage == Utils.TRY_WITH_CALLER_STACKTRACE_TRY_STACKTRACE)
+ assert(trySt.isDefined)
+ // calls under callDoTry should be present.
+ assert(trySt.get.getStackTrace.exists(_.getMethodName == "throwException"))
+ // callDoTry should be removed.
+ assert(!trySt.get.getStackTrace.exists(_.getMethodName == "callDoTry"))
+
+ val e2 = intercept[Exception] {
+ callGetTryAgain(t)
+ }
+ // Uncomment for manual inspection
+ // e.printStackTrace()
+ // Example:
+ // java.lang.Exception: test
+ // at org.apache.spark.util.UtilsSuite.throwException(UtilsSuite.scala:1640)
+ // at org.apache.spark.util.UtilsSuite.$anonfun$callDoTry$1(UtilsSuite.scala:1645)
+ // at scala.util.Try$.apply(Try.scala:213)
+ // at org.apache.spark.util.Utils$.doTryWithCallerStacktrace(Utils.scala:1586)
+ // at org.apache.spark.util.Utils$.getTryWithCallerStacktrace(Utils.scala:1639)
+ // at org.apache.spark.util.UtilsSuite.callGetTryAgain(UtilsSuite.scala:1654)
+ // at org.apache.spark.util.UtilsSuite.$anonfun$new$165(UtilsSuite.scala:1711)
+ // <- callGetTryAgain is seen as calling getTryWithCallerStacktrace
+
+ val st2 = e2.getStackTrace
+ // throwException should be on the stack trace
+ assert(st2.exists(_.getMethodName == "throwException"))
+ // callDoTry shouldn't be on the stack trace, but callGetTryAgain should be.
+ assert(!st2.exists(_.getMethodName == "callDoTry"))
+ assert(st2.exists(_.getMethodName == "callGetTryAgain"))
+ // callGetTry that we called before shouldn't be on the stack trace.
+ assert(!st2.exists(_.getMethodName == "callGetTry"))
+
+ // Unfortunately, this utility is not able to clone the exception, but modifies it in place,
+ // so now e1 is also pointing to "callGetTryAgain" instead of "callGetTry".
+ val st1Again = e1.getStackTrace
+ assert(st1Again.exists(_.getMethodName == "callGetTryAgain"))
+ assert(!st1Again.exists(_.getMethodName == "callGetTry"))
Review Comment:
This is a bit of ugliness of this, that since we are not able to clone the exception, it modifies it in place.
This is fine as long as the Lazy object with an error is not accessed concurrently by multiple retrievers...
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org