You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "cloud-fan (via GitHub)" <gi...@apache.org> on 2025/09/11 14:31:03 UTC
Re: [PR] [SPARK-53444][SQL] Rework execute immediate [spark]
cloud-fan commented on code in PR #52173:
URL: https://github.com/apache/spark/pull/52173#discussion_r2341079844
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/parameters.scala:
##########
@@ -203,6 +229,69 @@ object BindParameters extends Rule[LogicalPlan] with QueryErrorsBase {
args(posToIndex(pos))
}
+ case GeneralParameterizedQuery(child, args, paramNames)
+ if !child.containsPattern(UNRESOLVED_WITH) &&
+ args.forall(_.resolved) =>
+
+ // Check all arguments for validity (args are already evaluated expressions/literals)
+ val allArgs = if (paramNames.isEmpty) {
+ args.zipWithIndex.map { case (arg, idx) => s"_$idx" -> arg }
+ } else {
+ args.zip(paramNames).map { case (arg, name) => name -> arg }
+ }
+ checkArgs(allArgs)
+
+ // Collect parameter types used in the query to enforce invariants
+ var hasNamedParam = false
+ var hasPositionalParam = false
+ val positionalParams = scala.collection.mutable.Set.empty[Int]
Review Comment:
oh, since we need to track this set anyway, we can remove `hasPositionalParam`
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveExecuteImmediate.scala:
##########
@@ -0,0 +1,202 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.analysis
+
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.SqlScriptingContextManager
+import org.apache.spark.sql.catalyst.expressions.{Alias, EmptyRow, Expression, Literal,
+ VariableReference}
+import org.apache.spark.sql.catalyst.plans.logical.{Command, CompoundBody, LogicalPlan, SetVariable}
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.catalyst.trees.TreePattern.EXECUTE_IMMEDIATE
+import org.apache.spark.sql.connector.catalog.CatalogManager
+import org.apache.spark.sql.errors.QueryCompilationErrors
+import org.apache.spark.sql.types.StringType
+
+/**
+ * Analysis rule that resolves and executes EXECUTE IMMEDIATE statements during analysis,
+ * replacing them with the results, similar to how CALL statements work.
+ * This rule combines resolution and execution in a single pass.
+ */
+case class ResolveExecuteImmediate(sparkSession: SparkSession, catalogManager: CatalogManager)
+ extends Rule[LogicalPlan] {
+
+ override def apply(plan: LogicalPlan): LogicalPlan = {
+ plan.resolveOperatorsWithPruning(_.containsPattern(EXECUTE_IMMEDIATE), ruleId) {
+ case node @ UnresolvedExecuteImmediate(sqlStmtStr, args, targetVariables) =>
+ if (sqlStmtStr.resolved && targetVariables.forall(_.resolved) && args.forall(_.resolved)) {
+ // All resolved - execute immediately and handle INTO clause if present
+ if (targetVariables.nonEmpty) {
+ // EXECUTE IMMEDIATE ... INTO should generate SetVariable plan with eagerly executed
+ // source
+ val finalTargetVars = resolveTargetVariables(targetVariables)
+ val executedSource = executeImmediateQuery(sqlStmtStr, args, hasIntoClause = true)
+ SetVariable(finalTargetVars, executedSource)
+ } else {
+ // Regular EXECUTE IMMEDIATE without INTO - execute and return result directly
+ executeImmediateQuery(sqlStmtStr, args, hasIntoClause = false)
+ }
+ } else {
+ // Not all resolved yet - wait for next iteration
+ node
+ }
+ }
+ }
+
+ private def resolveTargetVariables(targetVariables: Seq[Expression]): Seq[VariableReference] = {
Review Comment:
`extractTargetVariables` seems a better name, as they are already resolved.
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveExecuteImmediate.scala:
##########
@@ -0,0 +1,202 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.analysis
+
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.SqlScriptingContextManager
+import org.apache.spark.sql.catalyst.expressions.{Alias, EmptyRow, Expression, Literal,
+ VariableReference}
+import org.apache.spark.sql.catalyst.plans.logical.{Command, CompoundBody, LogicalPlan, SetVariable}
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.catalyst.trees.TreePattern.EXECUTE_IMMEDIATE
+import org.apache.spark.sql.connector.catalog.CatalogManager
+import org.apache.spark.sql.errors.QueryCompilationErrors
+import org.apache.spark.sql.types.StringType
+
+/**
+ * Analysis rule that resolves and executes EXECUTE IMMEDIATE statements during analysis,
+ * replacing them with the results, similar to how CALL statements work.
+ * This rule combines resolution and execution in a single pass.
+ */
+case class ResolveExecuteImmediate(sparkSession: SparkSession, catalogManager: CatalogManager)
+ extends Rule[LogicalPlan] {
+
+ override def apply(plan: LogicalPlan): LogicalPlan = {
+ plan.resolveOperatorsWithPruning(_.containsPattern(EXECUTE_IMMEDIATE), ruleId) {
+ case node @ UnresolvedExecuteImmediate(sqlStmtStr, args, targetVariables) =>
+ if (sqlStmtStr.resolved && targetVariables.forall(_.resolved) && args.forall(_.resolved)) {
+ // All resolved - execute immediately and handle INTO clause if present
+ if (targetVariables.nonEmpty) {
+ // EXECUTE IMMEDIATE ... INTO should generate SetVariable plan with eagerly executed
+ // source
+ val finalTargetVars = resolveTargetVariables(targetVariables)
+ val executedSource = executeImmediateQuery(sqlStmtStr, args, hasIntoClause = true)
+ SetVariable(finalTargetVars, executedSource)
+ } else {
+ // Regular EXECUTE IMMEDIATE without INTO - execute and return result directly
+ executeImmediateQuery(sqlStmtStr, args, hasIntoClause = false)
+ }
+ } else {
+ // Not all resolved yet - wait for next iteration
+ node
+ }
+ }
+ }
+
+ private def resolveTargetVariables(targetVariables: Seq[Expression]): Seq[VariableReference] = {
+ targetVariables.map {
+ case alias: Alias =>
+ // Extract the VariableReference from the alias
+ alias.child match {
+ case varRef: VariableReference =>
+ // Use resolved VariableReference directly with canFold = false
+ varRef.copy(canFold = false)
+ case _ =>
+ throw QueryCompilationErrors.unsupportedParameterExpression(alias.child)
+ }
+ case varRef: VariableReference =>
+ // Use resolved VariableReference directly with canFold = false
+ varRef.copy(canFold = false)
+ case other =>
+ throw QueryCompilationErrors.unsupportedParameterExpression(other)
+ }
+ }
+
+ private def executeImmediateQuery(
+ sqlStmtStr: Expression,
+ args: Seq[Expression],
+ hasIntoClause: Boolean): LogicalPlan = {
+ // Extract the query string from the queryParam expression
+ val sqlString = extractQueryString(sqlStmtStr)
+
+ // Parse and validate the query
+ val parsedPlan = sparkSession.sessionState.sqlParser.parsePlan(sqlString)
+ validateQuery(sqlString, parsedPlan)
+
+ // Execute the query recursively with isolated local variable context
+ val result = if (args.isEmpty) {
+ // No parameters - execute directly
+ withIsolatedLocalVariableContext {
+ sparkSession.sql(sqlString)
+ }
+ } else {
+ // For parameterized queries, build parameter arrays
+ val (paramValues, paramNames) = buildUnifiedParameters(args)
+
+ withIsolatedLocalVariableContext {
+ sparkSession.asInstanceOf[org.apache.spark.sql.classic.SparkSession]
+ .sql(sqlString, paramValues, paramNames)
+ }
+ }
+
+ // If this EXECUTE IMMEDIATE has an INTO clause, commands are not allowed
+ if (hasIntoClause && result.queryExecution.analyzed.isInstanceOf[Command]) {
+ throw QueryCompilationErrors.invalidStatementForExecuteInto(sqlString)
+ }
+
+ result.queryExecution.analyzed
+ }
+
+ private def extractQueryString(queryExpr: Expression): String = {
+ // Ensure the expression resolves to string type
+ if (!queryExpr.dataType.sameType(StringType)) {
+ throw QueryCompilationErrors.invalidExecuteImmediateExpressionType(queryExpr.dataType)
+ }
+
+ // Evaluate the expression to get the query string
+ val value = queryExpr.eval(null)
+ if (value == null) {
+ // Extract the original text from the expression's origin for the error message
+ val originalText = extractOriginalText(queryExpr)
+ throw QueryCompilationErrors.nullSQLStringExecuteImmediate(originalText)
+ }
+
+ value.toString
+ }
+
+ private def extractOriginalText(queryExpr: Expression): String = {
+ val origin = queryExpr.origin
+ // Try to extract the original text from the origin information
+ (origin.sqlText, origin.startIndex, origin.stopIndex) match {
+ case (Some(sqlText), Some(startIndex), Some(stopIndex)) =>
+ // Extract the substring from the original SQL text
+ sqlText.substring(startIndex, stopIndex + 1)
+ case _ =>
+ // Fallback to the SQL representation if origin information is not available
+ queryExpr.sql
+ }
+ }
+
+ private def validateQuery(queryString: String, parsedPlan: LogicalPlan): Unit = {
+ // Check for compound bodies (SQL scripting)
+ if (parsedPlan.isInstanceOf[CompoundBody]) {
+ throw QueryCompilationErrors.sqlScriptInExecuteImmediate(queryString)
+ }
+ }
+
+ /**
+ * Builds parameter arrays for the sql() API.
+ */
+ private def buildUnifiedParameters(args: Seq[Expression]): (Array[Any], Array[String]) = {
+ val values = scala.collection.mutable.ListBuffer[Any]()
+ val names = scala.collection.mutable.ListBuffer[String]()
+
+ args.foreach {
+ case alias: Alias =>
+ val paramValue = evaluateParameterExpression(alias.child)
+ values += paramValue
+ names += alias.name
+ case expr =>
+ // Positional parameter: just a value
+ val paramValue = evaluateParameterExpression(expr)
+ values += paramValue
+ names += null // unnamed expression
Review Comment:
shall we use empty string?
##########
sql/core/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveExecuteImmediate.scala:
##########
@@ -0,0 +1,202 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.analysis
+
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.SqlScriptingContextManager
+import org.apache.spark.sql.catalyst.expressions.{Alias, EmptyRow, Expression, Literal,
+ VariableReference}
+import org.apache.spark.sql.catalyst.plans.logical.{Command, CompoundBody, LogicalPlan, SetVariable}
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.catalyst.trees.TreePattern.EXECUTE_IMMEDIATE
+import org.apache.spark.sql.connector.catalog.CatalogManager
+import org.apache.spark.sql.errors.QueryCompilationErrors
+import org.apache.spark.sql.types.StringType
+
+/**
+ * Analysis rule that resolves and executes EXECUTE IMMEDIATE statements during analysis,
+ * replacing them with the results, similar to how CALL statements work.
+ * This rule combines resolution and execution in a single pass.
+ */
+case class ResolveExecuteImmediate(sparkSession: SparkSession, catalogManager: CatalogManager)
+ extends Rule[LogicalPlan] {
+
+ override def apply(plan: LogicalPlan): LogicalPlan = {
+ plan.resolveOperatorsWithPruning(_.containsPattern(EXECUTE_IMMEDIATE), ruleId) {
+ case node @ UnresolvedExecuteImmediate(sqlStmtStr, args, targetVariables) =>
+ if (sqlStmtStr.resolved && targetVariables.forall(_.resolved) && args.forall(_.resolved)) {
+ // All resolved - execute immediately and handle INTO clause if present
+ if (targetVariables.nonEmpty) {
+ // EXECUTE IMMEDIATE ... INTO should generate SetVariable plan with eagerly executed
+ // source
+ val finalTargetVars = resolveTargetVariables(targetVariables)
+ val executedSource = executeImmediateQuery(sqlStmtStr, args, hasIntoClause = true)
+ SetVariable(finalTargetVars, executedSource)
+ } else {
+ // Regular EXECUTE IMMEDIATE without INTO - execute and return result directly
+ executeImmediateQuery(sqlStmtStr, args, hasIntoClause = false)
+ }
+ } else {
+ // Not all resolved yet - wait for next iteration
+ node
+ }
+ }
+ }
+
+ private def resolveTargetVariables(targetVariables: Seq[Expression]): Seq[VariableReference] = {
+ targetVariables.map {
+ case alias: Alias =>
+ // Extract the VariableReference from the alias
+ alias.child match {
+ case varRef: VariableReference =>
+ // Use resolved VariableReference directly with canFold = false
+ varRef.copy(canFold = false)
+ case _ =>
+ throw QueryCompilationErrors.unsupportedParameterExpression(alias.child)
+ }
+ case varRef: VariableReference =>
+ // Use resolved VariableReference directly with canFold = false
+ varRef.copy(canFold = false)
+ case other =>
+ throw QueryCompilationErrors.unsupportedParameterExpression(other)
+ }
+ }
+
+ private def executeImmediateQuery(
+ sqlStmtStr: Expression,
+ args: Seq[Expression],
+ hasIntoClause: Boolean): LogicalPlan = {
+ // Extract the query string from the queryParam expression
+ val sqlString = extractQueryString(sqlStmtStr)
+
+ // Parse and validate the query
+ val parsedPlan = sparkSession.sessionState.sqlParser.parsePlan(sqlString)
+ validateQuery(sqlString, parsedPlan)
+
+ // Execute the query recursively with isolated local variable context
+ val result = if (args.isEmpty) {
+ // No parameters - execute directly
+ withIsolatedLocalVariableContext {
+ sparkSession.sql(sqlString)
+ }
+ } else {
+ // For parameterized queries, build parameter arrays
+ val (paramValues, paramNames) = buildUnifiedParameters(args)
+
+ withIsolatedLocalVariableContext {
+ sparkSession.asInstanceOf[org.apache.spark.sql.classic.SparkSession]
+ .sql(sqlString, paramValues, paramNames)
+ }
+ }
+
+ // If this EXECUTE IMMEDIATE has an INTO clause, commands are not allowed
+ if (hasIntoClause && result.queryExecution.analyzed.isInstanceOf[Command]) {
+ throw QueryCompilationErrors.invalidStatementForExecuteInto(sqlString)
+ }
+
+ result.queryExecution.analyzed
+ }
+
+ private def extractQueryString(queryExpr: Expression): String = {
+ // Ensure the expression resolves to string type
+ if (!queryExpr.dataType.sameType(StringType)) {
+ throw QueryCompilationErrors.invalidExecuteImmediateExpressionType(queryExpr.dataType)
+ }
+
+ // Evaluate the expression to get the query string
+ val value = queryExpr.eval(null)
+ if (value == null) {
+ // Extract the original text from the expression's origin for the error message
+ val originalText = extractOriginalText(queryExpr)
+ throw QueryCompilationErrors.nullSQLStringExecuteImmediate(originalText)
+ }
+
+ value.toString
+ }
+
+ private def extractOriginalText(queryExpr: Expression): String = {
+ val origin = queryExpr.origin
+ // Try to extract the original text from the origin information
+ (origin.sqlText, origin.startIndex, origin.stopIndex) match {
+ case (Some(sqlText), Some(startIndex), Some(stopIndex)) =>
+ // Extract the substring from the original SQL text
+ sqlText.substring(startIndex, stopIndex + 1)
+ case _ =>
+ // Fallback to the SQL representation if origin information is not available
+ queryExpr.sql
+ }
+ }
+
+ private def validateQuery(queryString: String, parsedPlan: LogicalPlan): Unit = {
+ // Check for compound bodies (SQL scripting)
+ if (parsedPlan.isInstanceOf[CompoundBody]) {
+ throw QueryCompilationErrors.sqlScriptInExecuteImmediate(queryString)
+ }
+ }
+
+ /**
+ * Builds parameter arrays for the sql() API.
+ */
+ private def buildUnifiedParameters(args: Seq[Expression]): (Array[Any], Array[String]) = {
+ val values = scala.collection.mutable.ListBuffer[Any]()
+ val names = scala.collection.mutable.ListBuffer[String]()
+
+ args.foreach {
+ case alias: Alias =>
+ val paramValue = evaluateParameterExpression(alias.child)
+ values += paramValue
+ names += alias.name
+ case expr =>
+ // Positional parameter: just a value
+ val paramValue = evaluateParameterExpression(expr)
+ values += paramValue
+ names += null // unnamed expression
+ }
+
+ (values.toArray, names.toArray)
+ }
+
+ /**
+ * Evaluates a parameter expression. Validation for unsupported constructs like subqueries
+ * is already done during analysis in ResolveExecuteImmediate.validateExpressions().
+ */
+ private def evaluateParameterExpression(expr: Expression): Any = {
+ expr match {
+ case varRef: VariableReference =>
+ // Variable references should be evaluated to their values
+ varRef.eval(EmptyRow)
+ case foldable if foldable.foldable =>
+ Literal.create(foldable.eval(EmptyRow), foldable.dataType).value
+ case other =>
+ // Expression is not foldable - not supported for parameters
Review Comment:
this includes subquery expressions?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/parameters.scala:
##########
@@ -104,6 +104,31 @@ case class PosParameterizedQuery(child: LogicalPlan, args: Seq[Expression])
copy(child = newChild)
}
+/**
+ * The logical plan representing a parameterized query with general parameter support.
+ * This allows the query to use either positional or named parameters based on the
+ * parameter markers found in the query, with optional parameter names provided.
+ *
+ * @param child The parameterized logical plan.
+ * @param args The literal values or collection constructor functions such as `map()`,
+ * `array()`, `struct()` of parameters.
+ * @param paramNames Optional parameter names corresponding to args. If provided for an argument,
+ * that argument can be used for named parameter binding. If not provided
+ * parameters are treated as positional.
+ */
+case class GeneralParameterizedQuery(
+ child: LogicalPlan,
+ args: Seq[Expression],
+ paramNames: Seq[String])
+ extends ParameterizedQuery(child) {
+ assert(args.nonEmpty)
+ assert(paramNames.isEmpty || paramNames.length == args.length,
Review Comment:
going through the entire PR, I think we guarantee that `paramNames` is always the same length of `args`. If an arg has no name, the corresponding string in `paramNames` will be null/empty string.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/parameters.scala:
##########
@@ -203,6 +229,69 @@ object BindParameters extends Rule[LogicalPlan] with QueryErrorsBase {
args(posToIndex(pos))
}
+ case GeneralParameterizedQuery(child, args, paramNames)
+ if !child.containsPattern(UNRESOLVED_WITH) &&
+ args.forall(_.resolved) =>
+
+ // Check all arguments for validity (args are already evaluated expressions/literals)
+ val allArgs = if (paramNames.isEmpty) {
+ args.zipWithIndex.map { case (arg, idx) => s"_$idx" -> arg }
+ } else {
+ args.zip(paramNames).map { case (arg, name) => name -> arg }
+ }
+ checkArgs(allArgs)
+
+ // Collect parameter types used in the query to enforce invariants
+ var hasNamedParam = false
+ var hasPositionalParam = false
+ val positionalParams = scala.collection.mutable.Set.empty[Int]
+ bind(child) {
+ case p @ NamedParameter(_) => hasNamedParam = true; p
+ case p @ PosParameter(pos) =>
+ hasPositionalParam = true
+ positionalParams.add(pos)
+ p
+ }
+
+ // Validate: no mixing of positional and named parameters
+ if (hasNamedParam && hasPositionalParam) {
+ throw QueryCompilationErrors.invalidQueryMixedQueryParameters()
+ }
+
+ // Validate: if query uses named parameters, all USING expressions must have names
+ if (hasNamedParam && !hasPositionalParam) {
+ if (paramNames.isEmpty) {
+ // Query uses named parameters but no USING expressions provided
+ throw QueryCompilationErrors.invalidQueryAllParametersMustBeNamed(Seq.empty)
+ } else {
+ // Check that all USING expressions have names
+ val unnamedExpressions = paramNames.zipWithIndex.collect {
+ case (null, index) => index
+ case ("", index) => index // empty strings are unnamed
Review Comment:
let's be consistent with `ResolveExecuteImmediate`. My proposal is to use empty string, as null is discouraged in Scala.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/parameters.scala:
##########
@@ -203,6 +229,69 @@ object BindParameters extends Rule[LogicalPlan] with QueryErrorsBase {
args(posToIndex(pos))
}
+ case GeneralParameterizedQuery(child, args, paramNames)
+ if !child.containsPattern(UNRESOLVED_WITH) &&
+ args.forall(_.resolved) =>
+
+ // Check all arguments for validity (args are already evaluated expressions/literals)
+ val allArgs = if (paramNames.isEmpty) {
Review Comment:
Sorry I was wrong. `paramNames` can't be empty if `args` is not empty. So this should be
```
args.zip(paramNames).zipWithIndex.map { case ((arg, name), index) =>
val finalName = if (name.isEmpty) s"_$idx" else name
finalName -> arg
}
```
--
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