[SPARK-44287][SQL][FOLLOWUP] Do not trigger execution too early

### What changes were proposed in this pull request?

This is a followup of apache#41839 , to fix an unintentional change. That PR added an optimization to return an empty iterator directly if the input iterator is empty. However, checking `inputIterator.hasNext` may trigger query execution, which is different than before. It should be completely lazy and wait for the root operator's iterator to trigger the execution.

### Why are the changes needed?

fix unintentional change

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

no

### How was this patch tested?

existing tests

Closes apache#42226 from cloud-fan/fo.

Authored-by: Wenchen Fan <[email protected]>
Signed-off-by: Hyukjin Kwon <[email protected]>

Author: cloud-fan

sql/core/src/main/scala/org/apache/spark/sql/execution/ColumnarEvaluatorFactory.scala

@@ -70,42 +70,37 @@ class RowToColumnarEvaluatorFactory(
         inputs: Iterator[InternalRow]*): Iterator[ColumnarBatch] = {
       assert(inputs.length == 1)
       val rowIterator = inputs.head
+      new Iterator[ColumnarBatch] {
+        private lazy val converters = new RowToColumnConverter(schema)
+        private lazy val vectors: Seq[WritableColumnVector] = if (enableOffHeapColumnVector) {
+          OffHeapColumnVector.allocateColumns(numRows, schema)
+        } else {
+          OnHeapColumnVector.allocateColumns(numRows, schema)
+        }
+        private lazy val cb: ColumnarBatch = new ColumnarBatch(vectors.toArray)
 
-      if (rowIterator.hasNext) {
-        new Iterator[ColumnarBatch] {
-          private val converters = new RowToColumnConverter(schema)
-          private val vectors: Seq[WritableColumnVector] = if (enableOffHeapColumnVector) {
-            OffHeapColumnVector.allocateColumns(numRows, schema)
-          } else {
-            OnHeapColumnVector.allocateColumns(numRows, schema)
-          }
-          private val cb: ColumnarBatch = new ColumnarBatch(vectors.toArray)
-
-          TaskContext.get().addTaskCompletionListener[Unit] { _ =>
-            cb.close()
-          }
+        TaskContext.get().addTaskCompletionListener[Unit] { _ =>
+          cb.close()
+        }
 
-          override def hasNext: Boolean = {
-            rowIterator.hasNext
-          }
+        override def hasNext: Boolean = {
+          rowIterator.hasNext
+        }
 
-          override def next(): ColumnarBatch = {
-            cb.setNumRows(0)
-            vectors.foreach(_.reset())
-            var rowCount = 0
-            while (rowCount < numRows && rowIterator.hasNext) {
-              val row = rowIterator.next()
-              converters.convert(row, vectors.toArray)
-              rowCount += 1
-            }
-            cb.setNumRows(rowCount)
-            numInputRows += rowCount
-            numOutputBatches += 1
-            cb
+        override def next(): ColumnarBatch = {
+          cb.setNumRows(0)
+          vectors.foreach(_.reset())
+          var rowCount = 0
+          while (rowCount < numRows && rowIterator.hasNext) {
+            val row = rowIterator.next()
+            converters.convert(row, vectors.toArray)
+            rowCount += 1
           }
+          cb.setNumRows(rowCount)
+          numInputRows += rowCount
+          numOutputBatches += 1
+          cb
         }
-      } else {
-        Iterator.empty
       }
     }
   }