You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "miland-db (via GitHub)" <gi...@apache.org> on 2024/07/18 12:51:16 UTC
[PR] [WIP][SPARK-][SQL] SQL API change to support execution of compound statements [spark]
miland-db opened a new pull request, #47403:
URL: https://github.com/apache/spark/pull/47403
### What changes were proposed in this pull request?
Previous PRs introduced basic changes for SQL Scripting.
This PR is a follow-up to introduce changes to SQL API in `SparkSession` to support execution of compound statements.
### Why are the changes needed?
This change is needed to open the possibility for users to execute and get results from sql scripts.
### Does this PR introduce any user-facing change?
Users will now see be able to execute sql scripts using `spark.sql()` API.
### How was this patch tested?
There are tests for newly introduced parser changes:
`SqlScriptingExecutionNodeSuite` - existing unit tests for execution nodes.
### Was this patch authored or co-authored using generative AI tooling?
No.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48344][SQL] SQL API change to support execution of compound statements [spark]
Posted by "miland-db (via GitHub)" <gi...@apache.org>.
miland-db commented on code in PR #47403:
URL: https://github.com/apache/spark/pull/47403#discussion_r1688196798
##########
sql/core/src/main/scala/org/apache/spark/sql/SparkSession.scala:
##########
@@ -650,14 +657,27 @@ class SparkSession private(
private[sql] def sql(sqlText: String, args: Array[_], tracker: QueryPlanningTracker): DataFrame =
withActive {
val plan = tracker.measurePhase(QueryPlanningTracker.PARSING) {
- val parsedPlan = sessionState.sqlParser.parsePlan(sqlText)
- if (args.nonEmpty) {
- PosParameterizedQuery(parsedPlan, args.map(lit(_).expr).toImmutableArraySeq)
- } else {
- parsedPlan
+ val parsedPlan = sessionState.sqlParser.parseScript(sqlText)
+ parsedPlan match {
+ case CompoundBody(Seq(singleStmtPlan: SingleStatement), label) if args.nonEmpty =>
+ CompoundBody(Seq(SingleStatement(
+ PosParameterizedQuery(
+ singleStmtPlan.parsedPlan, args.map(lit(_).expr).toImmutableArraySeq))), label)
+ case p =>
+ assert(args.isEmpty, "Named parameters are not supported for batch queries")
+ p
}
}
- Dataset.ofRows(self, plan, tracker)
+
+ plan match {
+ case CompoundBody(Seq(singleStmtPlan: SingleStatement), _) =>
+ Dataset.ofRows(self, singleStmtPlan.parsedPlan, tracker)
+ case _ =>
+ // execute the plan directly if it is not a single statement
+ val lastRow = executeScript(plan).foldLeft(Array.empty[Row])((_, next) => next)
+ val attributes = DataTypeUtils.toAttributes(lastRow.head.schema)
+ Dataset.ofRows(self, LocalRelation.fromExternalRows(attributes, lastRow.toIndexedSeq))
Review Comment:
What is noop sink doing? Is it `df.write.format("noop").mode("overwrite").save()`? Is it the same as doing `df.collect()` but just throws away the result?
We have a hard time determining which statement is the last statement. That is the reason why we are doing it this way (we have to save the result of the last `dataframe`).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48344][SQL] SQL API change to support execution of compound statements [spark]
Posted by "miland-db (via GitHub)" <gi...@apache.org>.
miland-db closed pull request #47403: [SPARK-48344][SQL] SQL API change to support execution of compound statements
URL: https://github.com/apache/spark/pull/47403
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48344][SQL] SQL API change to support execution of compound statements [spark]
Posted by "miland-db (via GitHub)" <gi...@apache.org>.
miland-db commented on code in PR #47403:
URL: https://github.com/apache/spark/pull/47403#discussion_r1688196798
##########
sql/core/src/main/scala/org/apache/spark/sql/SparkSession.scala:
##########
@@ -650,14 +657,27 @@ class SparkSession private(
private[sql] def sql(sqlText: String, args: Array[_], tracker: QueryPlanningTracker): DataFrame =
withActive {
val plan = tracker.measurePhase(QueryPlanningTracker.PARSING) {
- val parsedPlan = sessionState.sqlParser.parsePlan(sqlText)
- if (args.nonEmpty) {
- PosParameterizedQuery(parsedPlan, args.map(lit(_).expr).toImmutableArraySeq)
- } else {
- parsedPlan
+ val parsedPlan = sessionState.sqlParser.parseScript(sqlText)
+ parsedPlan match {
+ case CompoundBody(Seq(singleStmtPlan: SingleStatement), label) if args.nonEmpty =>
+ CompoundBody(Seq(SingleStatement(
+ PosParameterizedQuery(
+ singleStmtPlan.parsedPlan, args.map(lit(_).expr).toImmutableArraySeq))), label)
+ case p =>
+ assert(args.isEmpty, "Named parameters are not supported for batch queries")
+ p
}
}
- Dataset.ofRows(self, plan, tracker)
+
+ plan match {
+ case CompoundBody(Seq(singleStmtPlan: SingleStatement), _) =>
+ Dataset.ofRows(self, singleStmtPlan.parsedPlan, tracker)
+ case _ =>
+ // execute the plan directly if it is not a single statement
+ val lastRow = executeScript(plan).foldLeft(Array.empty[Row])((_, next) => next)
+ val attributes = DataTypeUtils.toAttributes(lastRow.head.schema)
+ Dataset.ofRows(self, LocalRelation.fromExternalRows(attributes, lastRow.toIndexedSeq))
Review Comment:
We have a hard time determining which statement is the last statement. That is the reason why we are doing it this way (we have to save the result of the last `dataframe`).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48344][SQL] SQL API change to support execution of compound statements [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #47403:
URL: https://github.com/apache/spark/pull/47403#discussion_r1687748510
##########
sql/core/src/main/scala/org/apache/spark/sql/SparkSession.scala:
##########
@@ -650,14 +657,27 @@ class SparkSession private(
private[sql] def sql(sqlText: String, args: Array[_], tracker: QueryPlanningTracker): DataFrame =
withActive {
val plan = tracker.measurePhase(QueryPlanningTracker.PARSING) {
- val parsedPlan = sessionState.sqlParser.parsePlan(sqlText)
- if (args.nonEmpty) {
- PosParameterizedQuery(parsedPlan, args.map(lit(_).expr).toImmutableArraySeq)
- } else {
- parsedPlan
+ val parsedPlan = sessionState.sqlParser.parseScript(sqlText)
+ parsedPlan match {
+ case CompoundBody(Seq(singleStmtPlan: SingleStatement), label) if args.nonEmpty =>
+ CompoundBody(Seq(SingleStatement(
+ PosParameterizedQuery(
+ singleStmtPlan.parsedPlan, args.map(lit(_).expr).toImmutableArraySeq))), label)
+ case p =>
+ assert(args.isEmpty, "Named parameters are not supported for batch queries")
+ p
}
}
- Dataset.ofRows(self, plan, tracker)
+
+ plan match {
+ case CompoundBody(Seq(singleStmtPlan: SingleStatement), _) =>
+ Dataset.ofRows(self, singleStmtPlan.parsedPlan, tracker)
+ case _ =>
+ // execute the plan directly if it is not a single statement
+ val lastRow = executeScript(plan).foldLeft(Array.empty[Row])((_, next) => next)
+ val attributes = DataTypeUtils.toAttributes(lastRow.head.schema)
+ Dataset.ofRows(self, LocalRelation.fromExternalRows(attributes, lastRow.toIndexedSeq))
Review Comment:
shall we return the last DataFrame from the script instead of collecting the result at the driver side?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [WIP][SPARK-][SQL] SQL API change to support execution of compound statements [spark]
Posted by "davidm-db (via GitHub)" <gi...@apache.org>.
davidm-db commented on PR #47403:
URL: https://github.com/apache/spark/pull/47403#issuecomment-2239871449
Just as a reminder, we should probably include SQL config with this PR as well! Should be fairly simple, but let's talk about it on Monday.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [WIP][SPARK-][SQL] SQL API change to support execution of compound statements [spark]
Posted by "davidm-db (via GitHub)" <gi...@apache.org>.
davidm-db commented on code in PR #47403:
URL: https://github.com/apache/spark/pull/47403#discussion_r1684084517
##########
sql/core/src/main/scala/org/apache/spark/sql/internal/SessionState.scala:
##########
@@ -19,10 +19,8 @@ package org.apache.spark.sql.internal
import java.io.File
import java.net.URI
-
Review Comment:
same here
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48344][SQL] SQL API change to support execution of compound statements [spark]
Posted by "miland-db (via GitHub)" <gi...@apache.org>.
miland-db commented on code in PR #47403:
URL: https://github.com/apache/spark/pull/47403#discussion_r1688194979
##########
sql/core/src/main/scala/org/apache/spark/sql/scripting/SqlScriptingInterpreter.scala:
##########
@@ -73,11 +74,19 @@ case class SqlScriptingInterpreter() {
.map(new SingleStatementExec(_, Origin(), isInternal = true))
.reverse
new CompoundBodyExec(
- body.collection.map(st => transformTreeIntoExecutable(st)) ++ dropVariables)
+ body.collection.map(st => transformTreeIntoExecutable(st)) ++ dropVariables, session)
case sparkStatement: SingleStatement =>
new SingleStatementExec(
sparkStatement.parsedPlan,
sparkStatement.origin,
- isInternal = false)
+ shouldCollectResult = true)
}
+
+ def execute(compoundBody: CompoundBody): Iterator[Array[Row]] = {
Review Comment:
We decided to do it this way because each statement has to be executed as soon as we encounter it because it can throw an exception. Very soon we are introducing `handlers/conditions` to deal with this exceptions.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48344][SQL] SQL API change to support execution of compound statements [spark]
Posted by "miland-db (via GitHub)" <gi...@apache.org>.
miland-db commented on code in PR #47403:
URL: https://github.com/apache/spark/pull/47403#discussion_r1688196798
##########
sql/core/src/main/scala/org/apache/spark/sql/SparkSession.scala:
##########
@@ -650,14 +657,27 @@ class SparkSession private(
private[sql] def sql(sqlText: String, args: Array[_], tracker: QueryPlanningTracker): DataFrame =
withActive {
val plan = tracker.measurePhase(QueryPlanningTracker.PARSING) {
- val parsedPlan = sessionState.sqlParser.parsePlan(sqlText)
- if (args.nonEmpty) {
- PosParameterizedQuery(parsedPlan, args.map(lit(_).expr).toImmutableArraySeq)
- } else {
- parsedPlan
+ val parsedPlan = sessionState.sqlParser.parseScript(sqlText)
+ parsedPlan match {
+ case CompoundBody(Seq(singleStmtPlan: SingleStatement), label) if args.nonEmpty =>
+ CompoundBody(Seq(SingleStatement(
+ PosParameterizedQuery(
+ singleStmtPlan.parsedPlan, args.map(lit(_).expr).toImmutableArraySeq))), label)
+ case p =>
+ assert(args.isEmpty, "Named parameters are not supported for batch queries")
+ p
}
}
- Dataset.ofRows(self, plan, tracker)
+
+ plan match {
+ case CompoundBody(Seq(singleStmtPlan: SingleStatement), _) =>
+ Dataset.ofRows(self, singleStmtPlan.parsedPlan, tracker)
+ case _ =>
+ // execute the plan directly if it is not a single statement
+ val lastRow = executeScript(plan).foldLeft(Array.empty[Row])((_, next) => next)
+ val attributes = DataTypeUtils.toAttributes(lastRow.head.schema)
+ Dataset.ofRows(self, LocalRelation.fromExternalRows(attributes, lastRow.toIndexedSeq))
Review Comment:
What is noop sink doing? Is it `df.write.format("noop").mode("overwrite").save()`? Is it the same as doing `df.collect()` but it just throws away the result?
We have a hard time determining which statement is the last statement. That is the reason why we are doing it this way (we have to save the result of the last `dataframe`).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [WIP][SPARK-][SQL] SQL API change to support execution of compound statements [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on code in PR #47403:
URL: https://github.com/apache/spark/pull/47403#discussion_r1683625797
##########
sql/core/src/main/scala/org/apache/spark/sql/SparkSession.scala:
##########
@@ -21,11 +21,9 @@ import java.io.Closeable
import java.util.{ServiceLoader, UUID}
import java.util.concurrent.TimeUnit._
import java.util.concurrent.atomic.{AtomicBoolean, AtomicReference}
-
Review Comment:
Let's revert those changes back to follow scalastyle for imports
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48344][SQL] SQL API change to support execution of compound statements [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #47403:
URL: https://github.com/apache/spark/pull/47403#discussion_r1688120792
##########
sql/core/src/main/scala/org/apache/spark/sql/scripting/SqlScriptingInterpreter.scala:
##########
@@ -73,11 +74,19 @@ case class SqlScriptingInterpreter() {
.map(new SingleStatementExec(_, Origin(), isInternal = true))
.reverse
new CompoundBodyExec(
- body.collection.map(st => transformTreeIntoExecutable(st)) ++ dropVariables)
+ body.collection.map(st => transformTreeIntoExecutable(st)) ++ dropVariables, session)
case sparkStatement: SingleStatement =>
new SingleStatementExec(
sparkStatement.parsedPlan,
sparkStatement.origin,
- isInternal = false)
+ shouldCollectResult = true)
}
+
+ def execute(compoundBody: CompoundBody): Iterator[Array[Row]] = {
Review Comment:
can we decouple the two parts: 1) produce `DataFrame`s from the script. 2) execute the `DataFrame`s one by one w.r.t. the order.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48344][SQL] SQL API change to support execution of compound statements [spark]
Posted by "davidm-db (via GitHub)" <gi...@apache.org>.
davidm-db commented on code in PR #47403:
URL: https://github.com/apache/spark/pull/47403#discussion_r1687861052
##########
sql/core/src/main/scala/org/apache/spark/sql/SparkSession.scala:
##########
@@ -650,14 +657,27 @@ class SparkSession private(
private[sql] def sql(sqlText: String, args: Array[_], tracker: QueryPlanningTracker): DataFrame =
withActive {
val plan = tracker.measurePhase(QueryPlanningTracker.PARSING) {
- val parsedPlan = sessionState.sqlParser.parsePlan(sqlText)
- if (args.nonEmpty) {
- PosParameterizedQuery(parsedPlan, args.map(lit(_).expr).toImmutableArraySeq)
- } else {
- parsedPlan
+ val parsedPlan = sessionState.sqlParser.parseScript(sqlText)
+ parsedPlan match {
+ case CompoundBody(Seq(singleStmtPlan: SingleStatement), label) if args.nonEmpty =>
+ CompoundBody(Seq(SingleStatement(
+ PosParameterizedQuery(
+ singleStmtPlan.parsedPlan, args.map(lit(_).expr).toImmutableArraySeq))), label)
+ case p =>
+ assert(args.isEmpty, "Named parameters are not supported for batch queries")
+ p
}
}
- Dataset.ofRows(self, plan, tracker)
+
+ plan match {
+ case CompoundBody(Seq(singleStmtPlan: SingleStatement), _) =>
+ Dataset.ofRows(self, singleStmtPlan.parsedPlan, tracker)
+ case _ =>
+ // execute the plan directly if it is not a single statement
+ val lastRow = executeScript(plan).foldLeft(Array.empty[Row])((_, next) => next)
+ val attributes = DataTypeUtils.toAttributes(lastRow.head.schema)
+ Dataset.ofRows(self, LocalRelation.fromExternalRows(attributes, lastRow.toIndexedSeq))
Review Comment:
this is a bit of a hard topic at the moment... we decided on this approach for preview for multiple reasons:
- this what we will do for initial version of the changes for Spark Connect execution
- discussions around multiple topics are still open:
- multiple results API - decisions around this will affect single results API as well
- multiple results API - from correctness perspective, all statements (including SELECTs) need to be executed eagerly. It makes sense then to have the same behavior with single results API as well - in this case all statements are executed eagerly, but results for all of them except the last one are dropped.
- there is still an open discussion whether to include `return` statement
- a ton of questions about stored procedures are still an open topic
what will probably happen down the line is:
- sql() API remains unchanged and only last DataFrame is returned (as you suggested). Requires still a lot of work to support Connect execution, current approach works with Connect already.
- [optional] new API to do what we are doing at the moment.
- new API for multiple results, stored procedures, execute immediate, etc.
since the last part is still an open question, we figured out that we will do a simplest thing that works e2e in all cases and then, after we gather initial feedback from preview, and understand better what we want to do for stored procedures/multiple results, we should actually commit to implement all of the API changes.
please let us know your thoughts on this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [WIP][SPARK-][SQL] SQL API change to support execution of compound statements [spark]
Posted by "davidm-db (via GitHub)" <gi...@apache.org>.
davidm-db commented on code in PR #47403:
URL: https://github.com/apache/spark/pull/47403#discussion_r1684105368
##########
sql/core/src/main/scala/org/apache/spark/sql/scripting/SqlScriptingExecutionNode.scala:
##########
@@ -80,6 +89,11 @@ class SingleStatementExec(
*/
var isExecuted = false
+ /**
+ * Data returned after execution.
+ */
+ var data: Option[Array[Row]] = None
Review Comment:
let's use something other than `data` - it seems pretty generic. something like `result` seems more suitable?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48344][SQL] SQL API change to support execution of compound statements [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #47403:
URL: https://github.com/apache/spark/pull/47403#discussion_r1688094627
##########
sql/core/src/main/scala/org/apache/spark/sql/SparkSession.scala:
##########
@@ -650,14 +657,27 @@ class SparkSession private(
private[sql] def sql(sqlText: String, args: Array[_], tracker: QueryPlanningTracker): DataFrame =
withActive {
val plan = tracker.measurePhase(QueryPlanningTracker.PARSING) {
- val parsedPlan = sessionState.sqlParser.parsePlan(sqlText)
- if (args.nonEmpty) {
- PosParameterizedQuery(parsedPlan, args.map(lit(_).expr).toImmutableArraySeq)
- } else {
- parsedPlan
+ val parsedPlan = sessionState.sqlParser.parseScript(sqlText)
+ parsedPlan match {
+ case CompoundBody(Seq(singleStmtPlan: SingleStatement), label) if args.nonEmpty =>
+ CompoundBody(Seq(SingleStatement(
+ PosParameterizedQuery(
+ singleStmtPlan.parsedPlan, args.map(lit(_).expr).toImmutableArraySeq))), label)
+ case p =>
+ assert(args.isEmpty, "Named parameters are not supported for batch queries")
+ p
}
}
- Dataset.ofRows(self, plan, tracker)
+
+ plan match {
+ case CompoundBody(Seq(singleStmtPlan: SingleStatement), _) =>
+ Dataset.ofRows(self, singleStmtPlan.parsedPlan, tracker)
+ case _ =>
+ // execute the plan directly if it is not a single statement
+ val lastRow = executeScript(plan).foldLeft(Array.empty[Row])((_, next) => next)
+ val attributes = DataTypeUtils.toAttributes(lastRow.head.schema)
+ Dataset.ofRows(self, LocalRelation.fromExternalRows(attributes, lastRow.toIndexedSeq))
Review Comment:
> all statements (including SELECTs) need to be executed eagerly
We need to materialize the query result but we don't need to collect the result at the driver side (except for the last statement). E.g. we can write the query result to a noop sink.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48344][SQL] SQL API change to support execution of compound statements [spark]
Posted by "davidm-db (via GitHub)" <gi...@apache.org>.
davidm-db commented on code in PR #47403:
URL: https://github.com/apache/spark/pull/47403#discussion_r1688249618
##########
sql/core/src/main/scala/org/apache/spark/sql/scripting/SqlScriptingInterpreter.scala:
##########
@@ -73,11 +74,19 @@ case class SqlScriptingInterpreter() {
.map(new SingleStatementExec(_, Origin(), isInternal = true))
.reverse
new CompoundBodyExec(
- body.collection.map(st => transformTreeIntoExecutable(st)) ++ dropVariables)
+ body.collection.map(st => transformTreeIntoExecutable(st)) ++ dropVariables, session)
case sparkStatement: SingleStatement =>
new SingleStatementExec(
sparkStatement.parsedPlan,
sparkStatement.origin,
- isInternal = false)
+ shouldCollectResult = true)
}
+
+ def execute(compoundBody: CompoundBody): Iterator[Array[Row]] = {
Review Comment:
We did it the way you suggested before, but as soon as we started introducing exception handling, we figured out it's hardly extensible in the future. When SQL statements throws an exception, exception handlers need to be executed immediately, and we need interpreter for that - hence, we need to do execution within the interpreter as well.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48344][SQL] SQL API change to support execution of compound statements [spark]
Posted by "davidm-db (via GitHub)" <gi...@apache.org>.
davidm-db commented on code in PR #47403:
URL: https://github.com/apache/spark/pull/47403#discussion_r1686579965
##########
sql/core/src/main/scala/org/apache/spark/sql/SparkSession.scala:
##########
@@ -650,14 +657,27 @@ class SparkSession private(
private[sql] def sql(sqlText: String, args: Array[_], tracker: QueryPlanningTracker): DataFrame =
withActive {
val plan = tracker.measurePhase(QueryPlanningTracker.PARSING) {
- val parsedPlan = sessionState.sqlParser.parsePlan(sqlText)
- if (args.nonEmpty) {
- PosParameterizedQuery(parsedPlan, args.map(lit(_).expr).toImmutableArraySeq)
- } else {
- parsedPlan
+ val parsedPlan = sessionState.sqlParser.parseScript(sqlText)
+ parsedPlan match {
+ case CompoundBody(Seq(singleStmtPlan: SingleStatement), label) if args.nonEmpty =>
+ CompoundBody(List(SingleStatement(
Review Comment:
nit: `Seq` instead of `List`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48344][SQL] SQL API change to support execution of compound statements [spark]
Posted by "davidm-db (via GitHub)" <gi...@apache.org>.
davidm-db commented on code in PR #47403:
URL: https://github.com/apache/spark/pull/47403#discussion_r1687861052
##########
sql/core/src/main/scala/org/apache/spark/sql/SparkSession.scala:
##########
@@ -650,14 +657,27 @@ class SparkSession private(
private[sql] def sql(sqlText: String, args: Array[_], tracker: QueryPlanningTracker): DataFrame =
withActive {
val plan = tracker.measurePhase(QueryPlanningTracker.PARSING) {
- val parsedPlan = sessionState.sqlParser.parsePlan(sqlText)
- if (args.nonEmpty) {
- PosParameterizedQuery(parsedPlan, args.map(lit(_).expr).toImmutableArraySeq)
- } else {
- parsedPlan
+ val parsedPlan = sessionState.sqlParser.parseScript(sqlText)
+ parsedPlan match {
+ case CompoundBody(Seq(singleStmtPlan: SingleStatement), label) if args.nonEmpty =>
+ CompoundBody(Seq(SingleStatement(
+ PosParameterizedQuery(
+ singleStmtPlan.parsedPlan, args.map(lit(_).expr).toImmutableArraySeq))), label)
+ case p =>
+ assert(args.isEmpty, "Named parameters are not supported for batch queries")
+ p
}
}
- Dataset.ofRows(self, plan, tracker)
+
+ plan match {
+ case CompoundBody(Seq(singleStmtPlan: SingleStatement), _) =>
+ Dataset.ofRows(self, singleStmtPlan.parsedPlan, tracker)
+ case _ =>
+ // execute the plan directly if it is not a single statement
+ val lastRow = executeScript(plan).foldLeft(Array.empty[Row])((_, next) => next)
+ val attributes = DataTypeUtils.toAttributes(lastRow.head.schema)
+ Dataset.ofRows(self, LocalRelation.fromExternalRows(attributes, lastRow.toIndexedSeq))
Review Comment:
this is a bit of a hard topic at the moment... we decided on this approach for preview for multiple reasons:
- this what we will do for initial version of the changes for Spark Connect execution
- discussions around multiple topics are still open:
- multiple results API - decisions around this will affect single results API as well
- multiple results API - from correctness perspective, all statements (including SELECTs) need to be executed eagerly. It makes sense then to have the same behavior with single results API as well - in this case all statements are executed eagerly, but results for all of them except the last one) are dropped.
- there is still an open discussion whether to include `return` statement
- a ton of questions about stored procedures are still an open topic
what will probably happen down the line is:
- sql() API remains unchanged and only last DataFrame is returned (as you suggested). Requires still a lot of work to support Connect execution, current approach works with Connect already.
- [optional] new API to do what we are doing at the moment.
- new API for multiple results, stored procedures, execute immediate, etc.
since the last part is still an open question, we figured out that we will do a simplest thing that works e2e in all cases and then, after we gather initial feedback from preview, and understand better what we want to do for stored procedures/multiple results, we should actually commit to implement all of the API changes.
please let us know your thoughts on this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [WIP][SPARK-][SQL] SQL API change to support execution of compound statements [spark]
Posted by "miland-db (via GitHub)" <gi...@apache.org>.
miland-db commented on code in PR #47403:
URL: https://github.com/apache/spark/pull/47403#discussion_r1684244839
##########
sql/core/src/main/scala/org/apache/spark/sql/scripting/SqlScriptingInterpreter.scala:
##########
@@ -70,14 +71,22 @@ case class SqlScriptingInterpreter() {
}
val dropVariables = variables
.map(varName => DropVariable(varName, ifExists = true))
- .map(new SingleStatementExec(_, Origin(), isInternal = true))
+ .map(new SingleStatementExec(_, Origin(), isInternal = true, collectResult = false))
.reverse
new CompoundBodyExec(
- body.collection.map(st => transformTreeIntoExecutable(st)) ++ dropVariables)
+ body.collection.map(st => transformTreeIntoExecutable(st)) ++ dropVariables, session)
case sparkStatement: SingleStatement =>
new SingleStatementExec(
sparkStatement.parsedPlan,
sparkStatement.origin,
isInternal = false)
}
+
+ def execute(executionPlan: Iterator[CompoundStatementExec]): Iterator[Array[Row]] = {
Review Comment:
We are using `buildExecutionPlan` in `SqlScriptingInterpreterSuite`. Should we change the testing logic as well?
We can call `buildExecutionPlan` from `execute` but can we also leave it to be public?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48344][SQL] SQL API change to support execution of compound statements [spark]
Posted by "davidm-db (via GitHub)" <gi...@apache.org>.
davidm-db commented on code in PR #47403:
URL: https://github.com/apache/spark/pull/47403#discussion_r1700377123
##########
sql/core/src/main/scala/org/apache/spark/sql/scripting/SqlScriptingInterpreter.scala:
##########
@@ -73,11 +74,19 @@ case class SqlScriptingInterpreter() {
.map(new SingleStatementExec(_, Origin(), isInternal = true))
.reverse
new CompoundBodyExec(
- body.collection.map(st => transformTreeIntoExecutable(st)) ++ dropVariables)
+ body.collection.map(st => transformTreeIntoExecutable(st)) ++ dropVariables, session)
case sparkStatement: SingleStatement =>
new SingleStatementExec(
sparkStatement.parsedPlan,
sparkStatement.origin,
- isInternal = false)
+ shouldCollectResult = true)
}
+
+ def execute(compoundBody: CompoundBody): Iterator[Array[Row]] = {
Review Comment:
nit: let's keep public stuff at top, and private below
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [WIP][SPARK-][SQL] SQL API change to support execution of compound statements [spark]
Posted by "davidm-db (via GitHub)" <gi...@apache.org>.
davidm-db commented on code in PR #47403:
URL: https://github.com/apache/spark/pull/47403#discussion_r1684100460
##########
sql/core/src/main/scala/org/apache/spark/sql/scripting/SqlScriptingExecutionNode.scala:
##########
@@ -152,6 +173,46 @@ abstract class CompoundNestedStatementIteratorExec(collection: Seq[CompoundState
* Executable node for CompoundBody.
* @param statements
* Executable nodes for nested statements within the CompoundBody.
+ * @param session
+ * Spark session.
*/
-class CompoundBodyExec(statements: Seq[CompoundStatementExec])
- extends CompoundNestedStatementIteratorExec(statements)
+class CompoundBodyExec(
+ statements: Seq[CompoundStatementExec],
+ session: SparkSession = null)
Review Comment:
seems like we don't want to have any default value for `session` here?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [WIP][SPARK-][SQL] SQL API change to support execution of compound statements [spark]
Posted by "davidm-db (via GitHub)" <gi...@apache.org>.
davidm-db commented on code in PR #47403:
URL: https://github.com/apache/spark/pull/47403#discussion_r1684094681
##########
sql/core/src/main/scala/org/apache/spark/sql/scripting/SqlScriptingExecutionNode.scala:
##########
@@ -102,10 +122,11 @@ class SingleStatementExec(
abstract class CompoundNestedStatementIteratorExec(collection: Seq[CompoundStatementExec])
extends NonLeafStatementExec {
- private var localIterator = collection.iterator
- private var curr = if (localIterator.hasNext) Some(localIterator.next()) else None
+ protected var localIterator: Iterator[CompoundStatementExec] = collection.iterator
+ protected var curr: Option[CompoundStatementExec] =
+ if (localIterator.hasNext) Some(localIterator.next()) else None
- private lazy val treeIterator: Iterator[CompoundStatementExec] =
+ protected lazy val treeIterator: Iterator[CompoundStatementExec] =
Review Comment:
let's remove `treeIterator` implementation from `CompoundNestedStatementIteratorExec` for now, and let's then figure out offline if we need `CompoundNestedStatementIteratorExec` at all in the future
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48344][SQL] SQL API change to support execution of compound statements [spark]
Posted by "davidm-db (via GitHub)" <gi...@apache.org>.
davidm-db commented on code in PR #47403:
URL: https://github.com/apache/spark/pull/47403#discussion_r1686758572
##########
sql/core/src/main/scala/org/apache/spark/sql/scripting/SqlScriptingExecutionNode.scala:
##########
@@ -91,21 +105,41 @@ class SingleStatementExec(
}
override def reset(): Unit = isExecuted = false
+
+ def execute(session: SparkSession): Unit = {
Review Comment:
`override def execute...`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48344][SQL] SQL API change to support execution of compound statements [spark]
Posted by "davidm-db (via GitHub)" <gi...@apache.org>.
davidm-db commented on code in PR #47403:
URL: https://github.com/apache/spark/pull/47403#discussion_r1686780223
##########
sql/core/src/main/scala/org/apache/spark/sql/scripting/SqlScriptingExecutionNode.scala:
##########
@@ -91,21 +105,41 @@ class SingleStatementExec(
}
override def reset(): Unit = isExecuted = false
+
+ def execute(session: SparkSession): Unit = {
+ isExecuted = true
+ val rows = Some(Dataset.ofRows(session, parsedPlan).collect())
+ if (shouldCollectResult) {
+ result = rows
+ }
+ }
}
/**
- * Abstract class for all statements that contain nested statements.
- * Implements recursive iterator logic over all child execution nodes.
- * @param collection
- * Collection of child execution nodes.
+ * Executable node for CompoundBody.
+ * @param statements
+ * Executable nodes for nested statements within the CompoundBody.
+ * @param session
+ * Spark session.
*/
-abstract class CompoundNestedStatementIteratorExec(collection: Seq[CompoundStatementExec])
+class CompoundBodyExec(
+ statements: Seq[CompoundStatementExec],
+ session: SparkSession)
extends NonLeafStatementExec {
- private var localIterator = collection.iterator
- private var curr = if (localIterator.hasNext) Some(localIterator.next()) else None
+ protected var localIterator: Iterator[CompoundStatementExec] = statements.iterator
+ protected var curr: Option[CompoundStatementExec] =
+ if (localIterator.hasNext) Some(localIterator.next()) else None
+
+ def getTreeIterator: Iterator[CompoundStatementExec] = treeIterator
- private lazy val treeIterator: Iterator[CompoundStatementExec] =
+ override def reset(): Unit = {
+ statements.foreach(_.reset())
+ localIterator = statements.iterator
+ curr = if (localIterator.hasNext) Some(localIterator.next()) else None
+ }
+
+ protected lazy val treeIterator: Iterator[CompoundStatementExec] =
Review Comment:
nit: can be `private` now.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [WIP][SPARK-][SQL] SQL API change to support execution of compound statements [spark]
Posted by "davidm-db (via GitHub)" <gi...@apache.org>.
davidm-db commented on code in PR #47403:
URL: https://github.com/apache/spark/pull/47403#discussion_r1684077871
##########
sql/core/src/main/scala/org/apache/spark/sql/internal/BaseSessionStateBuilder.scala:
##########
@@ -146,6 +147,15 @@ abstract class BaseSessionStateBuilder(
extensions.buildParser(session, new SparkSqlParser())
}
+ /**
+ * Script interpreter that produces execution plan for sql batch procedural language.
Review Comment:
maybe change this message to something cleaner, like "Interpreter that produces execution plan and executes SQL scripts.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [WIP][SPARK-][SQL] SQL API change to support execution of compound statements [spark]
Posted by "davidm-db (via GitHub)" <gi...@apache.org>.
davidm-db commented on code in PR #47403:
URL: https://github.com/apache/spark/pull/47403#discussion_r1684111561
##########
sql/core/src/main/scala/org/apache/spark/sql/scripting/SqlScriptingInterpreter.scala:
##########
@@ -70,14 +71,22 @@ case class SqlScriptingInterpreter() {
}
val dropVariables = variables
.map(varName => DropVariable(varName, ifExists = true))
- .map(new SingleStatementExec(_, Origin(), isInternal = true))
+ .map(new SingleStatementExec(_, Origin(), isInternal = true, collectResult = false))
.reverse
new CompoundBodyExec(
- body.collection.map(st => transformTreeIntoExecutable(st)) ++ dropVariables)
+ body.collection.map(st => transformTreeIntoExecutable(st)) ++ dropVariables, session)
case sparkStatement: SingleStatement =>
new SingleStatementExec(
sparkStatement.parsedPlan,
sparkStatement.origin,
isInternal = false)
}
+
+ def execute(executionPlan: Iterator[CompoundStatementExec]): Iterator[Array[Row]] = {
Review Comment:
I would maybe make `buildExecutionPlan` private, and than call it from `execute` - seems like we will never need to call `buildExecutionPlan` from outside?
for example, if we add separate `execute*` function in the future, it can also call `buildExecutionPlan` internally.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [WIP][SPARK-][SQL] SQL API change to support execution of compound statements [spark]
Posted by "davidm-db (via GitHub)" <gi...@apache.org>.
davidm-db commented on code in PR #47403:
URL: https://github.com/apache/spark/pull/47403#discussion_r1684284889
##########
sql/core/src/main/scala/org/apache/spark/sql/scripting/SqlScriptingInterpreter.scala:
##########
@@ -70,14 +71,22 @@ case class SqlScriptingInterpreter() {
}
val dropVariables = variables
.map(varName => DropVariable(varName, ifExists = true))
- .map(new SingleStatementExec(_, Origin(), isInternal = true))
+ .map(new SingleStatementExec(_, Origin(), isInternal = true, collectResult = false))
.reverse
new CompoundBodyExec(
- body.collection.map(st => transformTreeIntoExecutable(st)) ++ dropVariables)
+ body.collection.map(st => transformTreeIntoExecutable(st)) ++ dropVariables, session)
case sparkStatement: SingleStatement =>
new SingleStatementExec(
sparkStatement.parsedPlan,
sparkStatement.origin,
isInternal = false)
}
+
+ def execute(executionPlan: Iterator[CompoundStatementExec]): Iterator[Array[Row]] = {
Review Comment:
This probably means that we should change the tests a bit, we are simulating `execute` in the tests as well, because before we didn't have it in the interpreter. Now that we have it, we should probably change tests.
Let's do what you said last for now, until other folks review the PR (leave this comment unresolved) and then we can figure if we want to change tests as well.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48344][SQL] SQL API change to support execution of compound statements [spark]
Posted by "davidm-db (via GitHub)" <gi...@apache.org>.
davidm-db commented on code in PR #47403:
URL: https://github.com/apache/spark/pull/47403#discussion_r1698602295
##########
sql/core/src/main/scala/org/apache/spark/sql/SparkSession.scala:
##########
@@ -650,14 +657,27 @@ class SparkSession private(
private[sql] def sql(sqlText: String, args: Array[_], tracker: QueryPlanningTracker): DataFrame =
withActive {
val plan = tracker.measurePhase(QueryPlanningTracker.PARSING) {
- val parsedPlan = sessionState.sqlParser.parsePlan(sqlText)
- if (args.nonEmpty) {
- PosParameterizedQuery(parsedPlan, args.map(lit(_).expr).toImmutableArraySeq)
- } else {
- parsedPlan
+ val parsedPlan = sessionState.sqlParser.parseScript(sqlText)
+ parsedPlan match {
+ case CompoundBody(Seq(singleStmtPlan: SingleStatement), label) if args.nonEmpty =>
+ CompoundBody(Seq(SingleStatement(
+ PosParameterizedQuery(
+ singleStmtPlan.parsedPlan, args.map(lit(_).expr).toImmutableArraySeq))), label)
+ case p =>
+ assert(args.isEmpty, "Named parameters are not supported for batch queries")
+ p
}
}
- Dataset.ofRows(self, plan, tracker)
+
+ plan match {
+ case CompoundBody(Seq(singleStmtPlan: SingleStatement), _) =>
+ Dataset.ofRows(self, singleStmtPlan.parsedPlan, tracker)
+ case _ =>
+ // execute the plan directly if it is not a single statement
+ val lastRow = executeScript(plan).foldLeft(Array.empty[Row])((_, next) => next)
Review Comment:
let's think if we want to do this exactly this way, because:
- `executeScript` is basically a simple one-liner and alias for interpreter's `execute` function
- when we introduce multiple results in the future, it seems best to:
- have `executeMultipleResults` in the interpreter
- each function (`execute` and `executeMultipleResults` and maybe something new?) should collect data based on the type of data it needs to return
I propose that `execute` family of methods in the interpreter should be responsible to handle the logic of which data is returned, instead of fetching last row here in `SparkSession`.
I didn't write a ton of details here, I'm writing this comment as a reminder and we can discuss more offline.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48344][SQL] SQL API change to support execution of compound statements [spark]
Posted by "davidm-db (via GitHub)" <gi...@apache.org>.
davidm-db commented on code in PR #47403:
URL: https://github.com/apache/spark/pull/47403#discussion_r1699605843
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/SqlScriptingParserSuite.scala:
##########
@@ -356,4 +357,20 @@ class SqlScriptingParserSuite extends SparkFunSuite with SQLHelper {
.replace("END", "")
.trim
}
+
+ test("SQL Scripting not enabled") {
Review Comment:
nit: move this above the `// Helper methods`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48344][SQL] SQL API change to support execution of compound statements [spark]
Posted by "davidm-db (via GitHub)" <gi...@apache.org>.
davidm-db commented on PR #47403:
URL: https://github.com/apache/spark/pull/47403#issuecomment-2252098868
@cloud-fan any unresolved topics here? Milan won't be available next week - I can finish the PR next week if needed, but let's try to figure out if there are any concerns so Milan can try to sort them out today.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [WIP][SPARK-][SQL] SQL API change to support execution of compound statements [spark]
Posted by "davidm-db (via GitHub)" <gi...@apache.org>.
davidm-db commented on code in PR #47403:
URL: https://github.com/apache/spark/pull/47403#discussion_r1684082230
##########
sql/core/src/main/scala/org/apache/spark/sql/internal/BaseSessionStateBuilder.scala:
##########
@@ -146,6 +147,15 @@ abstract class BaseSessionStateBuilder(
extensions.buildParser(session, new SparkSqlParser())
}
+ /**
+ * Script interpreter that produces execution plan for sql batch procedural language.
+ *
+ * Note: this depends on the `conf` field.
Review Comment:
do we know what this part of the comment actually means? I see that it's being reused around, but not sure what it means exactly... I mean, I don't see that `conf` is used anywhere in `scriptingInterpreter` creation...
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [WIP][SPARK-][SQL] SQL API change to support execution of compound statements [spark]
Posted by "davidm-db (via GitHub)" <gi...@apache.org>.
davidm-db commented on code in PR #47403:
URL: https://github.com/apache/spark/pull/47403#discussion_r1684082573
##########
sql/core/src/main/scala/org/apache/spark/sql/SparkSessionExtensions.scala:
##########
@@ -18,7 +18,6 @@
package org.apache.spark.sql
import scala.collection.mutable
-
Review Comment:
revert this empty line as well
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [WIP][SPARK-][SQL] SQL API change to support execution of compound statements [spark]
Posted by "miland-db (via GitHub)" <gi...@apache.org>.
miland-db commented on code in PR #47403:
URL: https://github.com/apache/spark/pull/47403#discussion_r1684244839
##########
sql/core/src/main/scala/org/apache/spark/sql/scripting/SqlScriptingInterpreter.scala:
##########
@@ -70,14 +71,22 @@ case class SqlScriptingInterpreter() {
}
val dropVariables = variables
.map(varName => DropVariable(varName, ifExists = true))
- .map(new SingleStatementExec(_, Origin(), isInternal = true))
+ .map(new SingleStatementExec(_, Origin(), isInternal = true, collectResult = false))
.reverse
new CompoundBodyExec(
- body.collection.map(st => transformTreeIntoExecutable(st)) ++ dropVariables)
+ body.collection.map(st => transformTreeIntoExecutable(st)) ++ dropVariables, session)
case sparkStatement: SingleStatement =>
new SingleStatementExec(
sparkStatement.parsedPlan,
sparkStatement.origin,
isInternal = false)
}
+
+ def execute(executionPlan: Iterator[CompoundStatementExec]): Iterator[Array[Row]] = {
Review Comment:
We are using `buildExecutionPlan` in `SqlScriptingInterpreterSuite`. Should we change the testing logic as well then?
We can call `buildExecutionPlan` from `execute` but can we also leave it to be public?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [WIP][SPARK-][SQL] SQL API change to support execution of compound statements [spark]
Posted by "miland-db (via GitHub)" <gi...@apache.org>.
miland-db commented on code in PR #47403:
URL: https://github.com/apache/spark/pull/47403#discussion_r1684225488
##########
sql/core/src/main/scala/org/apache/spark/sql/scripting/SqlScriptingExecutionNode.scala:
##########
@@ -91,6 +105,12 @@ class SingleStatementExec(
}
override def reset(): Unit = isExecuted = false
+
+ def execute(session: SparkSession): Unit = {
+ isExecuted = true
+ val result = Some(Dataset.ofRows(session, parsedPlan).collect())
+ if (collectResult) data = result
Review Comment:
Done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48344][SQL] SQL API change to support execution of compound statements [spark]
Posted by "davidm-db (via GitHub)" <gi...@apache.org>.
davidm-db commented on code in PR #47403:
URL: https://github.com/apache/spark/pull/47403#discussion_r1700272636
##########
sql/core/src/main/scala/org/apache/spark/sql/internal/BaseSessionStateBuilder.scala:
##########
@@ -146,6 +147,13 @@ abstract class BaseSessionStateBuilder(
extensions.buildParser(session, new SparkSqlParser())
}
+ /**
+ * Script interpreter that produces execution plan and executes SQL scripts.
+ */
+ protected lazy val scriptingInterpreter: SqlScriptingInterpreter = {
Review Comment:
nit: `sqlScriptingInterpreter` maybe?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48344][SQL] SQL API change to support execution of compound statements [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #47403:
URL: https://github.com/apache/spark/pull/47403#discussion_r1689275665
##########
sql/core/src/main/scala/org/apache/spark/sql/scripting/SqlScriptingInterpreter.scala:
##########
@@ -73,11 +74,19 @@ case class SqlScriptingInterpreter() {
.map(new SingleStatementExec(_, Origin(), isInternal = true))
.reverse
new CompoundBodyExec(
- body.collection.map(st => transformTreeIntoExecutable(st)) ++ dropVariables)
+ body.collection.map(st => transformTreeIntoExecutable(st)) ++ dropVariables, session)
case sparkStatement: SingleStatement =>
new SingleStatementExec(
sparkStatement.parsedPlan,
sparkStatement.origin,
- isInternal = false)
+ shouldCollectResult = true)
}
+
+ def execute(compoundBody: CompoundBody): Iterator[Array[Row]] = {
Review Comment:
I'm OK with a simple implementation (collect all the result rows) as a beginning, but we should build the basic framework with proper abstraction, otherwise I don't know how to review it. The idea of `handlers` makes sense.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48344][SQL] SQL API change to support execution of compound statements [spark]
Posted by "miland-db (via GitHub)" <gi...@apache.org>.
miland-db commented on code in PR #47403:
URL: https://github.com/apache/spark/pull/47403#discussion_r1691277611
##########
sql/core/src/main/scala/org/apache/spark/sql/scripting/SqlScriptingInterpreter.scala:
##########
@@ -73,11 +74,19 @@ case class SqlScriptingInterpreter() {
.map(new SingleStatementExec(_, Origin(), isInternal = true))
.reverse
new CompoundBodyExec(
- body.collection.map(st => transformTreeIntoExecutable(st)) ++ dropVariables)
+ body.collection.map(st => transformTreeIntoExecutable(st)) ++ dropVariables, session)
case sparkStatement: SingleStatement =>
new SingleStatementExec(
sparkStatement.parsedPlan,
sparkStatement.origin,
- isInternal = false)
+ shouldCollectResult = true)
}
+
+ def execute(compoundBody: CompoundBody): Iterator[Array[Row]] = {
Review Comment:
For now, because we don't know which one is the last statement to do only one `collect()` and use noop sink for the rest, we decided to do collect for all statements.
Either way, it is important to execute each statement as soon as we encounter it to be able to handle errors properly. [PR introducing handlers](https://github.com/apache/spark/pull/47423) is currently work in progress and will probably explain why we did things the way we did in this PR.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48344][SQL] SQL API change to support execution of compound statements [spark]
Posted by "davidm-db (via GitHub)" <gi...@apache.org>.
davidm-db commented on code in PR #47403:
URL: https://github.com/apache/spark/pull/47403#discussion_r1700093041
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -3344,6 +3344,14 @@ object SQLConf {
.version("2.3.0")
.fallbackConf(org.apache.spark.internal.config.STRING_REDACTION_PATTERN)
+ val SQL_SCRIPTING_ENABLED =
+ buildConf("spark.sql.scripting.enabled")
+ .doc("SQL Scripting feature is under development and its use should be done under this" +
+ "feature flag.")
+ .version("4.0.0")
+ .booleanConf
+ .createWithDefault(Utils.isTesting)
Review Comment:
Did we talk about changing this to `false` by default? Why did we decide not to do it?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [WIP][SPARK-][SQL] SQL API change to support execution of compound statements [spark]
Posted by "davidm-db (via GitHub)" <gi...@apache.org>.
davidm-db commented on code in PR #47403:
URL: https://github.com/apache/spark/pull/47403#discussion_r1684096518
##########
sql/core/src/main/scala/org/apache/spark/sql/scripting/SqlScriptingExecutionNode.scala:
##########
@@ -91,6 +105,12 @@ class SingleStatementExec(
}
override def reset(): Unit = isExecuted = false
+
+ def execute(session: SparkSession): Unit = {
+ isExecuted = true
+ val result = Some(Dataset.ofRows(session, parsedPlan).collect())
+ if (collectResult) data = result
Review Comment:
let's write it as
```
if (collectResult) {
data = result
}
```
even in such simple cases, for consistency and cleaner code
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48344][SQL] SQL API change to support execution of compound statements [spark]
Posted by "davidm-db (via GitHub)" <gi...@apache.org>.
davidm-db commented on code in PR #47403:
URL: https://github.com/apache/spark/pull/47403#discussion_r1686757282
##########
sql/core/src/main/scala/org/apache/spark/sql/scripting/SqlScriptingExecutionNode.scala:
##########
@@ -71,14 +85,14 @@ trait NonLeafStatementExec extends CompoundStatementExec {
class SingleStatementExec(
var parsedPlan: LogicalPlan,
override val origin: Origin,
- override val isInternal: Boolean)
+ override val isInternal: Boolean = false,
+ val shouldCollectResult: Boolean = false)
Review Comment:
Add doc comment for `shouldCollectResult`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [WIP][SPARK-][SQL] SQL API change to support execution of compound statements [spark]
Posted by "miland-db (via GitHub)" <gi...@apache.org>.
miland-db commented on code in PR #47403:
URL: https://github.com/apache/spark/pull/47403#discussion_r1684223652
##########
sql/core/src/main/scala/org/apache/spark/sql/internal/BaseSessionStateBuilder.scala:
##########
@@ -146,6 +147,15 @@ abstract class BaseSessionStateBuilder(
extensions.buildParser(session, new SparkSqlParser())
}
+ /**
+ * Script interpreter that produces execution plan for sql batch procedural language.
+ *
+ * Note: this depends on the `conf` field.
Review Comment:
Removed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48344][SQL] SQL API change to support execution of compound statements [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #47403:
URL: https://github.com/apache/spark/pull/47403#discussion_r1689271592
##########
sql/core/src/main/scala/org/apache/spark/sql/SparkSession.scala:
##########
@@ -650,14 +657,27 @@ class SparkSession private(
private[sql] def sql(sqlText: String, args: Array[_], tracker: QueryPlanningTracker): DataFrame =
withActive {
val plan = tracker.measurePhase(QueryPlanningTracker.PARSING) {
- val parsedPlan = sessionState.sqlParser.parsePlan(sqlText)
- if (args.nonEmpty) {
- PosParameterizedQuery(parsedPlan, args.map(lit(_).expr).toImmutableArraySeq)
- } else {
- parsedPlan
+ val parsedPlan = sessionState.sqlParser.parseScript(sqlText)
+ parsedPlan match {
+ case CompoundBody(Seq(singleStmtPlan: SingleStatement), label) if args.nonEmpty =>
+ CompoundBody(Seq(SingleStatement(
+ PosParameterizedQuery(
+ singleStmtPlan.parsedPlan, args.map(lit(_).expr).toImmutableArraySeq))), label)
+ case p =>
+ assert(args.isEmpty, "Named parameters are not supported for batch queries")
+ p
}
}
- Dataset.ofRows(self, plan, tracker)
+
+ plan match {
+ case CompoundBody(Seq(singleStmtPlan: SingleStatement), _) =>
+ Dataset.ofRows(self, singleStmtPlan.parsedPlan, tracker)
+ case _ =>
+ // execute the plan directly if it is not a single statement
+ val lastRow = executeScript(plan).foldLeft(Array.empty[Row])((_, next) => next)
+ val attributes = DataTypeUtils.toAttributes(lastRow.head.schema)
+ Dataset.ofRows(self, LocalRelation.fromExternalRows(attributes, lastRow.toIndexedSeq))
Review Comment:
Yea it's the same as doing `df.collect()` but it just throws away the result
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org