[SPARK-49808][SQL] Fix a deadlock in subquery execution due to lazy vals

core/src/main/scala/org/apache/spark/util/Lazy.scala

@@ -0,0 +1,42 @@
+/*
+ * 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
+
+/**
+ * Construct to lazily initialize a variable.
+ * This may be helpful for avoiding deadlocks in certain scenarios. For example,
+ *   a) Thread 1 entered a synchronized method, grabbing a coarse lock on the parent object.
+ *   b) Thread 2 gets spawned off, and tries to initialize a lazy value on the same parent object
+ *      (in our case, this was the logger). This causes scala to also try to grab a coarse lock on
+ *      the parent object.
+ *   c) If thread 1 waits for thread 2 to join, a deadlock occurs.
+ * The main difference between this and [[LazyTry]] is that this does not cache failures.
+ *
+ * @note
+ *   Scala 3 uses a different implementation of lazy vals which doesn't have this problem.
+ *   Please refer to <a
+ *   href="https://docs.scala-lang.org/scala3/reference/changed-features/lazy-vals-init.html">Lazy
+ *   Vals Initialization</a> for more details.
+ */
+private[spark] class Lazy[T](initializer: => T) extends Serializable {
+
+  private[this] lazy val value: T = initializer
+
+  def apply(): T = {
+    value
+  }
+}

core/src/test/scala/org/apache/spark/util/LazySuite.scala

@@ -0,0 +1,60 @@
+/*
+ * 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 java.io.{ByteArrayOutputStream, NotSerializableException, ObjectOutputStream}
+
+import org.apache.spark.SparkFunSuite
+
+class LazySuite extends SparkFunSuite {
+  test("Lazy val works") {
+    var test: Option[Object] = None
+
+    val lazyval = new Lazy({
+      test = Some(new Object())
+      test
+    })
+
+    // Ensure no initialization happened before the lazy value was dereferenced
+    assert(test.isEmpty)
+
+    // Ensure the first invocation creates a new object
+    assert(lazyval() == test && test.isDefined)
+
+    // Ensure the subsequent invocation serves the same object
+    assert(lazyval() == test && test.isDefined)
+  }
+
+  test("Lazy val is serializable") {
+    val lazyval = new Lazy({
+      new Object()
+    })
+
+    // Ensure we are serializable before the dereference
+    val oos = new ObjectOutputStream(new ByteArrayOutputStream())
+    oos.writeObject(lazyval)
+
+    @SuppressWarnings(Array("never used"))
+    val dereferenced = lazyval()
+
+    // Ensure we are not serializable after the dereference (type "Object" is not serializable)
+    intercept[NotSerializableException] {
+      val oos2 = new ObjectOutputStream(new ByteArrayOutputStream())
+      oos2.writeObject(lazyval)
+    }
+  }
+}

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/QueryPlan.scala

@@ -32,6 +32,7 @@ import org.apache.spark.sql.catalyst.trees.TreePatternBits
 import org.apache.spark.sql.catalyst.types.DataTypeUtils
 import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.types.{DataType, StructType}
+import org.apache.spark.util.Lazy
 import org.apache.spark.util.collection.BitSet
 
 /**
@@ -94,10 +95,11 @@ abstract class QueryPlan[PlanType <: QueryPlan[PlanType]]
    * All Attributes that appear in expressions from this operator.  Note that this set does not
    * include attributes that are implicitly referenced by being passed through to the output tuple.
    */
-  @transient
-  lazy val references: AttributeSet = {
-    AttributeSet.fromAttributeSets(expressions.map(_.references)) -- producedAttributes
-  }
+  def references: AttributeSet = _references()
+
+  private val _references = new Lazy({
+    AttributeSet(expressions) -- producedAttributes
+  })
 
   /**
    * Returns true when the all the expressions in the current node as well as all of its children