From d6d6b0454a06a33eb3579e3e4b8080742a3fb32f Mon Sep 17 00:00:00 2001
From: Slava Min <slava.min@databricks.com>
Date: Mon, 6 Jan 2025 12:06:19 -0800
Subject: [PATCH] =?UTF-8?q?Release=20the=20reference=20to=20the=20calling?=
 =?UTF-8?q?=20Spark=20session=20upon=20successful=20co=E2=80=A6=20(#4010)?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

…mpletion of NonFateSharingFuture

<!--
Thanks for sending a pull request!  Here are some tips for you:
1. If this is your first time, please read our contributor guidelines:
https://github.com/delta-io/delta/blob/master/CONTRIBUTING.md
2. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP]
Your PR title ...'.
  3. Be sure to keep the PR description updated to reflect all changes.
  4. Please write your PR title to summarize what this PR proposes.
5. If possible, provide a concise example to reproduce the issue for a
faster review.
6. If applicable, include the corresponding issue number in the PR title
and link it in the body.
-->

#### Which Delta project/connector is this regarding?
<!--
Please add the component selected below to the beginning of the pull
request title
For example: [Spark] Title of my pull request
-->

- [X] Spark
- [ ] Standalone
- [ ] Flink
- [ ] Kernel
- [ ] Other (fill in here)

## Description

SparkSession object which initiated the Delta snapshot is kept alive for
the lifetime of the snapshot. This change releases the reference to
SparkSession once the future completes.

## How was this patch tested?

## Does this PR introduce _any_ user-facing changes?

No
---
 .../spark/sql/delta/util/threads/DeltaThreadPool.scala       | 5 ++++-
 1 file changed, 4 insertions(+), 1 deletion(-)

diff --git a/spark/src/main/scala/org/apache/spark/sql/delta/util/threads/DeltaThreadPool.scala b/spark/src/main/scala/org/apache/spark/sql/delta/util/threads/DeltaThreadPool.scala
index d5ec23fb336..62d196e3640 100644
--- a/spark/src/main/scala/org/apache/spark/sql/delta/util/threads/DeltaThreadPool.scala
+++ b/spark/src/main/scala/org/apache/spark/sql/delta/util/threads/DeltaThreadPool.scala
@@ -107,7 +107,10 @@ class NonFateSharingFuture[T](pool: DeltaThreadPool)(f: SparkSession => T)
     // Prefer to get a prefetched result from the future, but never fail because of it.
     val futureResult = futureOpt.flatMap { case (ownerSession, future) =>
       try {
-        Some(ThreadUtils.awaitResult(future, timeout))
+        val result = Some(ThreadUtils.awaitResult(future, timeout))
+        // no reason to keep the reference to the calling session anymore
+        futureOpt = Some(null, future)
+        result
       } catch {
         // NOTE: ThreadUtils.awaitResult wraps all non-fatal exceptions other than TimeoutException
         // with SparkException. Meanwhile, Java Future.get only throws four exceptions: