Revert "[SPARK-49808][SQL] Fix a deadlock in subquery execution due o lazy vals"

[zhengruifeng]

This reverts commit d7abddc.

We had offline discussion, and @JoshRosen pointed out that:

The use of LazyTry vs. a non-try Lazy wrappers needs discussion:
LazyTry caches failures. As a result, there is a potential risk of cancellation exceptions being cached here.
This is possibly an issue in case AQE interrupts subquery execution threads during planning, or otherwise can interrupt a subset of a query plan.

[mridulm]

Can we add some details on why we are reverting to the description please ? For example: bug still exists in scenario foo, or new bug introduced for xyz, etc ?
Thanks

[zhengruifeng]

@mridulm thanks, let me update the description:
the reason is that the fix potentially causes other issues, so still need discussion and investigation.

[HyukjinKwon]

I am fine with this. If there's a concern to be discussed, let's open a PR again, and discuss it there.

[HyukjinKwon]

Merged to master.

[tedyu]

What if we introduce the following wrapper in place of the Scala lazy value ?

class LazyRetry[T](compute: => T) {
  private var initialized = false
  private var value: Option[T] = None

  def get: T = synchronized {
    if (!initialized) {
      try {
        value = Some(compute)
        initialized = true
      } catch {
        case e: Exception =>
          // So that it can retry on the next call
          initialized = false
          throw e
      }
    }
    value.getOrElse(throw new RuntimeException("Computation not successful"))
  }
}

This wrapper doesn't cache exceptions.