[Spark] Fix ClosableIterator to only open inner iterators when hasNex…

…t() is called (#4022)

<!--
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

<!--
- Describe what this PR changes.
- Describe why we need the change.
 
If this PR resolves an issue be sure to include "Resolves #XXX" to
correctly link and close the issue upon merge.
-->
Iterators cannot expect to be closed after instantiation, they can only
expect that the caller will call close() after some next() or hasNext()
call. For example, in
`DeltaSource.getStartingOffsetFromSpecificDeltaVersion`, we do not wrap
the iterator in a `finally` block directly after
[instantiation](https://github.com/delta-io/delta/blob/7381d5a8b27b6c721b8b8da9e095c695cfd727b2/spark/src/main/scala/org/apache/spark/sql/delta/sources/DeltaSource.scala#L411)
but rather only once we have [started
iterating](https://github.com/delta-io/delta/blob/7381d5a8b27b6c721b8b8da9e095c695cfd727b2/spark/src/main/scala/org/apache/spark/sql/delta/sources/DeltaSource.scala#L1458).

However, the current `ClosableIterator` implementation from
`flatMapWithClose` opens an inner iterator on instantiation, leaving a
small window of time between instantiation and iteration, in which, if
the thread gets interrupted, resources will not be cleaned up. This PR
fixes this race condition by only opening the first inner iterator on
the first `hasNext` call.

## How was this patch tested?

<!--
If tests were added, say they were added here. Please make sure to test
the changes thoroughly including negative and positive cases if
possible.
If the changes were tested in any way other than unit tests, please
clarify how you tested step by step (ideally copy and paste-able, so
that other reviewers can test and check, and descendants can verify in
the future).
If the changes were not tested, please explain why.
-->
Added new unit test.

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

<!--
If yes, please clarify the previous behavior and the change this PR
proposes - provide the console output, description and/or an example to
show the behavior difference if possible.
If possible, please also clarify if this is a user-facing change
compared to the released Delta Lake versions or within the unreleased
branches such as master.
If no, write 'No'.
-->
No

spark/src/main/scala/org/apache/spark/sql/delta/storage/ClosableIterator.scala

@@ -63,12 +63,7 @@ object ClosableIterator {
   implicit class IteratorFlatMapCloseOp[A](val closableIter: Iterator[A]) extends AnyVal {
     def flatMapWithClose[B](f: A => ClosableIterator[B]): ClosableIterator[B] =
       new ClosableIterator[B] {
-        private var iter_curr =
-          if (closableIter.hasNext) {
-            f(closableIter.next())
-          } else {
-            null
-          }
+        private var iter_curr: ClosableIterator[B] = null
         override def next(): B = {
           if (!hasNext) {
             throw new NoSuchElementException
@@ -77,6 +72,9 @@ object ClosableIterator {
         }
         @scala.annotation.tailrec
         override def hasNext: Boolean = {
+          if (iter_curr == null && closableIter.hasNext) {
+            iter_curr = f(closableIter.next())
+          }
           if (iter_curr == null) {
             false
           }

spark/src/test/scala/org/apache/spark/sql/delta/storage/LineClosableIteratorSuite.scala

@@ -122,6 +122,64 @@ abstract class LineClosableIteratorSuiteBase extends SparkFunSuite {
     iter.close()
     assert(closed == 1)
   }
+
+  test("flatMapWithClose does not open any iterators on creation") {
+    var opened = 0
+    var closed = 0
+    val outerReader = new StringReader("b\na\nr")
+    createIter(outerReader).flatMapWithClose(_ => {
+      val innerReader = new StringReader("f\no\no") {
+        opened += 1
+        override def close(): Unit = {
+          super.close()
+          closed += 1
+        }
+      }
+      createIter(innerReader)
+    })
+    assert(opened == 0)
+    assert(closed == 0)
+  }
+
+  test("flatMapWithClose calls close only for opened iterators") {
+    var opened = 0
+    var closed = 0
+    val outerReader = new StringReader("b\na\nr")
+    val iter = createIter(outerReader).flatMapWithClose(_ => {
+      val innerReader = new StringReader("f\no\no") {
+        opened += 1
+        override def close(): Unit = {
+          super.close()
+          closed += 1
+        }
+      }
+      createIter(innerReader)
+    })
+    assert(iter.take(5).toList == List("f", "o", "o", "f", "o"))
+    iter.close()
+    assert(opened == 2)
+    assert(closed == 2)
+  }
+
+  test("flatMapWithClose calls close only for opened iterators - iter boundary") {
+    var opened = 0
+    var closed = 0
+    val outerReader = new StringReader("b\na\nr")
+    val iter = createIter(outerReader).flatMapWithClose(_ => {
+      val innerReader = new StringReader("f\no\no") {
+        opened += 1
+        override def close(): Unit = {
+          super.close()
+          closed += 1
+        }
+      }
+      createIter(innerReader)
+    })
+    assert(iter.take(3).toList == List("f", "o", "o"))
+    iter.close()
+    assert(opened == 1)
+    assert(closed == 1)
+  }
 }
 
 class InternalLineClosableIteratorSuite extends LineClosableIteratorSuiteBase {