You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "Pajaraja (via GitHub)" <gi...@apache.org> on 2025/03/26 11:28:32 UTC
[PR] [SPARK-51609][SQL] Optimize Recursive CTE execution for simple queries [spark]
Pajaraja opened a new pull request, #50402:
URL: https://github.com/apache/spark/pull/50402
### What changes were proposed in this pull request?
Optimize Recursive CTEs where the recursion is simple, and everything can be done in-memory. Specifically for the cases where the anchor is a LocalRelation or OneRowRelation and the recursion can be done in-memory.
### Why are the changes needed?
To optimize Recursive CTEs.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
Add a new Suite for testing Recursive CTEs. In there two tests where we test one case each for both LocalRelation and OneRowRelation. Checked that they're faster now.
### Was this patch authored or co-authored using generative AI tooling?
No.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51609][SQL] Optimize Recursive CTE execution for simple queries [spark]
Posted by "Pajaraja (via GitHub)" <gi...@apache.org>.
Pajaraja commented on code in PR #50402:
URL: https://github.com/apache/spark/pull/50402#discussion_r2063257476
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/UnionLoopExec.scala:
##########
@@ -127,28 +134,67 @@ case class UnionLoopExec(
val numAnchorOutputRows = longMetric("numAnchorOutputRows")
val levelLimit = conf.getConf(SQLConf.CTE_RECURSION_LEVEL_LIMIT)
val rowLimit = conf.getConf(SQLConf.CTE_RECURSION_ROW_LIMIT)
+ val AnchorRowLimitToConverToLocalLimit =
+ conf.getConf(SQLConf.CTE_RECURSION_ANCHOR_ROWS_LIMIT_TO_CONVERT_TO_LOCAL_RELATION)
// currentLimit is initialized from the limit argument, and in each step it is decreased by
// the number of rows generated in that step.
// If limit is not passed down, currentLimit is set to be zero and won't be considered in the
// condition of while loop down (limit.isEmpty will be true).
var currentLimit = limit.getOrElse(-1)
- val unionChildren = mutable.ArrayBuffer.empty[LogicalRDD]
+ val unionChildren = mutable.ArrayBuffer.empty[LogicalPlan]
var (prevDF, prevCount) = executeAndCacheAndCount(anchor, currentLimit)
numAnchorOutputRows += prevCount
var currentLevel = 1
- var currentNumRows = 0
+ var currentNumRows: Long = 0
var limitReached: Boolean = false
val numPartitions = prevDF.queryExecution.toRdd.partitions.length
+
+ // In the case we return a sufficiently small number of rows when executing the anchor,
+ // we convert the result of the anchor into a LocalRelation, so that, if the recursion doesn't
+ // reference any external tables, we are able to calculate everything in the optimizer, using
+ // the ConvertToLocalRelation rule, which significantly improves runtime.
+ if (prevCount <= AnchorRowLimitToConverToLocalLimit &&
+ !prevDF.queryExecution.optimizedPlan.isInstanceOf[LocalRelation]) {
+ val local = LocalRelation.fromExternalRows(anchor.output, prevDF.collect().toIndexedSeq)
Review Comment:
Moved.
--
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-51609][SQL] Optimize Recursive CTE execution for simple queries [spark]
Posted by "Pajaraja (via GitHub)" <gi...@apache.org>.
Pajaraja commented on code in PR #50402:
URL: https://github.com/apache/spark/pull/50402#discussion_r2063300460
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/UnionLoopExec.scala:
##########
@@ -149,6 +156,33 @@ case class UnionLoopExec(
val numPartitions = prevDF.queryExecution.toRdd.partitions.length
// Main loop for obtaining the result of the recursive query.
while (prevCount > 0 && !limitReached) {
+ var prevPlan: LogicalPlan = null
+ // the current plan is created by substituting UnionLoopRef node with the project node of
+ // the previous plan.
+ // This way we support only UNION ALL case. Additional case should be added for UNION case.
+ // One way of supporting UNION case can be seen at SPARK-24497 PR from Peter Toth.
+ val newRecursion = recursion.transform {
+ case r: UnionLoopRef if r.loopId == loopId =>
+ prevDF.queryExecution.optimizedPlan match {
+ case l: LocalRelation =>
+ prevPlan = prevDF.queryExecution.optimizedPlan
+ l.copy(output = r.output)
+ case p @ Project(projectList, _: OneRowRelation) =>
+ prevPlan = prevDF.queryExecution.optimizedPlan
+ val prevPlanToRefMapping = projectList.zip(r.output).map {
+ case (fa: Alias, ta) => Alias(fa.child, ta.name)(ta.exprId)
Review Comment:
Done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51609][SQL] Optimize Recursive CTE execution for simple queries [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #50402:
URL: https://github.com/apache/spark/pull/50402#discussion_r2056359712
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/UnionLoopExec.scala:
##########
@@ -149,6 +156,33 @@ case class UnionLoopExec(
val numPartitions = prevDF.queryExecution.toRdd.partitions.length
// Main loop for obtaining the result of the recursive query.
while (prevCount > 0 && !limitReached) {
+ var prevPlan: LogicalPlan = null
+ // the current plan is created by substituting UnionLoopRef node with the project node of
+ // the previous plan.
+ // This way we support only UNION ALL case. Additional case should be added for UNION case.
+ // One way of supporting UNION case can be seen at SPARK-24497 PR from Peter Toth.
+ val newRecursion = recursion.transform {
+ case r: UnionLoopRef if r.loopId == loopId =>
+ prevDF.queryExecution.optimizedPlan match {
+ case l: LocalRelation =>
+ prevPlan = prevDF.queryExecution.optimizedPlan
+ l.copy(output = r.output)
+ case p @ Project(projectList, _: OneRowRelation) =>
+ prevPlan = prevDF.queryExecution.optimizedPlan
Review Comment:
```suggestion
prevPlan = p
```
--
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-51609][SQL] Optimize Recursive CTE execution for simple queries [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #50402:
URL: https://github.com/apache/spark/pull/50402#discussion_r2062942033
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/UnionLoopExec.scala:
##########
@@ -127,28 +134,67 @@ case class UnionLoopExec(
val numAnchorOutputRows = longMetric("numAnchorOutputRows")
val levelLimit = conf.getConf(SQLConf.CTE_RECURSION_LEVEL_LIMIT)
val rowLimit = conf.getConf(SQLConf.CTE_RECURSION_ROW_LIMIT)
+ val AnchorRowLimitToConverToLocalLimit =
+ conf.getConf(SQLConf.CTE_RECURSION_ANCHOR_ROWS_LIMIT_TO_CONVERT_TO_LOCAL_RELATION)
// currentLimit is initialized from the limit argument, and in each step it is decreased by
// the number of rows generated in that step.
// If limit is not passed down, currentLimit is set to be zero and won't be considered in the
// condition of while loop down (limit.isEmpty will be true).
var currentLimit = limit.getOrElse(-1)
- val unionChildren = mutable.ArrayBuffer.empty[LogicalRDD]
+ val unionChildren = mutable.ArrayBuffer.empty[LogicalPlan]
var (prevDF, prevCount) = executeAndCacheAndCount(anchor, currentLimit)
numAnchorOutputRows += prevCount
var currentLevel = 1
- var currentNumRows = 0
+ var currentNumRows: Long = 0
var limitReached: Boolean = false
val numPartitions = prevDF.queryExecution.toRdd.partitions.length
+
+ // In the case we return a sufficiently small number of rows when executing the anchor,
+ // we convert the result of the anchor into a LocalRelation, so that, if the recursion doesn't
+ // reference any external tables, we are able to calculate everything in the optimizer, using
+ // the ConvertToLocalRelation rule, which significantly improves runtime.
+ if (prevCount <= AnchorRowLimitToConverToLocalLimit &&
+ !prevDF.queryExecution.optimizedPlan.isInstanceOf[LocalRelation]) {
+ val local = LocalRelation.fromExternalRows(anchor.output, prevDF.collect().toIndexedSeq)
Review Comment:
I don't think this is the right place to do this optimization. We should do it in `def executeAndCacheAndCount`, so that we can collect query result directly without doing a `repartition`.
--
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-51609][SQL] Optimize Recursive CTE execution for simple queries [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #50402:
URL: https://github.com/apache/spark/pull/50402#discussion_r2065848650
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/UnionLoopExec.scala:
##########
@@ -107,9 +110,33 @@ case class UnionLoopExec(
plan
}
val df = Dataset.ofRows(session, planWithLimit)
- val materializedDF = df.repartition()
- val count = materializedDF.queryExecution.toRdd.count()
- (materializedDF, count)
+
+ df.queryExecution.optimizedPlan match {
+ case l: LocalRelation =>
+ (df, l.data.length.toLong)
+ case Project(_, _: OneRowRelation) =>
+ if (localRelationLimit != 0) {
+ val local = LocalRelation.fromExternalRows(anchor.output, df.collect().toIndexedSeq)
+ (Dataset.ofRows(session, local), 1.toLong)
+ } else {
+ (df, 1.toLong)
+ }
+ case _ =>
+ val materializedDF = df.repartition()
+ val count = materializedDF.queryExecution.toRdd.count()
+
+ // In the case we return a sufficiently small number of rows when executing any step of the
+ // recursion we convert the result into a LocalRelation, so that, if the recursion doesn't
+ // reference any external tables, we are able to calculate everything in the optimizer,
+ // using the ConvertToLocalRelation rule, which significantly improves runtime.
+ if (count <= localRelationLimit) {
+ val local = LocalRelation.fromExternalRows(anchor.output, df.collect().toIndexedSeq)
+ (Dataset.ofRows(session, local), count)
+ }
+ else {
Review Comment:
```suggestion
} else {
```
--
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-51609][SQL] Optimize Recursive CTE execution for simple queries [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #50402:
URL: https://github.com/apache/spark/pull/50402#discussion_r2056361492
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/UnionLoopExec.scala:
##########
@@ -149,6 +156,33 @@ case class UnionLoopExec(
val numPartitions = prevDF.queryExecution.toRdd.partitions.length
// Main loop for obtaining the result of the recursive query.
while (prevCount > 0 && !limitReached) {
+ var prevPlan: LogicalPlan = null
+ // the current plan is created by substituting UnionLoopRef node with the project node of
+ // the previous plan.
+ // This way we support only UNION ALL case. Additional case should be added for UNION case.
+ // One way of supporting UNION case can be seen at SPARK-24497 PR from Peter Toth.
+ val newRecursion = recursion.transform {
+ case r: UnionLoopRef if r.loopId == loopId =>
+ prevDF.queryExecution.optimizedPlan match {
+ case l: LocalRelation =>
+ prevPlan = prevDF.queryExecution.optimizedPlan
+ l.copy(output = r.output)
+ case p @ Project(projectList, _: OneRowRelation) =>
+ prevPlan = prevDF.queryExecution.optimizedPlan
+ val prevPlanToRefMapping = projectList.zip(r.output).map {
+ case (fa: Alias, ta) => Alias(fa.child, ta.name)(ta.exprId)
Review Comment:
and `withName`
--
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-51609][SQL] Optimize Recursive CTE execution for simple queries [spark]
Posted by "Pajaraja (via GitHub)" <gi...@apache.org>.
Pajaraja commented on code in PR #50402:
URL: https://github.com/apache/spark/pull/50402#discussion_r2053779830
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/UnionLoopExec.scala:
##########
@@ -149,6 +156,36 @@ case class UnionLoopExec(
val numPartitions = prevDF.queryExecution.toRdd.partitions.length
// Main loop for obtaining the result of the recursive query.
while (prevCount > 0 && !limitReached) {
+ var prevPlan: LogicalPlan = null
+ // the current plan is created by substituting UnionLoopRef node with the project node of
+ // the previous plan.
+ // This way we support only UNION ALL case. Additional case should be added for UNION case.
+ // One way of supporting UNION case can be seen at SPARK-24497 PR from Peter Toth.
+ val newRecursion = recursion.transform {
+ case r: UnionLoopRef if r.loopId == loopId =>
+ prevDF.queryExecution.optimizedPlan match {
+ case _: LocalRelation =>
+ prevPlan = prevDF.queryExecution.optimizedPlan
+ val prevPlanToRefMapping = prevPlan.output.zip(r.output).map {
+ case (fa, ta) => Alias(fa, ta.name)(ta.exprId)
+ }
+ Project(prevPlanToRefMapping, prevPlan)
Review Comment:
Changed!
--
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-51609][SQL] Optimize Recursive CTE execution for simple queries [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #50402:
URL: https://github.com/apache/spark/pull/50402#discussion_r2056360680
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/UnionLoopExec.scala:
##########
@@ -149,6 +156,33 @@ case class UnionLoopExec(
val numPartitions = prevDF.queryExecution.toRdd.partitions.length
// Main loop for obtaining the result of the recursive query.
while (prevCount > 0 && !limitReached) {
+ var prevPlan: LogicalPlan = null
+ // the current plan is created by substituting UnionLoopRef node with the project node of
+ // the previous plan.
+ // This way we support only UNION ALL case. Additional case should be added for UNION case.
+ // One way of supporting UNION case can be seen at SPARK-24497 PR from Peter Toth.
+ val newRecursion = recursion.transform {
+ case r: UnionLoopRef if r.loopId == loopId =>
+ prevDF.queryExecution.optimizedPlan match {
+ case l: LocalRelation =>
+ prevPlan = prevDF.queryExecution.optimizedPlan
+ l.copy(output = r.output)
+ case p @ Project(projectList, _: OneRowRelation) =>
+ prevPlan = prevDF.queryExecution.optimizedPlan
+ val prevPlanToRefMapping = projectList.zip(r.output).map {
+ case (fa: Alias, ta) => Alias(fa.child, ta.name)(ta.exprId)
Review Comment:
let's use `Alias#withExprId`
--
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-51609][SQL] Optimize Recursive CTE execution for simple queries [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #50402:
URL: https://github.com/apache/spark/pull/50402#discussion_r2056364081
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/UnionLoopExec.scala:
##########
@@ -157,9 +191,6 @@ case class UnionLoopExec(
cause = null)
}
- // Inherit stats and constraints from the dataset of the previous iteration.
- val prevPlan = LogicalRDD.fromDataset(prevDF.queryExecution.toRdd, prevDF, prevDF.isStreaming)
- .newInstance()
unionChildren += prevPlan
currentNumRows += prevCount.toInt
Review Comment:
not related to this PR, but `.toInt` has the risk of overflow. Shall we make `currentNumRows` a long as well?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/UnionLoopExec.scala:
##########
@@ -157,9 +203,6 @@ case class UnionLoopExec(
cause = null)
}
- // Inherit stats and constraints from the dataset of the previous iteration.
- val prevPlan = LogicalRDD.fromDataset(prevDF.queryExecution.toRdd, prevDF, prevDF.isStreaming)
- .newInstance()
unionChildren += prevPlan
currentNumRows += prevCount.toInt
Review Comment:
```suggestion
currentNumRows += prevCount
```
--
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-51609][SQL] Optimize Recursive CTE execution for simple queries [spark]
Posted by "Pajaraja (via GitHub)" <gi...@apache.org>.
Pajaraja commented on code in PR #50402:
URL: https://github.com/apache/spark/pull/50402#discussion_r2063196189
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/UnionLoopExec.scala:
##########
@@ -127,28 +134,67 @@ case class UnionLoopExec(
val numAnchorOutputRows = longMetric("numAnchorOutputRows")
val levelLimit = conf.getConf(SQLConf.CTE_RECURSION_LEVEL_LIMIT)
val rowLimit = conf.getConf(SQLConf.CTE_RECURSION_ROW_LIMIT)
+ val AnchorRowLimitToConverToLocalLimit =
+ conf.getConf(SQLConf.CTE_RECURSION_ANCHOR_ROWS_LIMIT_TO_CONVERT_TO_LOCAL_RELATION)
// currentLimit is initialized from the limit argument, and in each step it is decreased by
// the number of rows generated in that step.
// If limit is not passed down, currentLimit is set to be zero and won't be considered in the
// condition of while loop down (limit.isEmpty will be true).
var currentLimit = limit.getOrElse(-1)
- val unionChildren = mutable.ArrayBuffer.empty[LogicalRDD]
+ val unionChildren = mutable.ArrayBuffer.empty[LogicalPlan]
var (prevDF, prevCount) = executeAndCacheAndCount(anchor, currentLimit)
numAnchorOutputRows += prevCount
var currentLevel = 1
- var currentNumRows = 0
+ var currentNumRows: Long = 0
var limitReached: Boolean = false
val numPartitions = prevDF.queryExecution.toRdd.partitions.length
+
+ // In the case we return a sufficiently small number of rows when executing the anchor,
+ // we convert the result of the anchor into a LocalRelation, so that, if the recursion doesn't
+ // reference any external tables, we are able to calculate everything in the optimizer, using
+ // the ConvertToLocalRelation rule, which significantly improves runtime.
+ if (prevCount <= AnchorRowLimitToConverToLocalLimit &&
+ !prevDF.queryExecution.optimizedPlan.isInstanceOf[LocalRelation]) {
+ val local = LocalRelation.fromExternalRows(anchor.output, prevDF.collect().toIndexedSeq)
+ prevDF = Dataset.ofRows(session, local)
+ }
// Main loop for obtaining the result of the recursive query.
while (prevCount > 0 && !limitReached) {
+ var prevPlan: LogicalPlan = null
+ // the current plan is created by substituting UnionLoopRef node with the project node of
+ // the previous plan.
+ // This way we support only UNION ALL case. Additional case should be added for UNION case.
+ // One way of supporting UNION case can be seen at SPARK-24497 PR from Peter Toth.
+ val newRecursion = recursion.transform {
+ case r: UnionLoopRef if r.loopId == loopId =>
+ prevDF.queryExecution.optimizedPlan match {
+ case l: LocalRelation =>
+ prevPlan = l
+ l.copy(output = r.output)
+ case p @ Project(projectList, _: OneRowRelation) =>
Review Comment:
Added!
--
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-51609][SQL] Optimize Recursive CTE execution for simple queries [spark]
Posted by "Pajaraja (via GitHub)" <gi...@apache.org>.
Pajaraja commented on code in PR #50402:
URL: https://github.com/apache/spark/pull/50402#discussion_r2063256741
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/UnionLoopExec.scala:
##########
@@ -127,28 +135,71 @@ case class UnionLoopExec(
val numAnchorOutputRows = longMetric("numAnchorOutputRows")
val levelLimit = conf.getConf(SQLConf.CTE_RECURSION_LEVEL_LIMIT)
val rowLimit = conf.getConf(SQLConf.CTE_RECURSION_ROW_LIMIT)
+ val AnchorRowLimitToConvertToLocalLimit =
Review Comment:
Changed!
--
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-51609][SQL] Optimize Recursive CTE execution for simple queries [spark]
Posted by "Pajaraja (via GitHub)" <gi...@apache.org>.
Pajaraja commented on code in PR #50402:
URL: https://github.com/apache/spark/pull/50402#discussion_r2065917230
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/UnionLoopExec.scala:
##########
@@ -107,9 +110,33 @@ case class UnionLoopExec(
plan
}
val df = Dataset.ofRows(session, planWithLimit)
- val materializedDF = df.repartition()
- val count = materializedDF.queryExecution.toRdd.count()
- (materializedDF, count)
+
+ df.queryExecution.optimizedPlan match {
+ case l: LocalRelation =>
+ (df, l.data.length.toLong)
+ case Project(_, _: OneRowRelation) =>
+ if (localRelationLimit != 0) {
+ val local = LocalRelation.fromExternalRows(anchor.output, df.collect().toIndexedSeq)
Review Comment:
Changed!
--
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-51609][SQL] Optimize Recursive CTE execution for simple queries [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #50402:
URL: https://github.com/apache/spark/pull/50402#discussion_r2063199942
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/UnionLoopExec.scala:
##########
@@ -127,28 +135,71 @@ case class UnionLoopExec(
val numAnchorOutputRows = longMetric("numAnchorOutputRows")
val levelLimit = conf.getConf(SQLConf.CTE_RECURSION_LEVEL_LIMIT)
val rowLimit = conf.getConf(SQLConf.CTE_RECURSION_ROW_LIMIT)
+ val AnchorRowLimitToConvertToLocalLimit =
Review Comment:
```suggestion
val localRelationLimit =
```
--
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-51609][SQL] Optimize Recursive CTE execution for simple queries [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #50402:
URL: https://github.com/apache/spark/pull/50402#discussion_r2062943825
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/UnionLoopExec.scala:
##########
@@ -127,28 +134,67 @@ case class UnionLoopExec(
val numAnchorOutputRows = longMetric("numAnchorOutputRows")
val levelLimit = conf.getConf(SQLConf.CTE_RECURSION_LEVEL_LIMIT)
val rowLimit = conf.getConf(SQLConf.CTE_RECURSION_ROW_LIMIT)
+ val AnchorRowLimitToConverToLocalLimit =
+ conf.getConf(SQLConf.CTE_RECURSION_ANCHOR_ROWS_LIMIT_TO_CONVERT_TO_LOCAL_RELATION)
// currentLimit is initialized from the limit argument, and in each step it is decreased by
// the number of rows generated in that step.
// If limit is not passed down, currentLimit is set to be zero and won't be considered in the
// condition of while loop down (limit.isEmpty will be true).
var currentLimit = limit.getOrElse(-1)
- val unionChildren = mutable.ArrayBuffer.empty[LogicalRDD]
+ val unionChildren = mutable.ArrayBuffer.empty[LogicalPlan]
var (prevDF, prevCount) = executeAndCacheAndCount(anchor, currentLimit)
numAnchorOutputRows += prevCount
var currentLevel = 1
- var currentNumRows = 0
+ var currentNumRows: Long = 0
var limitReached: Boolean = false
val numPartitions = prevDF.queryExecution.toRdd.partitions.length
+
+ // In the case we return a sufficiently small number of rows when executing the anchor,
+ // we convert the result of the anchor into a LocalRelation, so that, if the recursion doesn't
+ // reference any external tables, we are able to calculate everything in the optimizer, using
+ // the ConvertToLocalRelation rule, which significantly improves runtime.
+ if (prevCount <= AnchorRowLimitToConverToLocalLimit &&
+ !prevDF.queryExecution.optimizedPlan.isInstanceOf[LocalRelation]) {
+ val local = LocalRelation.fromExternalRows(anchor.output, prevDF.collect().toIndexedSeq)
+ prevDF = Dataset.ofRows(session, local)
+ }
// Main loop for obtaining the result of the recursive query.
while (prevCount > 0 && !limitReached) {
+ var prevPlan: LogicalPlan = null
+ // the current plan is created by substituting UnionLoopRef node with the project node of
+ // the previous plan.
+ // This way we support only UNION ALL case. Additional case should be added for UNION case.
+ // One way of supporting UNION case can be seen at SPARK-24497 PR from Peter Toth.
+ val newRecursion = recursion.transform {
+ case r: UnionLoopRef if r.loopId == loopId =>
+ prevDF.queryExecution.optimizedPlan match {
+ case l: LocalRelation =>
+ prevPlan = l
+ l.copy(output = r.output)
+ case p @ Project(projectList, _: OneRowRelation) =>
Review Comment:
Can we add a comment to explain that this case is still needed if the to-local-relation optimization is disabled.
--
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-51609][SQL] Optimize Recursive CTE execution for simple queries [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #50402:
URL: https://github.com/apache/spark/pull/50402#discussion_r2050034665
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/UnionLoopExec.scala:
##########
@@ -149,6 +156,36 @@ case class UnionLoopExec(
val numPartitions = prevDF.queryExecution.toRdd.partitions.length
// Main loop for obtaining the result of the recursive query.
while (prevCount > 0 && !limitReached) {
+ var prevPlan: LogicalPlan = null
+ // the current plan is created by substituting UnionLoopRef node with the project node of
+ // the previous plan.
+ // This way we support only UNION ALL case. Additional case should be added for UNION case.
+ // One way of supporting UNION case can be seen at SPARK-24497 PR from Peter Toth.
+ val newRecursion = recursion.transform {
+ case r: UnionLoopRef if r.loopId == loopId =>
+ prevDF.queryExecution.optimizedPlan match {
+ case _: LocalRelation =>
+ prevPlan = prevDF.queryExecution.optimizedPlan
+ val prevPlanToRefMapping = prevPlan.output.zip(r.output).map {
+ case (fa, ta) => Alias(fa, ta.name)(ta.exprId)
+ }
+ Project(prevPlanToRefMapping, prevPlan)
Review Comment:
for LocalRelation, we can simply update its `output` field
--
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-51609][SQL] Optimize Recursive CTE execution for simple queries [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on PR #50402:
URL: https://github.com/apache/spark/pull/50402#issuecomment-2839128897
thanks, merging to master!
--
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-51609][SQL] Optimize Recursive CTE execution for simple queries [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on code in PR #50402:
URL: https://github.com/apache/spark/pull/50402#discussion_r2272011713
##########
sql/core/benchmarks/RecursiveCTEBenchmark-results.txt:
##########
@@ -0,0 +1,95 @@
+================================================================================================
+Recursive CTE with only LocalRelation
+================================================================================================
+
+OpenJDK 64-Bit Server VM 17.0.13+11-LTS on Mac OS X 15.4
+Apple M3 Max
Review Comment:
In addition, we need to generate both Java 17 and Java 21. Specifically, `RecursiveCTEBenchmark-jdk21-results.txt` is required together.
--
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-51609][SQL] Optimize Recursive CTE execution for simple queries [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on code in PR #50402:
URL: https://github.com/apache/spark/pull/50402#discussion_r2272010754
##########
sql/core/benchmarks/RecursiveCTEBenchmark-results.txt:
##########
@@ -0,0 +1,95 @@
+================================================================================================
+Recursive CTE with only LocalRelation
+================================================================================================
+
+OpenJDK 64-Bit Server VM 17.0.13+11-LTS on Mac OS X 15.4
+Apple M3 Max
Review Comment:
Hi, @Pajaraja .
FYI, Apache Spark community uses a standard way to generate benchmark result which is `benchmark.yaml` GitHub Action job. Please use that in your forked repository next time.
- https://github.com/apache/spark/actions/workflows/benchmark.yml
--
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-51609][SQL] Optimize Recursive CTE execution for simple queries [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #50402:
URL: https://github.com/apache/spark/pull/50402#discussion_r2063202471
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/UnionLoopExec.scala:
##########
@@ -127,28 +134,67 @@ case class UnionLoopExec(
val numAnchorOutputRows = longMetric("numAnchorOutputRows")
val levelLimit = conf.getConf(SQLConf.CTE_RECURSION_LEVEL_LIMIT)
val rowLimit = conf.getConf(SQLConf.CTE_RECURSION_ROW_LIMIT)
+ val AnchorRowLimitToConverToLocalLimit =
+ conf.getConf(SQLConf.CTE_RECURSION_ANCHOR_ROWS_LIMIT_TO_CONVERT_TO_LOCAL_RELATION)
// currentLimit is initialized from the limit argument, and in each step it is decreased by
// the number of rows generated in that step.
// If limit is not passed down, currentLimit is set to be zero and won't be considered in the
// condition of while loop down (limit.isEmpty will be true).
var currentLimit = limit.getOrElse(-1)
- val unionChildren = mutable.ArrayBuffer.empty[LogicalRDD]
+ val unionChildren = mutable.ArrayBuffer.empty[LogicalPlan]
var (prevDF, prevCount) = executeAndCacheAndCount(anchor, currentLimit)
numAnchorOutputRows += prevCount
var currentLevel = 1
- var currentNumRows = 0
+ var currentNumRows: Long = 0
var limitReached: Boolean = false
val numPartitions = prevDF.queryExecution.toRdd.partitions.length
+
+ // In the case we return a sufficiently small number of rows when executing the anchor,
+ // we convert the result of the anchor into a LocalRelation, so that, if the recursion doesn't
+ // reference any external tables, we are able to calculate everything in the optimizer, using
+ // the ConvertToLocalRelation rule, which significantly improves runtime.
+ if (prevCount <= AnchorRowLimitToConverToLocalLimit &&
+ !prevDF.queryExecution.optimizedPlan.isInstanceOf[LocalRelation]) {
+ val local = LocalRelation.fromExternalRows(anchor.output, prevDF.collect().toIndexedSeq)
Review Comment:
I think we should still put the code at the end of `def executeAndCacheAndCount`. Every iteration, we should try this optimization, in case the numRows dropped due to filter and can be optimized into local relation.
--
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-51609][SQL] Optimize Recursive CTE execution for simple queries [spark]
Posted by "Pajaraja (via GitHub)" <gi...@apache.org>.
Pajaraja commented on code in PR #50402:
URL: https://github.com/apache/spark/pull/50402#discussion_r2063196654
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/UnionLoopExec.scala:
##########
@@ -127,28 +134,67 @@ case class UnionLoopExec(
val numAnchorOutputRows = longMetric("numAnchorOutputRows")
val levelLimit = conf.getConf(SQLConf.CTE_RECURSION_LEVEL_LIMIT)
val rowLimit = conf.getConf(SQLConf.CTE_RECURSION_ROW_LIMIT)
+ val AnchorRowLimitToConverToLocalLimit =
+ conf.getConf(SQLConf.CTE_RECURSION_ANCHOR_ROWS_LIMIT_TO_CONVERT_TO_LOCAL_RELATION)
// currentLimit is initialized from the limit argument, and in each step it is decreased by
// the number of rows generated in that step.
// If limit is not passed down, currentLimit is set to be zero and won't be considered in the
// condition of while loop down (limit.isEmpty will be true).
var currentLimit = limit.getOrElse(-1)
- val unionChildren = mutable.ArrayBuffer.empty[LogicalRDD]
+ val unionChildren = mutable.ArrayBuffer.empty[LogicalPlan]
var (prevDF, prevCount) = executeAndCacheAndCount(anchor, currentLimit)
numAnchorOutputRows += prevCount
var currentLevel = 1
- var currentNumRows = 0
+ var currentNumRows: Long = 0
var limitReached: Boolean = false
val numPartitions = prevDF.queryExecution.toRdd.partitions.length
+
+ // In the case we return a sufficiently small number of rows when executing the anchor,
+ // we convert the result of the anchor into a LocalRelation, so that, if the recursion doesn't
+ // reference any external tables, we are able to calculate everything in the optimizer, using
+ // the ConvertToLocalRelation rule, which significantly improves runtime.
+ if (prevCount <= AnchorRowLimitToConverToLocalLimit &&
+ !prevDF.queryExecution.optimizedPlan.isInstanceOf[LocalRelation]) {
+ val local = LocalRelation.fromExternalRows(anchor.output, prevDF.collect().toIndexedSeq)
Review Comment:
Leaving it where it is then.
--
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-51609][SQL] Optimize Recursive CTE execution for simple queries [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan closed pull request #50402: [SPARK-51609][SQL] Optimize Recursive CTE execution for simple queries
URL: https://github.com/apache/spark/pull/50402
--
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-51609][SQL] Optimize Recursive CTE execution for simple queries [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #50402:
URL: https://github.com/apache/spark/pull/50402#discussion_r2065846065
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/UnionLoopExec.scala:
##########
@@ -107,9 +110,33 @@ case class UnionLoopExec(
plan
}
val df = Dataset.ofRows(session, planWithLimit)
- val materializedDF = df.repartition()
- val count = materializedDF.queryExecution.toRdd.count()
- (materializedDF, count)
+
+ df.queryExecution.optimizedPlan match {
+ case l: LocalRelation =>
+ (df, l.data.length.toLong)
+ case Project(_, _: OneRowRelation) =>
+ if (localRelationLimit != 0) {
+ val local = LocalRelation.fromExternalRows(anchor.output, df.collect().toIndexedSeq)
Review Comment:
nit: we should follow `ConvertToLocalRelation` and avoid submitting a job
```
case Project(projectList, OneRowRelation) if !projectList.exists(hasUnevaluableExpr) =>
val projection = new InterpretedMutableProjection(projectList, Nil)
projection.initialize(0)
LocalRelation(projectList.map(_.toAttribute), Seq(projection(InternalRow.empty)))
```
--
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-51609][SQL] Optimize Recursive CTE execution for simple queries [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #50402:
URL: https://github.com/apache/spark/pull/50402#discussion_r2056359238
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/UnionLoopExec.scala:
##########
@@ -149,6 +156,33 @@ case class UnionLoopExec(
val numPartitions = prevDF.queryExecution.toRdd.partitions.length
// Main loop for obtaining the result of the recursive query.
while (prevCount > 0 && !limitReached) {
+ var prevPlan: LogicalPlan = null
+ // the current plan is created by substituting UnionLoopRef node with the project node of
+ // the previous plan.
+ // This way we support only UNION ALL case. Additional case should be added for UNION case.
+ // One way of supporting UNION case can be seen at SPARK-24497 PR from Peter Toth.
+ val newRecursion = recursion.transform {
+ case r: UnionLoopRef if r.loopId == loopId =>
+ prevDF.queryExecution.optimizedPlan match {
+ case l: LocalRelation =>
+ prevPlan = prevDF.queryExecution.optimizedPlan
Review Comment:
```suggestion
prevPlan = l
```
--
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-51609][SQL] Optimize Recursive CTE execution for simple queries [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #50402:
URL: https://github.com/apache/spark/pull/50402#discussion_r2056364081
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/UnionLoopExec.scala:
##########
@@ -157,9 +191,6 @@ case class UnionLoopExec(
cause = null)
}
- // Inherit stats and constraints from the dataset of the previous iteration.
- val prevPlan = LogicalRDD.fromDataset(prevDF.queryExecution.toRdd, prevDF, prevDF.isStreaming)
- .newInstance()
unionChildren += prevPlan
currentNumRows += prevCount.toInt
Review Comment:
not related to this PR, but `.toInt` has the risk of overflow. Shall we make `currentNumRows` a long as well?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51609][SQL] Optimize Recursive CTE execution for simple queries [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #50402:
URL: https://github.com/apache/spark/pull/50402#discussion_r2063179054
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/UnionLoopExec.scala:
##########
@@ -127,28 +134,67 @@ case class UnionLoopExec(
val numAnchorOutputRows = longMetric("numAnchorOutputRows")
val levelLimit = conf.getConf(SQLConf.CTE_RECURSION_LEVEL_LIMIT)
val rowLimit = conf.getConf(SQLConf.CTE_RECURSION_ROW_LIMIT)
+ val AnchorRowLimitToConverToLocalLimit =
+ conf.getConf(SQLConf.CTE_RECURSION_ANCHOR_ROWS_LIMIT_TO_CONVERT_TO_LOCAL_RELATION)
// currentLimit is initialized from the limit argument, and in each step it is decreased by
// the number of rows generated in that step.
// If limit is not passed down, currentLimit is set to be zero and won't be considered in the
// condition of while loop down (limit.isEmpty will be true).
var currentLimit = limit.getOrElse(-1)
- val unionChildren = mutable.ArrayBuffer.empty[LogicalRDD]
+ val unionChildren = mutable.ArrayBuffer.empty[LogicalPlan]
var (prevDF, prevCount) = executeAndCacheAndCount(anchor, currentLimit)
numAnchorOutputRows += prevCount
var currentLevel = 1
- var currentNumRows = 0
+ var currentNumRows: Long = 0
var limitReached: Boolean = false
val numPartitions = prevDF.queryExecution.toRdd.partitions.length
+
+ // In the case we return a sufficiently small number of rows when executing the anchor,
+ // we convert the result of the anchor into a LocalRelation, so that, if the recursion doesn't
+ // reference any external tables, we are able to calculate everything in the optimizer, using
+ // the ConvertToLocalRelation rule, which significantly improves runtime.
+ if (prevCount <= AnchorRowLimitToConverToLocalLimit &&
+ !prevDF.queryExecution.optimizedPlan.isInstanceOf[LocalRelation]) {
+ val local = LocalRelation.fromExternalRows(anchor.output, prevDF.collect().toIndexedSeq)
Review Comment:
oh sorry I'm wrong, we need to do a shuffle so that we know the row count.
--
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