You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "allisonwang-db (via GitHub)" <gi...@apache.org> on 2025/01/13 22:01:24 UTC
[PR] [SPARK-50763][SQL] Add Analyzer rule for resolving SQL table UDFs [spark]
allisonwang-db opened a new pull request, #49471:
URL: https://github.com/apache/spark/pull/49471
<!--
Thanks for sending a pull request! Here are some tips for you:
1. If this is your first time, please read our contributor guidelines: https://spark.apache.org/contributing.html
2. Ensure you have added or run the appropriate tests for your PR: https://spark.apache.org/developer-tools.html
3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP][SPARK-XXXX] Your PR title ...'.
4. Be sure to keep the PR description updated to reflect all changes.
5. Please write your PR title to summarize what this PR proposes.
6. If possible, provide a concise example to reproduce the issue for a faster review.
7. If you want to add a new configuration, please read the guideline first for naming configurations in
'core/src/main/scala/org/apache/spark/internal/config/ConfigEntry.scala'.
8. If you want to add or modify an error type or message, please read the guideline first in
'common/utils/src/main/resources/error/README.md'.
-->
### What changes were proposed in this pull request?
<!--
Please clarify what changes you are proposing. The purpose of this section is to outline the changes and how this PR fixes the issue.
If possible, please consider writing useful notes for better and faster reviews in your PR. See the examples below.
1. If you refactor some codes with changing classes, showing the class hierarchy will help reviewers.
2. If you fix some SQL features, you can provide some references of other DBMSes.
3. If there is design documentation, please add the link.
4. If there is a discussion in the mailing list, please add the link.
-->
This PR adds an Analyzer rule to resolve SQL user-defined table functions.
### Why are the changes needed?
<!--
Please clarify why the changes are needed. For instance,
1. If you propose a new API, clarify the use case for a new API.
2. If you fix a bug, you can clarify why it is a bug.
-->
### Does this PR introduce _any_ user-facing change?
<!--
Note that it means *any* user-facing change including all aspects such as the documentation fix.
If yes, please clarify the previous behavior and the change this PR proposes - provide the console output, description and/or an example to show the behavior difference if possible.
If possible, please also clarify if this is a user-facing change compared to the released Spark versions or within the unreleased branches such as master.
If no, write 'No'.
-->
### How was this patch tested?
<!--
If tests were added, say they were added here. Please make sure to add some test cases that check the changes thoroughly including negative and positive cases if possible.
If it was tested in a way different from regular unit tests, please clarify how you tested step by step, ideally copy and paste-able, so that other reviewers can test and check, and descendants can verify in the future.
If tests were not added, please describe why they were not added and/or why it was difficult to add.
If benchmark tests were added, please run the benchmarks in GitHub Actions for the consistent environment, and the instructions could accord to: https://spark.apache.org/developer-tools.html#github-workflow-benchmarks.
-->
### Was this patch authored or co-authored using generative AI tooling?
<!--
If generative AI tooling has been used in the process of authoring this patch, please include the
phrase: 'Generated-by: ' followed by the name of the tool and its version.
If no, write 'No'.
Please refer to the [ASF Generative Tooling Guidance](https://www.apache.org/legal/generative-tooling.html) for details.
-->
--
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-50763][SQL] Add Analyzer rule for resolving SQL table functions [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #49471:
URL: https://github.com/apache/spark/pull/49471#discussion_r1928167189
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala:
##########
@@ -2474,13 +2475,13 @@ class Analyzer(override val catalogManager: CatalogManager) extends RuleExecutor
* Extract aggregate expressions from the given expression and replace
* them with attribute references.
* Example:
- * Before: foo(c1) + foo(max(c2)) + max(foo(c2))
- * After: foo(c1) + foo(max_c2) + max_foo_c2
- * Extracted expressions: [c1, max(c2) AS max_c2, max(foo(c2)) AS max_foo_c2]
+ * Before: foo(c1) + foo(max(c2)) + max(foo(c2))
+ * After: foo(c1) + foo(max_c2) + max_foo_c2
+ * Extracted expressions: [c1, max(c2) AS max_c2, max(foo(c2)) AS max_foo_c2]
*/
private def extractAndRewrite[T <: Expression](
- expression: T,
- extractedExprs: ArrayBuffer[NamedExpression]): T = {
+ expression: T,
+ extractedExprs: ArrayBuffer[NamedExpression]): T = {
Review Comment:
ditto
--
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-50763][SQL] Add Analyzer rule for resolving SQL table functions [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #49471:
URL: https://github.com/apache/spark/pull/49471#discussion_r1928170343
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala:
##########
@@ -2636,6 +2637,93 @@ class Analyzer(override val catalogManager: CatalogManager) extends RuleExecutor
}
}
+ /*
+ * This rule resolves SQL table functions.
+ */
+ object ResolveSQLTableFunctions extends Rule[LogicalPlan] with AliasHelper {
+
+ /**
+ * Check if a subquery plan is subject to the COUNT bug that can cause wrong results.
+ * A lateral correlation can only be removed if the lateral subquery is not subject to
+ * the COUNT bug. Currently only lateral correlation can handle it correctly.
+ */
+ private def hasCountBug(sub: LogicalPlan): Boolean = sub.find {
+ // The COUNT bug occurs when there is an Aggregate that satisfies all the following
+ // conditions:
+ // 1) is on the correlation path
+ // 2) has non-empty group by expressions
+ // 3) has one or more output columns that evaluate to non-null values with empty input.
+ // E.g: COUNT(empty row) = 0.
+ // For simplicity, we use a stricter criteria (1 and 2 only) to determine if a query
+ // is subject to the COUNT bug.
+ case a: Aggregate if a.groupingExpressions.nonEmpty => hasOuterReferences(a.child)
+ case _ => false
+ }.nonEmpty
+
+ /**
+ * Rewrite a resolved SQL table function plan by removing unnecessary lateral joins:
+ * Before:
+ * LateralJoin lateral-subquery [a], Inner
+ * : +- Project [c1, c2]
+ * : +- Filter [outer(a) == c1]
+ * : +- Relation [c1, c2]
+ * +- Project [1 AS a]
+ * +- OneRowRelation
+ * After:
+ * Project [c1, c2]
+ * +- Filter [1 == c1] <---- Replaced outer(a)
+ * +- Relation [c1, c2]
+ */
+ private def rewrite(plan: LogicalPlan): LogicalPlan = {
+ (plan transformUp {
+ case j @ LateralJoin(Project(aliases, _: OneRowRelation), sub: LateralSubquery, Inner, None)
+ if j.resolved && aliases.forall(_.deterministic) =>
+ val attrMap = AttributeMap(aliases.collect { case a: Alias => a.toAttribute -> a.child })
+ val newPlan = sub.plan.transformAllExpressionsWithPruning(
+ _.containsPattern(OUTER_REFERENCE)) {
+ // Avoid replacing outer references that do not belong to the current outer plan.
+ // This can happen if the child of an alias also contains outer references (nested
+ // table function references). E.g:
+ // LateralJoin
+ // : +- Filter [outer(a) == x]
+ // : +- Relation [x, y]
+ // +- Project [outer(c) AS a]
+ // +- OneRowRelation
+ case OuterReference(a: Attribute) if attrMap.contains(a) => attrMap(a) match {
+ case ne: NamedExpression => ne
+ case o => Alias(o, a.name)(exprId = a.exprId, qualifier = a.qualifier)
Review Comment:
instead of doing an extra `transformWithPruning`, shall we call `trimAlias(o)` 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-50763][SQL] Add Analyzer rule for resolving SQL table functions [spark]
Posted by "allisonwang-db (via GitHub)" <gi...@apache.org>.
allisonwang-db commented on code in PR #49471:
URL: https://github.com/apache/spark/pull/49471#discussion_r1929184608
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala:
##########
@@ -2515,7 +2516,7 @@ class Analyzer(override val catalogManager: CatalogManager) extends RuleExecutor
// Return if a sub-tree does not contain SQLFunctionExpression.
case p: LogicalPlan if !p.containsPattern(SQL_FUNCTION_EXPRESSION) => p
- case f @ Filter(cond, a: Aggregate)
+ case f@Filter(cond, a: Aggregate)
Review Comment:
Yea let me revert these unrelated changes.
--
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-50763][SQL] Add Analyzer rule for resolving SQL table functions [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #49471:
URL: https://github.com/apache/spark/pull/49471#discussion_r1990231759
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala:
##########
@@ -1675,6 +1676,86 @@ class SessionCatalog(
}
}
+ /**
+ * Constructs a SQL table function plan.
+ * This function should be invoked with the captured SQL configs from the function.
+ *
+ * Example SQL table function:
+ *
+ * CREATE FUNCTION foo(x INT) RETURNS TABLE(a INT) RETURN SELECT x + 1 AS x1
+ *
+ * Query:
+ *
+ * SELECT * FROM foo(1);
+ *
+ * Plan:
+ *
+ * Project [CAST(x1 AS INT) AS a]
+ * +- LateralJoin lateral-subquery [x]
+ * : +- Project [(outer(x) + 1) AS x1]
+ * : +- OneRowRelation
+ * +- Project [CAST(1 AS INT) AS x]
+ * +- OneRowRelation
+ */
+ def makeSQLTableFunctionPlan(
+ name: String,
+ function: SQLFunction,
+ input: Seq[Expression],
+ outputAttrs: Seq[Attribute]): LogicalPlan = {
+ assert(function.isTableFunc)
+ val funcName = function.name.funcName
+ val inputParam = function.inputParam
+ val returnParam = function.getTableFuncReturnCols
+ val (_, query) = function.getExpressionAndQuery(parser, isTableFunc = true)
+ assert(query.isDefined)
+
+ // Check function arguments
+ val paramSize = inputParam.map(_.size).getOrElse(0)
+ if (input.size > paramSize) {
+ throw QueryCompilationErrors.wrongNumArgsError(
+ name, paramSize.toString, input.size)
+ }
+
+ val body = if (inputParam.isDefined) {
+ val param = inputParam.get
+ // Attributes referencing the input parameters inside the function can use the
+ // function name as a qualifier.
+ val qualifier = Seq(funcName)
+ val paddedInput = input ++
+ param.takeRight(paramSize - input.size).map { p =>
Review Comment:
let's add a test to prove that default parameter fails as of 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] [SPARK-50763][SQL] Add Analyzer rule for resolving SQL table functions [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan closed pull request #49471: [SPARK-50763][SQL] Add Analyzer rule for resolving SQL table functions
URL: https://github.com/apache/spark/pull/49471
--
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-50763][SQL] Add Analyzer rule for resolving SQL table functions [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #49471:
URL: https://github.com/apache/spark/pull/49471#discussion_r1982863500
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala:
##########
@@ -1675,6 +1676,91 @@ class SessionCatalog(
}
}
+ /**
+ * Constructs a SQL table function plan.
+ *
+ * Example SQL table function:
+ *
+ * CREATE FUNCTION foo(x INT) RETURNS TABLE(a INT) RETURN SELECT x + 1 AS x1
+ *
+ * Query:
+ *
+ * SELECT * FROM foo(1);
+ *
+ * Plan:
+ *
+ * Project [CAST(x1 AS INT) AS a]
+ * +- LateralJoin lateral-subquery [x]
+ * : +- Project [(outer(x) + 1) AS x1]
+ * : +- OneRowRelation
+ * +- Project [CAST(1 AS INT) AS x]
+ * +- OneRowRelation
+ */
+ def makeSQLTableFunctionPlan(
+ name: String,
+ function: SQLFunction,
+ input: Seq[Expression],
+ outputAttrs: Seq[Attribute]): LogicalPlan = {
+ assert(function.isTableFunc)
+ val funcName = function.name.funcName
+
+ // Use captured SQL configs when parsing a SQL function.
+ val conf = new SQLConf()
+ function.getSQLConfigs.foreach { case (k, v) => conf.settings.put(k, v) }
+ SQLConf.withExistingConf(conf) {
+ val inputParam = function.inputParam
+ val returnParam = function.getTableFuncReturnCols
+ val (_, query) = function.getExpressionAndQuery(parser, isTableFunc = true)
+ assert(query.isDefined)
+
+ // Check function arguments
+ val paramSize = inputParam.map(_.size).getOrElse(0)
+ if (input.size > paramSize) {
+ throw QueryCompilationErrors.wrongNumArgsError(
+ name, paramSize.toString, input.size)
+ }
+
+ val body = if (inputParam.isDefined) {
+ val param = inputParam.get
+ // Attributes referencing the input parameters inside the function can use the
+ // function name as a qualifier.
+ val qualifier = Seq(funcName)
+ val paddedInput = input ++
+ param.takeRight(paramSize - input.size).map { p =>
+ val defaultExpr = p.getDefault()
+ if (defaultExpr.isDefined) {
+ Cast(parseDefault(defaultExpr.get, parser), p.dataType)
+ } else {
+ throw QueryCompilationErrors.wrongNumArgsError(
+ name, paramSize.toString, input.size)
+ }
+ }
+
+ val inputCast = paddedInput.zip(param.fields).map {
+ case (expr, param) =>
+ // Add outer references to all attributes in the function input.
+ val outer = expr.transform {
+ case a: Attribute => OuterReference(a)
+ }
+ Alias(Cast(outer, param.dataType), param.name)(qualifier = qualifier)
Review Comment:
So a simple way is we don't add cast in https://github.com/apache/spark/pull/49471/files#diff-9dd0899e5406230aeff96654432da54f35255f6dc60eecb87264a5c508a8c826R1732
--
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-50763][SQL] Add Analyzer rule for resolving SQL table functions [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #49471:
URL: https://github.com/apache/spark/pull/49471#discussion_r1928167396
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala:
##########
@@ -2515,7 +2516,7 @@ class Analyzer(override val catalogManager: CatalogManager) extends RuleExecutor
// Return if a sub-tree does not contain SQLFunctionExpression.
case p: LogicalPlan if !p.containsPattern(SQL_FUNCTION_EXPRESSION) => p
- case f @ Filter(cond, a: Aggregate)
+ case f@Filter(cond, a: Aggregate)
Review Comment:
the previous code style is more correct
--
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-50763][SQL] Add Analyzer rule for resolving SQL table functions [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #49471:
URL: https://github.com/apache/spark/pull/49471#discussion_r1928166999
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala:
##########
@@ -2430,8 +2431,8 @@ class Analyzer(override val catalogManager: CatalogManager) extends RuleExecutor
* function inputs into the given project list.
*/
private def rewriteSQLFunctions[E <: Expression](
- expression: E,
- projectList: ArrayBuffer[NamedExpression]): E = {
+ expression: E,
Review Comment:
nit: 4 spaces indentation
--
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-50763][SQL] Add Analyzer rule for resolving SQL table functions [spark]
Posted by "allisonwang-db (via GitHub)" <gi...@apache.org>.
allisonwang-db commented on code in PR #49471:
URL: https://github.com/apache/spark/pull/49471#discussion_r1985588345
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala:
##########
@@ -1675,6 +1676,86 @@ class SessionCatalog(
}
}
+ /**
+ * Constructs a SQL table function plan.
+ * This function should be invoked with the captured SQL configs from the function.
+ *
+ * Example SQL table function:
+ *
+ * CREATE FUNCTION foo(x INT) RETURNS TABLE(a INT) RETURN SELECT x + 1 AS x1
+ *
+ * Query:
+ *
+ * SELECT * FROM foo(1);
+ *
+ * Plan:
+ *
+ * Project [CAST(x1 AS INT) AS a]
+ * +- LateralJoin lateral-subquery [x]
+ * : +- Project [(outer(x) + 1) AS x1]
+ * : +- OneRowRelation
+ * +- Project [CAST(1 AS INT) AS x]
+ * +- OneRowRelation
+ */
+ def makeSQLTableFunctionPlan(
+ name: String,
+ function: SQLFunction,
+ input: Seq[Expression],
+ outputAttrs: Seq[Attribute]): LogicalPlan = {
+ assert(function.isTableFunc)
+ val funcName = function.name.funcName
+ val inputParam = function.inputParam
+ val returnParam = function.getTableFuncReturnCols
+ val (_, query) = function.getExpressionAndQuery(parser, isTableFunc = true)
+ assert(query.isDefined)
+
+ // Check function arguments
+ val paramSize = inputParam.map(_.size).getOrElse(0)
+ if (input.size > paramSize) {
+ throw QueryCompilationErrors.wrongNumArgsError(
+ name, paramSize.toString, input.size)
+ }
+
+ val body = if (inputParam.isDefined) {
+ val param = inputParam.get
+ // Attributes referencing the input parameters inside the function can use the
+ // function name as a qualifier.
+ val qualifier = Seq(funcName)
+ val paddedInput = input ++
+ param.takeRight(paramSize - input.size).map { p =>
Review Comment:
This `defaultRearrange` is only invoked when the there are named arguments. The rearrangeArguments will simply return the input expression if there is no named arguments:
https://github.com/apache/spark/blob/1c14b06f8224d0f5db072bdeabb1f36d7359b05a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala#L1792-L1796
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala:
##########
@@ -1675,6 +1676,86 @@ class SessionCatalog(
}
}
+ /**
+ * Constructs a SQL table function plan.
+ * This function should be invoked with the captured SQL configs from the function.
+ *
+ * Example SQL table function:
+ *
+ * CREATE FUNCTION foo(x INT) RETURNS TABLE(a INT) RETURN SELECT x + 1 AS x1
+ *
+ * Query:
+ *
+ * SELECT * FROM foo(1);
+ *
+ * Plan:
+ *
+ * Project [CAST(x1 AS INT) AS a]
+ * +- LateralJoin lateral-subquery [x]
+ * : +- Project [(outer(x) + 1) AS x1]
+ * : +- OneRowRelation
+ * +- Project [CAST(1 AS INT) AS x]
+ * +- OneRowRelation
+ */
+ def makeSQLTableFunctionPlan(
+ name: String,
+ function: SQLFunction,
+ input: Seq[Expression],
+ outputAttrs: Seq[Attribute]): LogicalPlan = {
+ assert(function.isTableFunc)
+ val funcName = function.name.funcName
+ val inputParam = function.inputParam
+ val returnParam = function.getTableFuncReturnCols
+ val (_, query) = function.getExpressionAndQuery(parser, isTableFunc = true)
+ assert(query.isDefined)
+
+ // Check function arguments
+ val paramSize = inputParam.map(_.size).getOrElse(0)
+ if (input.size > paramSize) {
+ throw QueryCompilationErrors.wrongNumArgsError(
+ name, paramSize.toString, input.size)
+ }
+
+ val body = if (inputParam.isDefined) {
+ val param = inputParam.get
+ // Attributes referencing the input parameters inside the function can use the
+ // function name as a qualifier.
+ val qualifier = Seq(funcName)
+ val paddedInput = input ++
+ param.takeRight(paramSize - input.size).map { p =>
Review Comment:
Actually I realized the default argument is not really properly supported:
```
scala> spark.sql("""create function foo(x int default 0) returns int return x""")
org.apache.spark.sql.catalyst.parser.ParseException:
[PARSE_SYNTAX_ERROR] Syntax error at or near 'default'. SQLSTATE: 42601
```
I will create a separate PR to make this work.
--
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-50763][SQL] Add Analyzer rule for resolving SQL table functions [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #49471:
URL: https://github.com/apache/spark/pull/49471#discussion_r1982862272
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala:
##########
@@ -1675,6 +1676,91 @@ class SessionCatalog(
}
}
+ /**
+ * Constructs a SQL table function plan.
+ *
+ * Example SQL table function:
+ *
+ * CREATE FUNCTION foo(x INT) RETURNS TABLE(a INT) RETURN SELECT x + 1 AS x1
+ *
+ * Query:
+ *
+ * SELECT * FROM foo(1);
+ *
+ * Plan:
+ *
+ * Project [CAST(x1 AS INT) AS a]
+ * +- LateralJoin lateral-subquery [x]
+ * : +- Project [(outer(x) + 1) AS x1]
+ * : +- OneRowRelation
+ * +- Project [CAST(1 AS INT) AS x]
+ * +- OneRowRelation
+ */
+ def makeSQLTableFunctionPlan(
+ name: String,
+ function: SQLFunction,
+ input: Seq[Expression],
+ outputAttrs: Seq[Attribute]): LogicalPlan = {
+ assert(function.isTableFunc)
+ val funcName = function.name.funcName
+
+ // Use captured SQL configs when parsing a SQL function.
+ val conf = new SQLConf()
+ function.getSQLConfigs.foreach { case (k, v) => conf.settings.put(k, v) }
+ SQLConf.withExistingConf(conf) {
+ val inputParam = function.inputParam
+ val returnParam = function.getTableFuncReturnCols
+ val (_, query) = function.getExpressionAndQuery(parser, isTableFunc = true)
+ assert(query.isDefined)
+
+ // Check function arguments
+ val paramSize = inputParam.map(_.size).getOrElse(0)
+ if (input.size > paramSize) {
+ throw QueryCompilationErrors.wrongNumArgsError(
+ name, paramSize.toString, input.size)
+ }
+
+ val body = if (inputParam.isDefined) {
+ val param = inputParam.get
+ // Attributes referencing the input parameters inside the function can use the
+ // function name as a qualifier.
+ val qualifier = Seq(funcName)
+ val paddedInput = input ++
+ param.takeRight(paramSize - input.size).map { p =>
+ val defaultExpr = p.getDefault()
+ if (defaultExpr.isDefined) {
+ Cast(parseDefault(defaultExpr.get, parser), p.dataType)
+ } else {
+ throw QueryCompilationErrors.wrongNumArgsError(
+ name, paramSize.toString, input.size)
+ }
+ }
+
+ val inputCast = paddedInput.zip(param.fields).map {
+ case (expr, param) =>
+ // Add outer references to all attributes in the function input.
+ val outer = expr.transform {
+ case a: Attribute => OuterReference(a)
+ }
+ Alias(Cast(outer, param.dataType), param.name)(qualifier = qualifier)
Review Comment:
We only need it for the actual inputs, not the default values.
--
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-50763][SQL] Add Analyzer rule for resolving SQL table functions [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #49471:
URL: https://github.com/apache/spark/pull/49471#discussion_r1984459708
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala:
##########
@@ -1675,6 +1676,86 @@ class SessionCatalog(
}
}
+ /**
+ * Constructs a SQL table function plan.
+ * This function should be invoked with the captured SQL configs from the function.
+ *
+ * Example SQL table function:
+ *
+ * CREATE FUNCTION foo(x INT) RETURNS TABLE(a INT) RETURN SELECT x + 1 AS x1
+ *
+ * Query:
+ *
+ * SELECT * FROM foo(1);
+ *
+ * Plan:
+ *
+ * Project [CAST(x1 AS INT) AS a]
+ * +- LateralJoin lateral-subquery [x]
+ * : +- Project [(outer(x) + 1) AS x1]
+ * : +- OneRowRelation
+ * +- Project [CAST(1 AS INT) AS x]
+ * +- OneRowRelation
+ */
+ def makeSQLTableFunctionPlan(
+ name: String,
+ function: SQLFunction,
+ input: Seq[Expression],
+ outputAttrs: Seq[Attribute]): LogicalPlan = {
+ assert(function.isTableFunc)
+ val funcName = function.name.funcName
+ val inputParam = function.inputParam
+ val returnParam = function.getTableFuncReturnCols
+ val (_, query) = function.getExpressionAndQuery(parser, isTableFunc = true)
+ assert(query.isDefined)
+
+ // Check function arguments
+ val paramSize = inputParam.map(_.size).getOrElse(0)
+ if (input.size > paramSize) {
+ throw QueryCompilationErrors.wrongNumArgsError(
+ name, paramSize.toString, input.size)
+ }
+
+ val body = if (inputParam.isDefined) {
+ val param = inputParam.get
+ // Attributes referencing the input parameters inside the function can use the
+ // function name as a qualifier.
+ val qualifier = Seq(funcName)
+ val paddedInput = input ++
+ param.takeRight(paramSize - input.size).map { p =>
Review Comment:
This is the confusing part. If we support by-name parameters, how can this get the missing parameters correctly? Shouldn't we search by name and filter out the parameters that are not referenced by the named arguments?
--
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-50763][SQL] Add Analyzer rule for resolving SQL table functions [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #49471:
URL: https://github.com/apache/spark/pull/49471#discussion_r1982841614
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala:
##########
@@ -2655,6 +2656,93 @@ class Analyzer(override val catalogManager: CatalogManager) extends RuleExecutor
}
}
+ /*
+ * This rule resolves SQL table functions.
+ */
+ object ResolveSQLTableFunctions extends Rule[LogicalPlan] with AliasHelper {
+
+ /**
+ * Check if a subquery plan is subject to the COUNT bug that can cause wrong results.
+ * A lateral correlation can only be removed if the lateral subquery is not subject to
+ * the COUNT bug. Currently only lateral correlation can handle it correctly.
+ */
+ private def hasCountBug(sub: LogicalPlan): Boolean = sub.find {
+ // The COUNT bug occurs when there is an Aggregate that satisfies all the following
+ // conditions:
+ // 1) is on the correlation path
+ // 2) has non-empty group by expressions
+ // 3) has one or more output columns that evaluate to non-null values with empty input.
+ // E.g: COUNT(empty row) = 0.
+ // For simplicity, we use a stricter criteria (1 and 2 only) to determine if a query
+ // is subject to the COUNT bug.
+ case a: Aggregate if a.groupingExpressions.nonEmpty => hasOuterReferences(a.child)
Review Comment:
is it possible that the Aggregate node itself has outer references?
--
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-50763][SQL] Add Analyzer rule for resolving SQL table functions [spark]
Posted by "allisonwang-db (via GitHub)" <gi...@apache.org>.
allisonwang-db commented on code in PR #49471:
URL: https://github.com/apache/spark/pull/49471#discussion_r1984333875
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala:
##########
@@ -2655,6 +2656,93 @@ class Analyzer(override val catalogManager: CatalogManager) extends RuleExecutor
}
}
+ /*
+ * This rule resolves SQL table functions.
+ */
+ object ResolveSQLTableFunctions extends Rule[LogicalPlan] with AliasHelper {
+
+ /**
+ * Check if a subquery plan is subject to the COUNT bug that can cause wrong results.
+ * A lateral correlation can only be removed if the lateral subquery is not subject to
+ * the COUNT bug. Currently only lateral correlation can handle it correctly.
+ */
+ private def hasCountBug(sub: LogicalPlan): Boolean = sub.find {
+ // The COUNT bug occurs when there is an Aggregate that satisfies all the following
+ // conditions:
+ // 1) is on the correlation path
+ // 2) has non-empty group by expressions
+ // 3) has one or more output columns that evaluate to non-null values with empty input.
+ // E.g: COUNT(empty row) = 0.
+ // For simplicity, we use a stricter criteria (1 and 2 only) to determine if a query
+ // is subject to the COUNT bug.
+ case a: Aggregate if a.groupingExpressions.nonEmpty => hasOuterReferences(a.child)
Review Comment:
We don't allow Aggregate node to host any outer references.
--
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-50763][SQL] Add Analyzer rule for resolving SQL table functions [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on PR #49471:
URL: https://github.com/apache/spark/pull/49471#issuecomment-2723198248
thanks, merging to master/4.0! (This is the last piece of the SQL UDF feature)
--
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-50763][SQL] Add Analyzer rule for resolving SQL table functions [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #49471:
URL: https://github.com/apache/spark/pull/49471#discussion_r1984468771
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala:
##########
@@ -1675,6 +1676,86 @@ class SessionCatalog(
}
}
+ /**
+ * Constructs a SQL table function plan.
+ * This function should be invoked with the captured SQL configs from the function.
+ *
+ * Example SQL table function:
+ *
+ * CREATE FUNCTION foo(x INT) RETURNS TABLE(a INT) RETURN SELECT x + 1 AS x1
+ *
+ * Query:
+ *
+ * SELECT * FROM foo(1);
+ *
+ * Plan:
+ *
+ * Project [CAST(x1 AS INT) AS a]
+ * +- LateralJoin lateral-subquery [x]
+ * : +- Project [(outer(x) + 1) AS x1]
+ * : +- OneRowRelation
+ * +- Project [CAST(1 AS INT) AS x]
+ * +- OneRowRelation
+ */
+ def makeSQLTableFunctionPlan(
+ name: String,
+ function: SQLFunction,
+ input: Seq[Expression],
+ outputAttrs: Seq[Attribute]): LogicalPlan = {
+ assert(function.isTableFunc)
+ val funcName = function.name.funcName
+ val inputParam = function.inputParam
+ val returnParam = function.getTableFuncReturnCols
+ val (_, query) = function.getExpressionAndQuery(parser, isTableFunc = true)
+ assert(query.isDefined)
+
+ // Check function arguments
+ val paramSize = inputParam.map(_.size).getOrElse(0)
+ if (input.size > paramSize) {
+ throw QueryCompilationErrors.wrongNumArgsError(
+ name, paramSize.toString, input.size)
+ }
+
+ val body = if (inputParam.isDefined) {
+ val param = inputParam.get
+ // Attributes referencing the input parameters inside the function can use the
+ // function name as a qualifier.
+ val qualifier = Seq(funcName)
+ val paddedInput = input ++
+ param.takeRight(paramSize - input.size).map { p =>
Review Comment:
This looks like dead code, `SQLTableFunction` is created by `SessionCatalog#makeSQLTableFunctionBuilder`, which already handles by-name parameters and fills default values. If there are missing parameters that has no default value, it should already failed: https://github.com/apache/spark/blob/master/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/FunctionBuilderBase.scala#L192-L193
--
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-50763][SQL] Add Analyzer rule for resolving SQL table functions [spark]
Posted by "allisonwang-db (via GitHub)" <gi...@apache.org>.
allisonwang-db commented on code in PR #49471:
URL: https://github.com/apache/spark/pull/49471#discussion_r1985591167
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala:
##########
@@ -1675,6 +1676,86 @@ class SessionCatalog(
}
}
+ /**
+ * Constructs a SQL table function plan.
+ * This function should be invoked with the captured SQL configs from the function.
+ *
+ * Example SQL table function:
+ *
+ * CREATE FUNCTION foo(x INT) RETURNS TABLE(a INT) RETURN SELECT x + 1 AS x1
+ *
+ * Query:
+ *
+ * SELECT * FROM foo(1);
+ *
+ * Plan:
+ *
+ * Project [CAST(x1 AS INT) AS a]
+ * +- LateralJoin lateral-subquery [x]
+ * : +- Project [(outer(x) + 1) AS x1]
+ * : +- OneRowRelation
+ * +- Project [CAST(1 AS INT) AS x]
+ * +- OneRowRelation
+ */
+ def makeSQLTableFunctionPlan(
+ name: String,
+ function: SQLFunction,
+ input: Seq[Expression],
+ outputAttrs: Seq[Attribute]): LogicalPlan = {
+ assert(function.isTableFunc)
+ val funcName = function.name.funcName
+ val inputParam = function.inputParam
+ val returnParam = function.getTableFuncReturnCols
+ val (_, query) = function.getExpressionAndQuery(parser, isTableFunc = true)
+ assert(query.isDefined)
+
+ // Check function arguments
+ val paramSize = inputParam.map(_.size).getOrElse(0)
+ if (input.size > paramSize) {
+ throw QueryCompilationErrors.wrongNumArgsError(
+ name, paramSize.toString, input.size)
+ }
+
+ val body = if (inputParam.isDefined) {
+ val param = inputParam.get
+ // Attributes referencing the input parameters inside the function can use the
+ // function name as a qualifier.
+ val qualifier = Seq(funcName)
+ val paddedInput = input ++
+ param.takeRight(paramSize - input.size).map { p =>
Review Comment:
Actually I realized the default argument is not really properly supported (for both scalar and table functions):
```
scala> spark.sql("""create function foo(x int default 0) returns int return x""")
org.apache.spark.sql.catalyst.parser.ParseException:
[PARSE_SYNTAX_ERROR] Syntax error at or near 'default'. SQLSTATE: 42601
```
I will create a separate PR to make this work. [SPARK-51439](https://issues.apache.org/jira/browse/SPARK-51439)
--
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-50763][SQL] Add Analyzer rule for resolving SQL table functions [spark]
Posted by "allisonwang-db (via GitHub)" <gi...@apache.org>.
allisonwang-db commented on code in PR #49471:
URL: https://github.com/apache/spark/pull/49471#discussion_r1985591167
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala:
##########
@@ -1675,6 +1676,86 @@ class SessionCatalog(
}
}
+ /**
+ * Constructs a SQL table function plan.
+ * This function should be invoked with the captured SQL configs from the function.
+ *
+ * Example SQL table function:
+ *
+ * CREATE FUNCTION foo(x INT) RETURNS TABLE(a INT) RETURN SELECT x + 1 AS x1
+ *
+ * Query:
+ *
+ * SELECT * FROM foo(1);
+ *
+ * Plan:
+ *
+ * Project [CAST(x1 AS INT) AS a]
+ * +- LateralJoin lateral-subquery [x]
+ * : +- Project [(outer(x) + 1) AS x1]
+ * : +- OneRowRelation
+ * +- Project [CAST(1 AS INT) AS x]
+ * +- OneRowRelation
+ */
+ def makeSQLTableFunctionPlan(
+ name: String,
+ function: SQLFunction,
+ input: Seq[Expression],
+ outputAttrs: Seq[Attribute]): LogicalPlan = {
+ assert(function.isTableFunc)
+ val funcName = function.name.funcName
+ val inputParam = function.inputParam
+ val returnParam = function.getTableFuncReturnCols
+ val (_, query) = function.getExpressionAndQuery(parser, isTableFunc = true)
+ assert(query.isDefined)
+
+ // Check function arguments
+ val paramSize = inputParam.map(_.size).getOrElse(0)
+ if (input.size > paramSize) {
+ throw QueryCompilationErrors.wrongNumArgsError(
+ name, paramSize.toString, input.size)
+ }
+
+ val body = if (inputParam.isDefined) {
+ val param = inputParam.get
+ // Attributes referencing the input parameters inside the function can use the
+ // function name as a qualifier.
+ val qualifier = Seq(funcName)
+ val paddedInput = input ++
+ param.takeRight(paramSize - input.size).map { p =>
Review Comment:
Actually I realized the default argument is not really properly supported (for both scalar and table functions):
```
scala> spark.sql("""create function foo(x int default 0) returns int return x""")
org.apache.spark.sql.catalyst.parser.ParseException:
[PARSE_SYNTAX_ERROR] Syntax error at or near 'default'. SQLSTATE: 42601
```
I will create a separate PR to make this work.
--
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-50763][SQL] Add Analyzer rule for resolving SQL table functions [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #49471:
URL: https://github.com/apache/spark/pull/49471#discussion_r1928176325
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala:
##########
@@ -1675,6 +1676,91 @@ class SessionCatalog(
}
}
+ /**
+ * Constructs a table SQL function plan.
+ *
+ * Example SQL table function:
+ *
+ * CREATE FUNCTION foo(x INT) RETURNS TABLE(a INT) RETURN SELECT x + 1 AS x1
+ *
+ * Query:
+ *
+ * SELECT * FROM foo(1);
+ *
+ * Plan:
+ *
+ * Project [CAST(x1 AS INT) AS a]
+ * +- LateralJoin lateral-subquery [x]
+ * : +- Project [(outer(x) + 1) AS x1]
+ * : +- OneRowRelation
+ * +- Project [CAST(1 AS INT) AS x]
+ * +- OneRowRelation
+ */
+ def makeSQLTableFunctionPlan(
+ name: String,
+ function: SQLFunction,
+ input: Seq[Expression],
+ outputAttrs: Seq[Attribute]): LogicalPlan = {
+ assert(function.isTableFunc)
+ val funcName = function.name.funcName
+
+ // Use captured SQL configs when parsing a SQL function.
+ val conf = new SQLConf()
+ function.getSQLConfigs.foreach { case (k, v) => conf.settings.put(k, v) }
+ SQLConf.withExistingConf(conf) {
+ val inputParam = function.inputParam
+ val returnParam = function.getTableFuncReturnCols
+ val (_, query) = function.getExpressionAndQuery(parser, isTableFunc = true)
+ assert(query.isDefined)
+
+ // Check function arguments
+ val paramSize = inputParam.map(_.size).getOrElse(0)
+ if (input.size > paramSize) {
+ throw QueryCompilationErrors.wrongNumArgsError(
+ name, paramSize.toString, input.size)
+ }
+
+ val body = if (inputParam.isDefined) {
+ val param = inputParam.get
+ // Attributes referencing the input parameters inside the function can use the
+ // function name as a qualifier.
+ val qualifier = Seq(funcName)
+ val paddedInput = input ++
+ param.takeRight(paramSize - input.size).map { p =>
Review Comment:
do we support by name parameters?
--
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-50763][SQL] Add Analyzer rule for resolving SQL table functions [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #49471:
URL: https://github.com/apache/spark/pull/49471#discussion_r1982858588
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala:
##########
@@ -1675,6 +1676,91 @@ class SessionCatalog(
}
}
+ /**
+ * Constructs a SQL table function plan.
+ *
+ * Example SQL table function:
+ *
+ * CREATE FUNCTION foo(x INT) RETURNS TABLE(a INT) RETURN SELECT x + 1 AS x1
+ *
+ * Query:
+ *
+ * SELECT * FROM foo(1);
+ *
+ * Plan:
+ *
+ * Project [CAST(x1 AS INT) AS a]
+ * +- LateralJoin lateral-subquery [x]
+ * : +- Project [(outer(x) + 1) AS x1]
+ * : +- OneRowRelation
+ * +- Project [CAST(1 AS INT) AS x]
+ * +- OneRowRelation
+ */
+ def makeSQLTableFunctionPlan(
+ name: String,
+ function: SQLFunction,
+ input: Seq[Expression],
+ outputAttrs: Seq[Attribute]): LogicalPlan = {
+ assert(function.isTableFunc)
+ val funcName = function.name.funcName
+
+ // Use captured SQL configs when parsing a SQL function.
+ val conf = new SQLConf()
+ function.getSQLConfigs.foreach { case (k, v) => conf.settings.put(k, v) }
+ SQLConf.withExistingConf(conf) {
+ val inputParam = function.inputParam
+ val returnParam = function.getTableFuncReturnCols
+ val (_, query) = function.getExpressionAndQuery(parser, isTableFunc = true)
+ assert(query.isDefined)
+
+ // Check function arguments
+ val paramSize = inputParam.map(_.size).getOrElse(0)
+ if (input.size > paramSize) {
+ throw QueryCompilationErrors.wrongNumArgsError(
+ name, paramSize.toString, input.size)
+ }
+
+ val body = if (inputParam.isDefined) {
+ val param = inputParam.get
+ // Attributes referencing the input parameters inside the function can use the
+ // function name as a qualifier.
+ val qualifier = Seq(funcName)
+ val paddedInput = input ++
+ param.takeRight(paramSize - input.size).map { p =>
+ val defaultExpr = p.getDefault()
+ if (defaultExpr.isDefined) {
+ Cast(parseDefault(defaultExpr.get, parser), p.dataType)
+ } else {
+ throw QueryCompilationErrors.wrongNumArgsError(
+ name, paramSize.toString, input.size)
+ }
+ }
+
+ val inputCast = paddedInput.zip(param.fields).map {
+ case (expr, param) =>
+ // Add outer references to all attributes in the function input.
+ val outer = expr.transform {
+ case a: Attribute => OuterReference(a)
+ }
+ Alias(Cast(outer, param.dataType), param.name)(qualifier = qualifier)
Review Comment:
Is this Cast duplicated? It's already added in https://github.com/apache/spark/pull/49471/files#diff-9dd0899e5406230aeff96654432da54f35255f6dc60eecb87264a5c508a8c826R1732
--
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-50763][SQL] Add Analyzer rule for resolving SQL table functions [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #49471:
URL: https://github.com/apache/spark/pull/49471#discussion_r1982859590
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala:
##########
@@ -1675,6 +1676,91 @@ class SessionCatalog(
}
}
+ /**
+ * Constructs a table SQL function plan.
+ *
+ * Example SQL table function:
+ *
+ * CREATE FUNCTION foo(x INT) RETURNS TABLE(a INT) RETURN SELECT x + 1 AS x1
+ *
+ * Query:
+ *
+ * SELECT * FROM foo(1);
+ *
+ * Plan:
+ *
+ * Project [CAST(x1 AS INT) AS a]
+ * +- LateralJoin lateral-subquery [x]
+ * : +- Project [(outer(x) + 1) AS x1]
+ * : +- OneRowRelation
+ * +- Project [CAST(1 AS INT) AS x]
+ * +- OneRowRelation
+ */
+ def makeSQLTableFunctionPlan(
+ name: String,
+ function: SQLFunction,
+ input: Seq[Expression],
+ outputAttrs: Seq[Attribute]): LogicalPlan = {
+ assert(function.isTableFunc)
+ val funcName = function.name.funcName
+
+ // Use captured SQL configs when parsing a SQL function.
+ val conf = new SQLConf()
+ function.getSQLConfigs.foreach { case (k, v) => conf.settings.put(k, v) }
+ SQLConf.withExistingConf(conf) {
+ val inputParam = function.inputParam
+ val returnParam = function.getTableFuncReturnCols
+ val (_, query) = function.getExpressionAndQuery(parser, isTableFunc = true)
+ assert(query.isDefined)
+
+ // Check function arguments
+ val paramSize = inputParam.map(_.size).getOrElse(0)
+ if (input.size > paramSize) {
+ throw QueryCompilationErrors.wrongNumArgsError(
+ name, paramSize.toString, input.size)
+ }
+
+ val body = if (inputParam.isDefined) {
+ val param = inputParam.get
+ // Attributes referencing the input parameters inside the function can use the
+ // function name as a qualifier.
+ val qualifier = Seq(funcName)
+ val paddedInput = input ++
+ param.takeRight(paramSize - input.size).map { p =>
Review Comment:
ping @allisonwang-db , if we don't support, do we have a clear error message if users invoke SQL TVF with named parameters?
--
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-50763][SQL] Add Analyzer rule for resolving SQL table functions [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #49471:
URL: https://github.com/apache/spark/pull/49471#discussion_r1928172054
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala:
##########
@@ -2636,6 +2637,93 @@ class Analyzer(override val catalogManager: CatalogManager) extends RuleExecutor
}
}
+ /*
+ * This rule resolves SQL table functions.
+ */
+ object ResolveSQLTableFunctions extends Rule[LogicalPlan] with AliasHelper {
+
+ /**
+ * Check if a subquery plan is subject to the COUNT bug that can cause wrong results.
+ * A lateral correlation can only be removed if the lateral subquery is not subject to
+ * the COUNT bug. Currently only lateral correlation can handle it correctly.
+ */
+ private def hasCountBug(sub: LogicalPlan): Boolean = sub.find {
+ // The COUNT bug occurs when there is an Aggregate that satisfies all the following
+ // conditions:
+ // 1) is on the correlation path
+ // 2) has non-empty group by expressions
+ // 3) has one or more output columns that evaluate to non-null values with empty input.
+ // E.g: COUNT(empty row) = 0.
+ // For simplicity, we use a stricter criteria (1 and 2 only) to determine if a query
+ // is subject to the COUNT bug.
+ case a: Aggregate if a.groupingExpressions.nonEmpty => hasOuterReferences(a.child)
+ case _ => false
+ }.nonEmpty
+
+ /**
+ * Rewrite a resolved SQL table function plan by removing unnecessary lateral joins:
+ * Before:
+ * LateralJoin lateral-subquery [a], Inner
+ * : +- Project [c1, c2]
+ * : +- Filter [outer(a) == c1]
+ * : +- Relation [c1, c2]
+ * +- Project [1 AS a]
+ * +- OneRowRelation
+ * After:
+ * Project [c1, c2]
+ * +- Filter [1 == c1] <---- Replaced outer(a)
+ * +- Relation [c1, c2]
+ */
+ private def rewrite(plan: LogicalPlan): LogicalPlan = {
+ (plan transformUp {
+ case j @ LateralJoin(Project(aliases, _: OneRowRelation), sub: LateralSubquery, Inner, None)
+ if j.resolved && aliases.forall(_.deterministic) =>
+ val attrMap = AttributeMap(aliases.collect { case a: Alias => a.toAttribute -> a.child })
+ val newPlan = sub.plan.transformAllExpressionsWithPruning(
+ _.containsPattern(OUTER_REFERENCE)) {
+ // Avoid replacing outer references that do not belong to the current outer plan.
+ // This can happen if the child of an alias also contains outer references (nested
+ // table function references). E.g:
+ // LateralJoin
+ // : +- Filter [outer(a) == x]
+ // : +- Relation [x, y]
+ // +- Project [outer(c) AS a]
+ // +- OneRowRelation
+ case OuterReference(a: Attribute) if attrMap.contains(a) => attrMap(a) match {
+ case ne: NamedExpression => ne
+ case o => Alias(o, a.name)(exprId = a.exprId, qualifier = a.qualifier)
Review Comment:
nvm, this is the inner alias, and we don't know if we can omit it or not 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-50763][SQL] Add Analyzer rule for resolving SQL table functions [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #49471:
URL: https://github.com/apache/spark/pull/49471#discussion_r1928173142
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala:
##########
@@ -1675,6 +1676,91 @@ class SessionCatalog(
}
}
+ /**
+ * Constructs a table SQL function plan.
Review Comment:
```suggestion
* Constructs a SQL table function plan.
```
--
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-50763][SQL] Add Analyzer rule for resolving SQL table functions [spark]
Posted by "allisonwang-db (via GitHub)" <gi...@apache.org>.
allisonwang-db commented on code in PR #49471:
URL: https://github.com/apache/spark/pull/49471#discussion_r1984271562
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala:
##########
@@ -1675,6 +1676,91 @@ class SessionCatalog(
}
}
+ /**
+ * Constructs a table SQL function plan.
+ *
+ * Example SQL table function:
+ *
+ * CREATE FUNCTION foo(x INT) RETURNS TABLE(a INT) RETURN SELECT x + 1 AS x1
+ *
+ * Query:
+ *
+ * SELECT * FROM foo(1);
+ *
+ * Plan:
+ *
+ * Project [CAST(x1 AS INT) AS a]
+ * +- LateralJoin lateral-subquery [x]
+ * : +- Project [(outer(x) + 1) AS x1]
+ * : +- OneRowRelation
+ * +- Project [CAST(1 AS INT) AS x]
+ * +- OneRowRelation
+ */
+ def makeSQLTableFunctionPlan(
+ name: String,
+ function: SQLFunction,
+ input: Seq[Expression],
+ outputAttrs: Seq[Attribute]): LogicalPlan = {
+ assert(function.isTableFunc)
+ val funcName = function.name.funcName
+
+ // Use captured SQL configs when parsing a SQL function.
+ val conf = new SQLConf()
+ function.getSQLConfigs.foreach { case (k, v) => conf.settings.put(k, v) }
+ SQLConf.withExistingConf(conf) {
+ val inputParam = function.inputParam
+ val returnParam = function.getTableFuncReturnCols
+ val (_, query) = function.getExpressionAndQuery(parser, isTableFunc = true)
+ assert(query.isDefined)
+
+ // Check function arguments
+ val paramSize = inputParam.map(_.size).getOrElse(0)
+ if (input.size > paramSize) {
+ throw QueryCompilationErrors.wrongNumArgsError(
+ name, paramSize.toString, input.size)
+ }
+
+ val body = if (inputParam.isDefined) {
+ val param = inputParam.get
+ // Attributes referencing the input parameters inside the function can use the
+ // function name as a qualifier.
+ val qualifier = Seq(funcName)
+ val paddedInput = input ++
+ param.takeRight(paramSize - input.size).map { p =>
Review Comment:
Yea we support it. It's in `makeSQLTableFunctionBuilder` (rearrangeArguments)
--
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-50763][SQL] Add Analyzer rule for resolving SQL table functions [spark]
Posted by "allisonwang-db (via GitHub)" <gi...@apache.org>.
allisonwang-db commented on code in PR #49471:
URL: https://github.com/apache/spark/pull/49471#discussion_r1984402797
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala:
##########
@@ -2655,6 +2656,93 @@ class Analyzer(override val catalogManager: CatalogManager) extends RuleExecutor
}
}
+ /*
+ * This rule resolves SQL table functions.
+ */
+ object ResolveSQLTableFunctions extends Rule[LogicalPlan] with AliasHelper {
+
+ /**
+ * Check if a subquery plan is subject to the COUNT bug that can cause wrong results.
+ * A lateral correlation can only be removed if the lateral subquery is not subject to
+ * the COUNT bug. Currently only lateral correlation can handle it correctly.
+ */
+ private def hasCountBug(sub: LogicalPlan): Boolean = sub.find {
+ // The COUNT bug occurs when there is an Aggregate that satisfies all the following
+ // conditions:
+ // 1) is on the correlation path
+ // 2) has non-empty group by expressions
+ // 3) has one or more output columns that evaluate to non-null values with empty input.
+ // E.g: COUNT(empty row) = 0.
+ // For simplicity, we use a stricter criteria (1 and 2 only) to determine if a query
+ // is subject to the COUNT bug.
+ case a: Aggregate if a.groupingExpressions.nonEmpty => hasOuterReferences(a.child)
Review Comment:
https://github.com/apache/spark/blob/1c14b06f8224d0f5db072bdeabb1f36d7359b05a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ValidateSubqueryExpression.scala#L434-L439
--
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-50763][SQL] Add Analyzer rule for resolving SQL table functions [spark]
Posted by "allisonwang-db (via GitHub)" <gi...@apache.org>.
allisonwang-db commented on PR #49471:
URL: https://github.com/apache/spark/pull/49471#issuecomment-2611271597
cc @cloud-fan @srielau
--
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-50763][SQL] Add Analyzer rule for resolving SQL table functions [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #49471:
URL: https://github.com/apache/spark/pull/49471#discussion_r1928174592
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala:
##########
@@ -2636,6 +2637,93 @@ class Analyzer(override val catalogManager: CatalogManager) extends RuleExecutor
}
}
+ /*
+ * This rule resolves SQL table functions.
+ */
+ object ResolveSQLTableFunctions extends Rule[LogicalPlan] with AliasHelper {
+
+ /**
+ * Check if a subquery plan is subject to the COUNT bug that can cause wrong results.
+ * A lateral correlation can only be removed if the lateral subquery is not subject to
+ * the COUNT bug. Currently only lateral correlation can handle it correctly.
+ */
+ private def hasCountBug(sub: LogicalPlan): Boolean = sub.find {
+ // The COUNT bug occurs when there is an Aggregate that satisfies all the following
+ // conditions:
+ // 1) is on the correlation path
+ // 2) has non-empty group by expressions
+ // 3) has one or more output columns that evaluate to non-null values with empty input.
+ // E.g: COUNT(empty row) = 0.
+ // For simplicity, we use a stricter criteria (1 and 2 only) to determine if a query
+ // is subject to the COUNT bug.
+ case a: Aggregate if a.groupingExpressions.nonEmpty => hasOuterReferences(a.child)
+ case _ => false
+ }.nonEmpty
+
+ /**
+ * Rewrite a resolved SQL table function plan by removing unnecessary lateral joins:
+ * Before:
+ * LateralJoin lateral-subquery [a], Inner
+ * : +- Project [c1, c2]
+ * : +- Filter [outer(a) == c1]
+ * : +- Relation [c1, c2]
+ * +- Project [1 AS a]
+ * +- OneRowRelation
+ * After:
+ * Project [c1, c2]
+ * +- Filter [1 == c1] <---- Replaced outer(a)
+ * +- Relation [c1, c2]
+ */
+ private def rewrite(plan: LogicalPlan): LogicalPlan = {
+ (plan transformUp {
+ case j @ LateralJoin(Project(aliases, _: OneRowRelation), sub: LateralSubquery, Inner, None)
+ if j.resolved && aliases.forall(_.deterministic) =>
+ val attrMap = AttributeMap(aliases.collect { case a: Alias => a.toAttribute -> a.child })
+ val newPlan = sub.plan.transformAllExpressionsWithPruning(
+ _.containsPattern(OUTER_REFERENCE)) {
+ // Avoid replacing outer references that do not belong to the current outer plan.
+ // This can happen if the child of an alias also contains outer references (nested
+ // table function references). E.g:
+ // LateralJoin
+ // : +- Filter [outer(a) == x]
+ // : +- Relation [x, y]
+ // +- Project [outer(c) AS a]
+ // +- OneRowRelation
+ case OuterReference(a: Attribute) if attrMap.contains(a) => attrMap(a) match {
+ case ne: NamedExpression => ne
+ case o => Alias(o, a.name)(exprId = a.exprId, qualifier = a.qualifier)
+ }
+ }
+ // Keep the original lateral join if the new plan is subject to the count bug.
+ if (hasCountBug(newPlan)) j else newPlan
+ }).transformWithPruning(_.containsPattern(ALIAS)) {
+ // As a result of the above rewriting, we may end-up introducing nested Aliases (i.e.,
+ // Aliases defined inside Aliases). This is problematic for the plan canonicalization as
+ // it doesn't expect nested Aliases. Therefore, here we remove non-top level Aliases.
+ case node => node.mapExpressions(trimNonTopLevelAliases)
+ }
+ }
+
+ def apply(plan: LogicalPlan): LogicalPlan = plan.resolveOperatorsUpWithPruning(
+ _.containsPattern(SQL_TABLE_FUNCTION)) {
+ case SQLTableFunction(name, function, inputs, output) =>
+ val plan = v1SessionCatalog.makeSQLTableFunctionPlan(name, function, inputs, output)
+ // Resolve the SQL table function plan using its function context.
+ val conf = new SQLConf()
+ function.getSQLConfigs.foreach { case (k, v) => conf.settings.put(k, v) }
+ val resolved = SQLConf.withExistingConf(conf) {
Review Comment:
nit: shall we put `v1SessionCatalog.makeSQLTableFunctionPlan` inside here as well? Then we don't need to construct the SQLConf in `makeSQLTableFunctionPlan`
--
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-50763][SQL] Add Analyzer rule for resolving SQL table functions [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #49471:
URL: https://github.com/apache/spark/pull/49471#discussion_r1982850033
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala:
##########
@@ -2636,6 +2637,93 @@ class Analyzer(override val catalogManager: CatalogManager) extends RuleExecutor
}
}
+ /*
+ * This rule resolves SQL table functions.
+ */
+ object ResolveSQLTableFunctions extends Rule[LogicalPlan] with AliasHelper {
+
+ /**
+ * Check if a subquery plan is subject to the COUNT bug that can cause wrong results.
+ * A lateral correlation can only be removed if the lateral subquery is not subject to
+ * the COUNT bug. Currently only lateral correlation can handle it correctly.
+ */
+ private def hasCountBug(sub: LogicalPlan): Boolean = sub.find {
+ // The COUNT bug occurs when there is an Aggregate that satisfies all the following
+ // conditions:
+ // 1) is on the correlation path
+ // 2) has non-empty group by expressions
+ // 3) has one or more output columns that evaluate to non-null values with empty input.
+ // E.g: COUNT(empty row) = 0.
+ // For simplicity, we use a stricter criteria (1 and 2 only) to determine if a query
+ // is subject to the COUNT bug.
+ case a: Aggregate if a.groupingExpressions.nonEmpty => hasOuterReferences(a.child)
+ case _ => false
+ }.nonEmpty
+
+ /**
+ * Rewrite a resolved SQL table function plan by removing unnecessary lateral joins:
+ * Before:
+ * LateralJoin lateral-subquery [a], Inner
+ * : +- Project [c1, c2]
+ * : +- Filter [outer(a) == c1]
+ * : +- Relation [c1, c2]
+ * +- Project [1 AS a]
+ * +- OneRowRelation
+ * After:
+ * Project [c1, c2]
+ * +- Filter [1 == c1] <---- Replaced outer(a)
+ * +- Relation [c1, c2]
+ */
+ private def rewrite(plan: LogicalPlan): LogicalPlan = {
+ (plan transformUp {
+ case j @ LateralJoin(Project(aliases, _: OneRowRelation), sub: LateralSubquery, Inner, None)
+ if j.resolved && aliases.forall(_.deterministic) =>
+ val attrMap = AttributeMap(aliases.collect { case a: Alias => a.toAttribute -> a.child })
+ val newPlan = sub.plan.transformAllExpressionsWithPruning(
+ _.containsPattern(OUTER_REFERENCE)) {
+ // Avoid replacing outer references that do not belong to the current outer plan.
+ // This can happen if the child of an alias also contains outer references (nested
+ // table function references). E.g:
+ // LateralJoin
+ // : +- Filter [outer(a) == x]
+ // : +- Relation [x, y]
+ // +- Project [outer(c) AS a]
+ // +- OneRowRelation
+ case OuterReference(a: Attribute) if attrMap.contains(a) => attrMap(a) match {
+ case ne: NamedExpression => ne
+ case o => Alias(o, a.name)(exprId = a.exprId, qualifier = a.qualifier)
+ }
+ }
+ // Keep the original lateral join if the new plan is subject to the count bug.
+ if (hasCountBug(newPlan)) j else newPlan
+ }).transformWithPruning(_.containsPattern(ALIAS)) {
+ // As a result of the above rewriting, we may end-up introducing nested Aliases (i.e.,
+ // Aliases defined inside Aliases). This is problematic for the plan canonicalization as
+ // it doesn't expect nested Aliases. Therefore, here we remove non-top level Aliases.
+ case node => node.mapExpressions(trimNonTopLevelAliases)
+ }
+ }
+
+ def apply(plan: LogicalPlan): LogicalPlan = plan.resolveOperatorsUpWithPruning(
+ _.containsPattern(SQL_TABLE_FUNCTION)) {
+ case SQLTableFunction(name, function, inputs, output) =>
+ val plan = v1SessionCatalog.makeSQLTableFunctionPlan(name, function, inputs, output)
+ // Resolve the SQL table function plan using its function context.
+ val conf = new SQLConf()
+ function.getSQLConfigs.foreach { case (k, v) => conf.settings.put(k, v) }
+ val resolved = SQLConf.withExistingConf(conf) {
Review Comment:
ping @allisonwang-db
--
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-50763][SQL] Add Analyzer rule for resolving SQL table functions [spark]
Posted by "allisonwang-db (via GitHub)" <gi...@apache.org>.
allisonwang-db commented on code in PR #49471:
URL: https://github.com/apache/spark/pull/49471#discussion_r1984269314
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala:
##########
@@ -2636,6 +2637,93 @@ class Analyzer(override val catalogManager: CatalogManager) extends RuleExecutor
}
}
+ /*
+ * This rule resolves SQL table functions.
+ */
+ object ResolveSQLTableFunctions extends Rule[LogicalPlan] with AliasHelper {
+
+ /**
+ * Check if a subquery plan is subject to the COUNT bug that can cause wrong results.
+ * A lateral correlation can only be removed if the lateral subquery is not subject to
+ * the COUNT bug. Currently only lateral correlation can handle it correctly.
+ */
+ private def hasCountBug(sub: LogicalPlan): Boolean = sub.find {
+ // The COUNT bug occurs when there is an Aggregate that satisfies all the following
+ // conditions:
+ // 1) is on the correlation path
+ // 2) has non-empty group by expressions
+ // 3) has one or more output columns that evaluate to non-null values with empty input.
+ // E.g: COUNT(empty row) = 0.
+ // For simplicity, we use a stricter criteria (1 and 2 only) to determine if a query
+ // is subject to the COUNT bug.
+ case a: Aggregate if a.groupingExpressions.nonEmpty => hasOuterReferences(a.child)
+ case _ => false
+ }.nonEmpty
+
+ /**
+ * Rewrite a resolved SQL table function plan by removing unnecessary lateral joins:
+ * Before:
+ * LateralJoin lateral-subquery [a], Inner
+ * : +- Project [c1, c2]
+ * : +- Filter [outer(a) == c1]
+ * : +- Relation [c1, c2]
+ * +- Project [1 AS a]
+ * +- OneRowRelation
+ * After:
+ * Project [c1, c2]
+ * +- Filter [1 == c1] <---- Replaced outer(a)
+ * +- Relation [c1, c2]
+ */
+ private def rewrite(plan: LogicalPlan): LogicalPlan = {
+ (plan transformUp {
+ case j @ LateralJoin(Project(aliases, _: OneRowRelation), sub: LateralSubquery, Inner, None)
+ if j.resolved && aliases.forall(_.deterministic) =>
+ val attrMap = AttributeMap(aliases.collect { case a: Alias => a.toAttribute -> a.child })
+ val newPlan = sub.plan.transformAllExpressionsWithPruning(
+ _.containsPattern(OUTER_REFERENCE)) {
+ // Avoid replacing outer references that do not belong to the current outer plan.
+ // This can happen if the child of an alias also contains outer references (nested
+ // table function references). E.g:
+ // LateralJoin
+ // : +- Filter [outer(a) == x]
+ // : +- Relation [x, y]
+ // +- Project [outer(c) AS a]
+ // +- OneRowRelation
+ case OuterReference(a: Attribute) if attrMap.contains(a) => attrMap(a) match {
+ case ne: NamedExpression => ne
+ case o => Alias(o, a.name)(exprId = a.exprId, qualifier = a.qualifier)
+ }
+ }
+ // Keep the original lateral join if the new plan is subject to the count bug.
+ if (hasCountBug(newPlan)) j else newPlan
+ }).transformWithPruning(_.containsPattern(ALIAS)) {
+ // As a result of the above rewriting, we may end-up introducing nested Aliases (i.e.,
+ // Aliases defined inside Aliases). This is problematic for the plan canonicalization as
+ // it doesn't expect nested Aliases. Therefore, here we remove non-top level Aliases.
+ case node => node.mapExpressions(trimNonTopLevelAliases)
+ }
+ }
+
+ def apply(plan: LogicalPlan): LogicalPlan = plan.resolveOperatorsUpWithPruning(
+ _.containsPattern(SQL_TABLE_FUNCTION)) {
+ case SQLTableFunction(name, function, inputs, output) =>
+ val plan = v1SessionCatalog.makeSQLTableFunctionPlan(name, function, inputs, output)
+ // Resolve the SQL table function plan using its function context.
+ val conf = new SQLConf()
+ function.getSQLConfigs.foreach { case (k, v) => conf.settings.put(k, v) }
+ val resolved = SQLConf.withExistingConf(conf) {
Review Comment:
Good point!
--
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