[SPARK-48344][SQL] Enhance SQL Script Execution: Replace NOOP with COLLECT for Result DataFrames

[miland-db]

What changes were proposed in this pull request?

This pull request proposes replacing the noop operation with collect for all result DataFrames on the caller side of the SQL Script execution process.

This is the 4th PR in the series of introducing SQL Scripting Execution into Spark.

Why are the changes needed?

The proposed change is necessary to maintain a critical invariant during SQL Script execution: when SqlScriptingExecution returns the next available result statement, it must be executed before proceeding with iteration.

Implementation details

SQL Script execution is based on iterating over interpreted statements and executing them as they are encountered. For certain statement types (result statements), execution is delegated to the caller side (SparkSession). To achieve this, the iteration process is divided into two stages:

• All Compound Execution Nodes (Begin-End block, control flow structures, loops) implement iterator accessible via the getTreeIterator method.
• SqlScriptingExecution serves as a second-level iterator, iterating over all statements and executing those that are not result statements. Result statements are returned to the caller for execution on the caller side. The caller must adhere to the contract of executing the returned statement before continuing iteration.

Due to the nature of this contract between the caller and the SqlScriptingExecution API, the implementation of the hasNext method is not idempotent. We will enforce correct usage of SqlScriptingExecution API through the future PR review process.

In this approach we collect every DataFrame to eliminate concerns about which one needs to be returned last. This strategy will also be utilized when we introduce multiple-results API functionality.

Impact and Considerations

This change enhances the robustness of our SQL Script execution process and lays the groundwork for future improvements, including the implementation of a multiple-results API. Reviewers should pay particular attention to the handling of DataFrame collection and the maintenance of execution order integrity.

Does this PR introduce any user-facing change?

No

How was this patch tested?

Existing tests.

Was this patch authored or co-authored using generative AI tooling?

No

[miland-db]

Adding @cloud-fan @davidm-db @dusantism-db @dejankrak-db @MaxGekk

[cloud-fan]

This violates the Java Iterator contract. Can we make sure it gets the next result?

[miland-db]

Because of the nature of contract between the caller and SqlScriptingExecution API which is to execute returned statement before proceeding with iteration, we can't advance further before first executing the returned statement.

Advancing in the next method immediately after returning the current statement would break the contract.
Advancing to the next result statement or end of script has to be done after current result statement is executed. For that reason next can't be the one advancing the iteration. Here is the example of the code that would not work correctly:

override def hasNext(): Boolean = current.isDefined

override def next(): DataFrame = {
  val tmp = current
  current = getNextResult
  return tmp
}

The idea is to enforce correct usage of SqlScriptingExecution API through the future PR review process, and keep the behavior of hasNext and next well documented.

cc. @davidm-db @dejankrak-db

[cloud-fan]

Then we should not use Java iterator here but ask the caller to invoke getNextResult directly. The contract is caller must execute the returned DataFrame before getting the next one.

[miland-db]

We still have to know when we have reached the end of script and there are multiple solutions to this problem.
I have two proposals:

• Remove extends Iterator[DataFrame] from SqlScriptingExecution and continue to use hasNext and next as we use them now. That way we don't have to respect Java iterator contract, and we still have user friendly way to iterate through script.
• Remove everything related to iterator and execute the script in a way similar to this:

var result = Option[Seq[Row]] = None
var currentDF: Option[DataFrame] = sse.getNextResult()
while (currentDF.isDefined) {
  result = Some(currentDF.collect().toSeq)
  currentDF = sse.getNextResult()
}

return result...

[cloud-fan]

Yea proposal 2 is what I was proposing.