You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "peter-toth (via GitHub)" <gi...@apache.org> on 2025/01/06 19:50:56 UTC
[PR] [SPARK-50722][SQL] Detect self-contained WITH nodes [spark]
peter-toth opened a new pull request, #49379:
URL: https://github.com/apache/spark/pull/49379
### What changes were proposed in this pull request?
This is a follow-up of https://github.com/apache/spark/pull/48284 to make the CTE inline decision smarter but without introducing overly complex linage tracking. This PR is an alternative fix to https://github.com/apache/spark/pull/49352 with tracking CTE references a different way.
First of all, this PR reverts https://github.com/apache/spark/pull/48284 and keeps tracking "contains" relation in `outgoingRefs`. An example for this relation is "Case 2" in https://github.com/apache/spark/pull/49352 in which `t2` contains `t3`. But while https://github.com/apache/spark/pull/48284 strores the reference from `t2` -> `t3` in `t2`'s `indirectOutgoingRefSources`, this PR reverts to store the reference in `t2`'s `outgoingRefs` similar to other usual references.
Obviously, the problem with this revert is that queries with duplicate CTE definitions can fail, so let's revisit the problematic query from https://github.com/apache/spark/pull/48284:
```
WithCTE
CTEDef r0
...
CTEDef r1
View v
WithCTE
CTEDef t1
OneRowRelation
CTEDef t2
CTERef t1
CTERef t2 // main query of the inner WithCTE
CTEDef r2
View v // exactly the same as the view v above
WithCTE
CTEDef t1
OneRowRelation
CTEDef t2
CTERef t1
CTERef t2
CTERef r2 // main query of the outer WithCTE
```
The source of the problem is view `v`, which duplicates a `WithCTE` node and introduces conflicting CTE definitions.
In the above example, tracking both the `r1` -> `t2` and the `r2` -> `t2` "contains" relations in `r1`'s and `r2`'s `outgoingRefs`s cause reference counting problems, as those `t2`s are actually diffrent instances and so they can have different inlining depending on how many times `r1` and `r2` are referenced.
Fortunately, this duplicate issue can only happen when the conflicting `WithCTE` nodes are self-contained (as those nodes can't contain any references to any outer definitions like `r0`). The main advantage of self-contained `WithCTE` nodes is that we need don't need to track `r1` -> `t2` and `r2` -> `t2` references, as the inlining of `t2` (and `t1`) inside the `WithCTE` node don't depend on references coming from outer (`r1`, `r2`) definition. (Please note that what depends on `r1` and `r2` is the inlined result of the whole inner `WithCTE` node.)
So to recap:
- we do need to track references from "contains" relation when the inner `WithCTE` is not self-contained,
- we don't need to (and actually it causes issues if we do) when the inner `WithCTE` is self-contained and the node appears multiple times in the query.
- if the inner `WithCTE` is self-contained, but it only appears only once in the quey, then the both ways are fine
This PR suggests the following:
- As during the traversal of `buildCTEMap()` we don't know in advance if a `WithCTE` node is self-contained or not, let's:
- track the references from "contains" relations
- and also record what is the container of a definition.
E.g. when we encounter `t2` in `r1`, then record `t2` in `r1`'s `outgoingRefs` and record `r1` as the container of `t2`.
- If we encounter a duplicate definition, then we know it is self-contained so we:
- don't need to visit it again
- and we can also remove the reference from its container to the definition.
E.g. when we encounter `t2` in `r2`, then clear `t2` from `r1`'s `outgoingRefs` (we know that `r1` was `t2`'s container).
### Why are the changes needed?
Make sure we can inline CTE when we should.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
Added tests from https://github.com/apache/spark/pull/49352.
### 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-50722][SQL] Detect self-contained WITH nodes [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on PR #49379:
URL: https://github.com/apache/spark/pull/49379#issuecomment-2574257420
Yea, this is a simpler approach. Although it still unnecessarily tracks the `outgoingRefs` for the "contains" relationship if `WithCTE` is not duplicated, but it's not a big issue.
--
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-50722][SQL] Detect self-contained WITH nodes [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #49379:
URL: https://github.com/apache/spark/pull/49379#discussion_r1905420707
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InlineCTE.scala:
##########
@@ -71,53 +71,64 @@ case class InlineCTE(
* @param plan The plan to collect the CTEs from
* @param cteMap A mutable map that accumulates the CTEs and their reference information by CTE
* ids.
- * @param collectCTERefs A function to collect CTE references so that the caller side can do some
- * bookkeeping work.
+ * @param outerCTEId While collecting the map we use this optional CTE id to identify the
+ * current outer CTE.
*/
private def buildCTEMap(
plan: LogicalPlan,
cteMap: mutable.Map[Long, CTEReferenceInfo],
- collectCTERefs: CTERelationRef => Unit = _ => ()): Unit = {
+ outerCTEId: Option[Long] = None): Unit = {
plan match {
case WithCTE(child, cteDefs) =>
- cteDefs.foreach { cteDef =>
- cteMap(cteDef.id) = CTEReferenceInfo(
- cteDef = cteDef,
- refCount = 0,
- outgoingRefs = mutable.Map.empty.withDefaultValue(0),
- shouldInline = true
- )
- }
- cteDefs.foreach { cteDef =>
- buildCTEMap(cteDef, cteMap, ref => {
- // A CTE relation can references CTE relations defined before it in the same `WithCTE`.
- // Here we update the out-going-ref-count for it, in case this CTE relation is not
- // referenced at all and can be optimized out, and we need to decrease the ref counts
- // for CTE relations that are referenced by it.
- if (cteDefs.exists(_.id == ref.cteId)) {
- cteMap(cteDef.id).increaseOutgoingRefCount(ref.cteId, 1)
- }
- // Similarly, a CTE relation can reference CTE relations defined in the outer `WithCTE`.
- // Here we call the `collectCTERefs` function so that the outer CTE can also update the
- // out-going-ref-count if needed.
- collectCTERefs(ref)
- })
+ val isDuplicated = cteDefs.forall(cteDef => cteMap.contains(cteDef.id))
+ if (isDuplicated) {
+ // If we have seen this `WithCTE` node then it must be self-contained so we can clear
+ // the references from containers to the definitions, and we don't need to process it
+ // again
+
+ cteDefs.foreach { cteDef =>
+ cteMap(cteDef.id).container.foreach(c => cteMap(c).outgoingRefs -= cteDef.id)
+ }
+ } else {
+ cteDefs.foreach { cteDef =>
+ cteMap(cteDef.id) = CTEReferenceInfo(
+ cteDef = cteDef,
+ refCount = 0,
+ outgoingRefs = mutable.Map.empty.withDefaultValue(0),
+ shouldInline = true,
+ container = outerCTEId
+ )
+ }
+
+ cteDefs.foreach { cteDef =>
+ buildCTEMap(cteDef, cteMap, Some(cteDef.id))
+ }
+ buildCTEMap(child, cteMap, outerCTEId)
}
- buildCTEMap(child, cteMap, collectCTERefs)
case ref: CTERelationRef =>
cteMap(ref.cteId) = cteMap(ref.cteId).withRefCountIncreased(1)
- collectCTERefs(ref)
+
+ // The `outerCTEId` CTE definition can either reference `cteId` definition if `cteId` is in
+ // the same or in an outer `WithCTE` node, or `outerCTEId` can contain `cteId` definition if
+ // `cteId` is an inner `WithCTE` node inside `outerCTEId`.
+ // In both cases we can track the relations in `outgoingRefs` when we see a definition the
+ // first time. But if we encounter a conflicting duplicated contains relation later, then we
+ // will remove the references of the first contains relation.
+ outerCTEId.foreach { cteId =>
Review Comment:
one idea: if we carry over not only the outer CTE id, but also all ids of the CTE relations that are defined within the same `WithCTE` of that outer CTE, then we only call `increaseOutgoingRefCount` if this is really a reference.
Does it 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-50722][SQL] Detect self-contained WITH nodes [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #49379:
URL: https://github.com/apache/spark/pull/49379#discussion_r1905417173
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InlineCTE.scala:
##########
@@ -71,53 +71,64 @@ case class InlineCTE(
* @param plan The plan to collect the CTEs from
* @param cteMap A mutable map that accumulates the CTEs and their reference information by CTE
* ids.
- * @param collectCTERefs A function to collect CTE references so that the caller side can do some
- * bookkeeping work.
+ * @param outerCTEId While collecting the map we use this optional CTE id to identify the
+ * current outer CTE.
*/
private def buildCTEMap(
plan: LogicalPlan,
cteMap: mutable.Map[Long, CTEReferenceInfo],
- collectCTERefs: CTERelationRef => Unit = _ => ()): Unit = {
+ outerCTEId: Option[Long] = None): Unit = {
plan match {
case WithCTE(child, cteDefs) =>
- cteDefs.foreach { cteDef =>
- cteMap(cteDef.id) = CTEReferenceInfo(
- cteDef = cteDef,
- refCount = 0,
- outgoingRefs = mutable.Map.empty.withDefaultValue(0),
- shouldInline = true
- )
- }
- cteDefs.foreach { cteDef =>
- buildCTEMap(cteDef, cteMap, ref => {
- // A CTE relation can references CTE relations defined before it in the same `WithCTE`.
- // Here we update the out-going-ref-count for it, in case this CTE relation is not
- // referenced at all and can be optimized out, and we need to decrease the ref counts
- // for CTE relations that are referenced by it.
- if (cteDefs.exists(_.id == ref.cteId)) {
- cteMap(cteDef.id).increaseOutgoingRefCount(ref.cteId, 1)
- }
- // Similarly, a CTE relation can reference CTE relations defined in the outer `WithCTE`.
- // Here we call the `collectCTERefs` function so that the outer CTE can also update the
- // out-going-ref-count if needed.
- collectCTERefs(ref)
- })
+ val isDuplicated = cteDefs.forall(cteDef => cteMap.contains(cteDef.id))
+ if (isDuplicated) {
+ // If we have seen this `WithCTE` node then it must be self-contained so we can clear
+ // the references from containers to the definitions, and we don't need to process it
+ // again
+
+ cteDefs.foreach { cteDef =>
+ cteMap(cteDef.id).container.foreach(c => cteMap(c).outgoingRefs -= cteDef.id)
Review Comment:
Yeah, technically this is what we do. We exclude all CTE definitions of the inner `WithCTE` from the container's `outgoingRefs`. (We don't collect and exclude only those defs that are in the inner's main query, because we can simply exclude all.)
--
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-50722][SQL] Detect self-contained WITH nodes [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #49379:
URL: https://github.com/apache/spark/pull/49379#discussion_r1905478528
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InlineCTE.scala:
##########
@@ -71,53 +71,64 @@ case class InlineCTE(
* @param plan The plan to collect the CTEs from
* @param cteMap A mutable map that accumulates the CTEs and their reference information by CTE
* ids.
- * @param collectCTERefs A function to collect CTE references so that the caller side can do some
- * bookkeeping work.
+ * @param outerCTEId While collecting the map we use this optional CTE id to identify the
+ * current outer CTE.
*/
private def buildCTEMap(
plan: LogicalPlan,
cteMap: mutable.Map[Long, CTEReferenceInfo],
- collectCTERefs: CTERelationRef => Unit = _ => ()): Unit = {
+ outerCTEId: Option[Long] = None): Unit = {
plan match {
case WithCTE(child, cteDefs) =>
- cteDefs.foreach { cteDef =>
- cteMap(cteDef.id) = CTEReferenceInfo(
- cteDef = cteDef,
- refCount = 0,
- outgoingRefs = mutable.Map.empty.withDefaultValue(0),
- shouldInline = true
- )
- }
- cteDefs.foreach { cteDef =>
- buildCTEMap(cteDef, cteMap, ref => {
- // A CTE relation can references CTE relations defined before it in the same `WithCTE`.
- // Here we update the out-going-ref-count for it, in case this CTE relation is not
- // referenced at all and can be optimized out, and we need to decrease the ref counts
- // for CTE relations that are referenced by it.
- if (cteDefs.exists(_.id == ref.cteId)) {
- cteMap(cteDef.id).increaseOutgoingRefCount(ref.cteId, 1)
- }
- // Similarly, a CTE relation can reference CTE relations defined in the outer `WithCTE`.
- // Here we call the `collectCTERefs` function so that the outer CTE can also update the
- // out-going-ref-count if needed.
- collectCTERefs(ref)
- })
+ val isDuplicated = cteDefs.forall(cteDef => cteMap.contains(cteDef.id))
+ if (isDuplicated) {
+ // If we have seen this `WithCTE` node then it must be self-contained so we can clear
+ // the references from containers to the definitions, and we don't need to process it
+ // again
+
+ cteDefs.foreach { cteDef =>
+ cteMap(cteDef.id).container.foreach(c => cteMap(c).outgoingRefs -= cteDef.id)
Review Comment:
If it is confusuing that we store the container for each elements of `cteDefs` (as all have the same container), then we could store it for the first CTE def, and determine `isDuplicated` based on the presence of the first CTE def in the map. If that looks better...
--
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-50722][SQL] Detect self-contained WITH nodes [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #49379:
URL: https://github.com/apache/spark/pull/49379#discussion_r1905229292
##########
sql/core/src/test/scala/org/apache/spark/sql/CTEInlineSuite.scala:
##########
@@ -735,6 +736,86 @@ abstract class CTEInlineSuiteBase
checkAnswer(df, Row(1))
}
}
+
Review Comment:
Fixed in: https://github.com/apache/spark/pull/49379/commits/c62c6c4bf2ec4591fee1c5129315a2d9878159a9
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InlineCTE.scala:
##########
@@ -225,12 +233,15 @@ case class InlineCTE(
* from other CTE relations and regular places.
* @param outgoingRefs A mutable map that tracks outgoing reference counts to other CTE relations.
* @param shouldInline If true, this CTE relation should be inlined in the places that reference it.
+ * @param container The container of a CTE definition is another CTE definition in witch the
Review Comment:
Done in https://github.com/apache/spark/pull/49379/commits/c62c6c4bf2ec4591fee1c5129315a2d9878159a9.
--
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-50722][SQL] Detect self-contained WITH nodes [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #49379:
URL: https://github.com/apache/spark/pull/49379#discussion_r1905320893
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InlineCTE.scala:
##########
@@ -71,53 +71,64 @@ case class InlineCTE(
* @param plan The plan to collect the CTEs from
* @param cteMap A mutable map that accumulates the CTEs and their reference information by CTE
* ids.
- * @param collectCTERefs A function to collect CTE references so that the caller side can do some
- * bookkeeping work.
+ * @param outerCTEId While collecting the map we use this optional CTE id to identify the
+ * current outer CTE.
*/
private def buildCTEMap(
plan: LogicalPlan,
cteMap: mutable.Map[Long, CTEReferenceInfo],
- collectCTERefs: CTERelationRef => Unit = _ => ()): Unit = {
+ outerCTEId: Option[Long] = None): Unit = {
plan match {
case WithCTE(child, cteDefs) =>
- cteDefs.foreach { cteDef =>
- cteMap(cteDef.id) = CTEReferenceInfo(
- cteDef = cteDef,
- refCount = 0,
- outgoingRefs = mutable.Map.empty.withDefaultValue(0),
- shouldInline = true
- )
- }
- cteDefs.foreach { cteDef =>
- buildCTEMap(cteDef, cteMap, ref => {
- // A CTE relation can references CTE relations defined before it in the same `WithCTE`.
- // Here we update the out-going-ref-count for it, in case this CTE relation is not
- // referenced at all and can be optimized out, and we need to decrease the ref counts
- // for CTE relations that are referenced by it.
- if (cteDefs.exists(_.id == ref.cteId)) {
- cteMap(cteDef.id).increaseOutgoingRefCount(ref.cteId, 1)
- }
- // Similarly, a CTE relation can reference CTE relations defined in the outer `WithCTE`.
- // Here we call the `collectCTERefs` function so that the outer CTE can also update the
- // out-going-ref-count if needed.
- collectCTERefs(ref)
- })
+ val isDuplicated = cteDefs.forall(cteDef => cteMap.contains(cteDef.id))
+ if (isDuplicated) {
+ // If we have seen this `WithCTE` node then it must be self-contained so we can clear
+ // the references from containers to the definitions, and we don't need to process it
+ // again
+
+ cteDefs.foreach { cteDef =>
+ cteMap(cteDef.id).container.foreach(c => cteMap(c).outgoingRefs -= cteDef.id)
Review Comment:
A nested `WithCTE` is also duplicated, but it can legally reference outer CTE relations that are defined within this self-contained CTE.
--
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-50722][SQL] Detect self-contained WITH nodes [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on PR #49379:
URL: https://github.com/apache/spark/pull/49379#issuecomment-2576514544
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-50722][SQL] Detect self-contained WITH nodes [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #49379:
URL: https://github.com/apache/spark/pull/49379#discussion_r1905328356
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InlineCTE.scala:
##########
@@ -71,53 +71,64 @@ case class InlineCTE(
* @param plan The plan to collect the CTEs from
* @param cteMap A mutable map that accumulates the CTEs and their reference information by CTE
* ids.
- * @param collectCTERefs A function to collect CTE references so that the caller side can do some
- * bookkeeping work.
+ * @param outerCTEId While collecting the map we use this optional CTE id to identify the
+ * current outer CTE.
*/
private def buildCTEMap(
plan: LogicalPlan,
cteMap: mutable.Map[Long, CTEReferenceInfo],
- collectCTERefs: CTERelationRef => Unit = _ => ()): Unit = {
+ outerCTEId: Option[Long] = None): Unit = {
plan match {
case WithCTE(child, cteDefs) =>
- cteDefs.foreach { cteDef =>
- cteMap(cteDef.id) = CTEReferenceInfo(
- cteDef = cteDef,
- refCount = 0,
- outgoingRefs = mutable.Map.empty.withDefaultValue(0),
- shouldInline = true
- )
- }
- cteDefs.foreach { cteDef =>
- buildCTEMap(cteDef, cteMap, ref => {
- // A CTE relation can references CTE relations defined before it in the same `WithCTE`.
- // Here we update the out-going-ref-count for it, in case this CTE relation is not
- // referenced at all and can be optimized out, and we need to decrease the ref counts
- // for CTE relations that are referenced by it.
- if (cteDefs.exists(_.id == ref.cteId)) {
- cteMap(cteDef.id).increaseOutgoingRefCount(ref.cteId, 1)
- }
- // Similarly, a CTE relation can reference CTE relations defined in the outer `WithCTE`.
- // Here we call the `collectCTERefs` function so that the outer CTE can also update the
- // out-going-ref-count if needed.
- collectCTERefs(ref)
- })
+ val isDuplicated = cteDefs.forall(cteDef => cteMap.contains(cteDef.id))
+ if (isDuplicated) {
+ // If we have seen this `WithCTE` node then it must be self-contained so we can clear
+ // the references from containers to the definitions, and we don't need to process it
+ // again
+
+ cteDefs.foreach { cteDef =>
+ cteMap(cteDef.id).container.foreach(c => cteMap(c).outgoingRefs -= cteDef.id)
Review Comment:
Can you give an example? But please note that we never visit any duplicates 2 times...
--
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-50722][SQL] Detect self-contained WITH nodes [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #49379:
URL: https://github.com/apache/spark/pull/49379#discussion_r1905319934
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InlineCTE.scala:
##########
@@ -71,53 +71,64 @@ case class InlineCTE(
* @param plan The plan to collect the CTEs from
* @param cteMap A mutable map that accumulates the CTEs and their reference information by CTE
* ids.
- * @param collectCTERefs A function to collect CTE references so that the caller side can do some
- * bookkeeping work.
+ * @param outerCTEId While collecting the map we use this optional CTE id to identify the
+ * current outer CTE.
*/
private def buildCTEMap(
plan: LogicalPlan,
cteMap: mutable.Map[Long, CTEReferenceInfo],
- collectCTERefs: CTERelationRef => Unit = _ => ()): Unit = {
+ outerCTEId: Option[Long] = None): Unit = {
plan match {
case WithCTE(child, cteDefs) =>
- cteDefs.foreach { cteDef =>
- cteMap(cteDef.id) = CTEReferenceInfo(
- cteDef = cteDef,
- refCount = 0,
- outgoingRefs = mutable.Map.empty.withDefaultValue(0),
- shouldInline = true
- )
- }
- cteDefs.foreach { cteDef =>
- buildCTEMap(cteDef, cteMap, ref => {
- // A CTE relation can references CTE relations defined before it in the same `WithCTE`.
- // Here we update the out-going-ref-count for it, in case this CTE relation is not
- // referenced at all and can be optimized out, and we need to decrease the ref counts
- // for CTE relations that are referenced by it.
- if (cteDefs.exists(_.id == ref.cteId)) {
- cteMap(cteDef.id).increaseOutgoingRefCount(ref.cteId, 1)
- }
- // Similarly, a CTE relation can reference CTE relations defined in the outer `WithCTE`.
- // Here we call the `collectCTERefs` function so that the outer CTE can also update the
- // out-going-ref-count if needed.
- collectCTERefs(ref)
- })
+ val isDuplicated = cteDefs.forall(cteDef => cteMap.contains(cteDef.id))
+ if (isDuplicated) {
+ // If we have seen this `WithCTE` node then it must be self-contained so we can clear
+ // the references from containers to the definitions, and we don't need to process it
+ // again
+
+ cteDefs.foreach { cteDef =>
+ cteMap(cteDef.id).container.foreach(c => cteMap(c).outgoingRefs -= cteDef.id)
Review Comment:
Oh wait, is there a problem for nested `WithCTE` inside duplicated `WithCTE`? We should not update `outgoingRefs` from nested `WithCTE`, which might mess up the ref count for this self-contained CTE.
--
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-50722][SQL] Detect self-contained WITH nodes [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #49379:
URL: https://github.com/apache/spark/pull/49379#discussion_r1904820016
##########
sql/core/src/test/scala/org/apache/spark/sql/CTEInlineSuite.scala:
##########
@@ -735,6 +736,86 @@ abstract class CTEInlineSuiteBase
checkAnswer(df, Row(1))
}
}
+
Review Comment:
shall we update the test name of the origin `SPARK-49816` test?
--
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-50722][SQL] Detect self-contained WITH nodes [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #49379:
URL: https://github.com/apache/spark/pull/49379#discussion_r1905420707
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InlineCTE.scala:
##########
@@ -71,53 +71,64 @@ case class InlineCTE(
* @param plan The plan to collect the CTEs from
* @param cteMap A mutable map that accumulates the CTEs and their reference information by CTE
* ids.
- * @param collectCTERefs A function to collect CTE references so that the caller side can do some
- * bookkeeping work.
+ * @param outerCTEId While collecting the map we use this optional CTE id to identify the
+ * current outer CTE.
*/
private def buildCTEMap(
plan: LogicalPlan,
cteMap: mutable.Map[Long, CTEReferenceInfo],
- collectCTERefs: CTERelationRef => Unit = _ => ()): Unit = {
+ outerCTEId: Option[Long] = None): Unit = {
plan match {
case WithCTE(child, cteDefs) =>
- cteDefs.foreach { cteDef =>
- cteMap(cteDef.id) = CTEReferenceInfo(
- cteDef = cteDef,
- refCount = 0,
- outgoingRefs = mutable.Map.empty.withDefaultValue(0),
- shouldInline = true
- )
- }
- cteDefs.foreach { cteDef =>
- buildCTEMap(cteDef, cteMap, ref => {
- // A CTE relation can references CTE relations defined before it in the same `WithCTE`.
- // Here we update the out-going-ref-count for it, in case this CTE relation is not
- // referenced at all and can be optimized out, and we need to decrease the ref counts
- // for CTE relations that are referenced by it.
- if (cteDefs.exists(_.id == ref.cteId)) {
- cteMap(cteDef.id).increaseOutgoingRefCount(ref.cteId, 1)
- }
- // Similarly, a CTE relation can reference CTE relations defined in the outer `WithCTE`.
- // Here we call the `collectCTERefs` function so that the outer CTE can also update the
- // out-going-ref-count if needed.
- collectCTERefs(ref)
- })
+ val isDuplicated = cteDefs.forall(cteDef => cteMap.contains(cteDef.id))
+ if (isDuplicated) {
+ // If we have seen this `WithCTE` node then it must be self-contained so we can clear
+ // the references from containers to the definitions, and we don't need to process it
+ // again
+
+ cteDefs.foreach { cteDef =>
+ cteMap(cteDef.id).container.foreach(c => cteMap(c).outgoingRefs -= cteDef.id)
+ }
+ } else {
+ cteDefs.foreach { cteDef =>
+ cteMap(cteDef.id) = CTEReferenceInfo(
+ cteDef = cteDef,
+ refCount = 0,
+ outgoingRefs = mutable.Map.empty.withDefaultValue(0),
+ shouldInline = true,
+ container = outerCTEId
+ )
+ }
+
+ cteDefs.foreach { cteDef =>
+ buildCTEMap(cteDef, cteMap, Some(cteDef.id))
+ }
+ buildCTEMap(child, cteMap, outerCTEId)
}
- buildCTEMap(child, cteMap, collectCTERefs)
case ref: CTERelationRef =>
cteMap(ref.cteId) = cteMap(ref.cteId).withRefCountIncreased(1)
- collectCTERefs(ref)
+
+ // The `outerCTEId` CTE definition can either reference `cteId` definition if `cteId` is in
+ // the same or in an outer `WithCTE` node, or `outerCTEId` can contain `cteId` definition if
+ // `cteId` is an inner `WithCTE` node inside `outerCTEId`.
+ // In both cases we can track the relations in `outgoingRefs` when we see a definition the
+ // first time. But if we encounter a conflicting duplicated contains relation later, then we
+ // will remove the references of the first contains relation.
+ outerCTEId.foreach { cteId =>
Review Comment:
one idea: if we carry over not only the outer CTE id, but also all ids of the CTE relations that are defined within the same `WithCTE` of that outer CTE, then we only call `increaseOutgoingRefCount` if this is really a reference.
Does it 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-50722][SQL] Detect self-contained WITH nodes [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on PR #49379:
URL: https://github.com/apache/spark/pull/49379#issuecomment-2575264547
For future reviewers:
The main problem is when we see
```
WITH
t0 AS (SELECT 1),
t1 AS (
WITH t2 (...)
SELECT * FROM t2
)
```
Shall we say that `t1` references `t2`? It depends on whether `t2` references `t0` or not. If not, the inner `WITH` is self-contained and `t1` doesn't really reference `t2`. Most of the time we don't need to be so strict. It's no harm to always say that `t1` references `t2`, as the ref counting works fine. However, the issue occurs when there are duplicated `WithCTE` in the plan tree.
Given that it's not cheap to detect self-contained CTE ("reference lineage" from https://github.com/apache/spark/pull/49352 is a bit complicated), when we see duplicated `WithCTE`, we know it's self-contained and we exclude `t2` from `t1`'s `outgoingRefs`.
--
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-50722][SQL] Detect self-contained WITH nodes [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #49379:
URL: https://github.com/apache/spark/pull/49379#discussion_r1904819037
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InlineCTE.scala:
##########
@@ -71,53 +71,61 @@ case class InlineCTE(
* @param plan The plan to collect the CTEs from
* @param cteMap A mutable map that accumulates the CTEs and their reference information by CTE
* ids.
- * @param collectCTERefs A function to collect CTE references so that the caller side can do some
- * bookkeeping work.
+ * @param outerCTEId While collecting the map we use this optional CTE id to identify the
+ * current outer CTE.
*/
private def buildCTEMap(
plan: LogicalPlan,
cteMap: mutable.Map[Long, CTEReferenceInfo],
- collectCTERefs: CTERelationRef => Unit = _ => ()): Unit = {
+ outerCTEId: Option[Long] = None): Unit = {
plan match {
case WithCTE(child, cteDefs) =>
+ var visit = true
Review Comment:
shall we more explicit about duplicated `WithCTE`? e.g.
```
val isDuplicated = cteDefs.forall(d => cteMap.contains(d.id))
```
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InlineCTE.scala:
##########
@@ -71,53 +71,61 @@ case class InlineCTE(
* @param plan The plan to collect the CTEs from
* @param cteMap A mutable map that accumulates the CTEs and their reference information by CTE
* ids.
- * @param collectCTERefs A function to collect CTE references so that the caller side can do some
- * bookkeeping work.
+ * @param outerCTEId While collecting the map we use this optional CTE id to identify the
+ * current outer CTE.
*/
private def buildCTEMap(
plan: LogicalPlan,
cteMap: mutable.Map[Long, CTEReferenceInfo],
- collectCTERefs: CTERelationRef => Unit = _ => ()): Unit = {
+ outerCTEId: Option[Long] = None): Unit = {
plan match {
case WithCTE(child, cteDefs) =>
+ var visit = true
Review Comment:
shall we be more explicit about duplicated `WithCTE`? e.g.
```
val isDuplicated = cteDefs.forall(d => cteMap.contains(d.id))
```
--
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-50722][SQL] Detect self-contained WITH nodes [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #49379:
URL: https://github.com/apache/spark/pull/49379#discussion_r1905310642
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InlineCTE.scala:
##########
@@ -71,53 +71,64 @@ case class InlineCTE(
* @param plan The plan to collect the CTEs from
* @param cteMap A mutable map that accumulates the CTEs and their reference information by CTE
* ids.
- * @param collectCTERefs A function to collect CTE references so that the caller side can do some
- * bookkeeping work.
+ * @param outerCTEId While collecting the map we use this optional CTE id to identify the
+ * current outer CTE.
*/
private def buildCTEMap(
plan: LogicalPlan,
cteMap: mutable.Map[Long, CTEReferenceInfo],
- collectCTERefs: CTERelationRef => Unit = _ => ()): Unit = {
+ outerCTEId: Option[Long] = None): Unit = {
plan match {
case WithCTE(child, cteDefs) =>
- cteDefs.foreach { cteDef =>
- cteMap(cteDef.id) = CTEReferenceInfo(
- cteDef = cteDef,
- refCount = 0,
- outgoingRefs = mutable.Map.empty.withDefaultValue(0),
- shouldInline = true
- )
- }
- cteDefs.foreach { cteDef =>
- buildCTEMap(cteDef, cteMap, ref => {
- // A CTE relation can references CTE relations defined before it in the same `WithCTE`.
- // Here we update the out-going-ref-count for it, in case this CTE relation is not
- // referenced at all and can be optimized out, and we need to decrease the ref counts
- // for CTE relations that are referenced by it.
- if (cteDefs.exists(_.id == ref.cteId)) {
- cteMap(cteDef.id).increaseOutgoingRefCount(ref.cteId, 1)
- }
- // Similarly, a CTE relation can reference CTE relations defined in the outer `WithCTE`.
- // Here we call the `collectCTERefs` function so that the outer CTE can also update the
- // out-going-ref-count if needed.
- collectCTERefs(ref)
- })
+ val isDuplicated = cteDefs.forall(cteDef => cteMap.contains(cteDef.id))
+ if (isDuplicated) {
+ // If we have seen this `WithCTE` node then it must be self-contained so we can clear
+ // the references from containers to the definitions, and we don't need to process it
+ // again
+
+ cteDefs.foreach { cteDef =>
+ cteMap(cteDef.id).container.foreach(c => cteMap(c).outgoingRefs -= cteDef.id)
Review Comment:
note: Ideally we should do this for all self-contained CTEs, but only duplicated `WithCTE` has the problem, so this fix is fine.
--
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-50722][SQL] Detect self-contained WITH nodes [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan closed pull request #49379: [SPARK-50722][SQL] Detect self-contained WITH nodes
URL: https://github.com/apache/spark/pull/49379
--
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-50722][SQL] Detect self-contained WITH nodes [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #49379:
URL: https://github.com/apache/spark/pull/49379#discussion_r1905228788
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InlineCTE.scala:
##########
@@ -71,53 +71,61 @@ case class InlineCTE(
* @param plan The plan to collect the CTEs from
* @param cteMap A mutable map that accumulates the CTEs and their reference information by CTE
* ids.
- * @param collectCTERefs A function to collect CTE references so that the caller side can do some
- * bookkeeping work.
+ * @param outerCTEId While collecting the map we use this optional CTE id to identify the
+ * current outer CTE.
*/
private def buildCTEMap(
plan: LogicalPlan,
cteMap: mutable.Map[Long, CTEReferenceInfo],
- collectCTERefs: CTERelationRef => Unit = _ => ()): Unit = {
+ outerCTEId: Option[Long] = None): Unit = {
plan match {
case WithCTE(child, cteDefs) =>
+ var visit = true
Review Comment:
Yeah, sure. Fixed in https://github.com/apache/spark/pull/49379/commits/c62c6c4bf2ec4591fee1c5129315a2d9878159a9.
--
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-50722][SQL] Detect self-contained WITH nodes [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #49379:
URL: https://github.com/apache/spark/pull/49379#discussion_r1905417173
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InlineCTE.scala:
##########
@@ -71,53 +71,64 @@ case class InlineCTE(
* @param plan The plan to collect the CTEs from
* @param cteMap A mutable map that accumulates the CTEs and their reference information by CTE
* ids.
- * @param collectCTERefs A function to collect CTE references so that the caller side can do some
- * bookkeeping work.
+ * @param outerCTEId While collecting the map we use this optional CTE id to identify the
+ * current outer CTE.
*/
private def buildCTEMap(
plan: LogicalPlan,
cteMap: mutable.Map[Long, CTEReferenceInfo],
- collectCTERefs: CTERelationRef => Unit = _ => ()): Unit = {
+ outerCTEId: Option[Long] = None): Unit = {
plan match {
case WithCTE(child, cteDefs) =>
- cteDefs.foreach { cteDef =>
- cteMap(cteDef.id) = CTEReferenceInfo(
- cteDef = cteDef,
- refCount = 0,
- outgoingRefs = mutable.Map.empty.withDefaultValue(0),
- shouldInline = true
- )
- }
- cteDefs.foreach { cteDef =>
- buildCTEMap(cteDef, cteMap, ref => {
- // A CTE relation can references CTE relations defined before it in the same `WithCTE`.
- // Here we update the out-going-ref-count for it, in case this CTE relation is not
- // referenced at all and can be optimized out, and we need to decrease the ref counts
- // for CTE relations that are referenced by it.
- if (cteDefs.exists(_.id == ref.cteId)) {
- cteMap(cteDef.id).increaseOutgoingRefCount(ref.cteId, 1)
- }
- // Similarly, a CTE relation can reference CTE relations defined in the outer `WithCTE`.
- // Here we call the `collectCTERefs` function so that the outer CTE can also update the
- // out-going-ref-count if needed.
- collectCTERefs(ref)
- })
+ val isDuplicated = cteDefs.forall(cteDef => cteMap.contains(cteDef.id))
+ if (isDuplicated) {
+ // If we have seen this `WithCTE` node then it must be self-contained so we can clear
+ // the references from containers to the definitions, and we don't need to process it
+ // again
+
+ cteDefs.foreach { cteDef =>
+ cteMap(cteDef.id).container.foreach(c => cteMap(c).outgoingRefs -= cteDef.id)
Review Comment:
Yeah, technically this is what we do. We exclude all CTE definitions of the inner `WithCTE` from the container's `outgoingRefs`. (We don't collect and exclude only those defs that are in the inner's main query.)
--
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-50722][SQL] Detect self-contained WITH nodes [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #49379:
URL: https://github.com/apache/spark/pull/49379#discussion_r1905399443
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InlineCTE.scala:
##########
@@ -71,53 +71,64 @@ case class InlineCTE(
* @param plan The plan to collect the CTEs from
* @param cteMap A mutable map that accumulates the CTEs and their reference information by CTE
* ids.
- * @param collectCTERefs A function to collect CTE references so that the caller side can do some
- * bookkeeping work.
+ * @param outerCTEId While collecting the map we use this optional CTE id to identify the
+ * current outer CTE.
*/
private def buildCTEMap(
plan: LogicalPlan,
cteMap: mutable.Map[Long, CTEReferenceInfo],
- collectCTERefs: CTERelationRef => Unit = _ => ()): Unit = {
+ outerCTEId: Option[Long] = None): Unit = {
plan match {
case WithCTE(child, cteDefs) =>
- cteDefs.foreach { cteDef =>
- cteMap(cteDef.id) = CTEReferenceInfo(
- cteDef = cteDef,
- refCount = 0,
- outgoingRefs = mutable.Map.empty.withDefaultValue(0),
- shouldInline = true
- )
- }
- cteDefs.foreach { cteDef =>
- buildCTEMap(cteDef, cteMap, ref => {
- // A CTE relation can references CTE relations defined before it in the same `WithCTE`.
- // Here we update the out-going-ref-count for it, in case this CTE relation is not
- // referenced at all and can be optimized out, and we need to decrease the ref counts
- // for CTE relations that are referenced by it.
- if (cteDefs.exists(_.id == ref.cteId)) {
- cteMap(cteDef.id).increaseOutgoingRefCount(ref.cteId, 1)
- }
- // Similarly, a CTE relation can reference CTE relations defined in the outer `WithCTE`.
- // Here we call the `collectCTERefs` function so that the outer CTE can also update the
- // out-going-ref-count if needed.
- collectCTERefs(ref)
- })
+ val isDuplicated = cteDefs.forall(cteDef => cteMap.contains(cteDef.id))
+ if (isDuplicated) {
+ // If we have seen this `WithCTE` node then it must be self-contained so we can clear
+ // the references from containers to the definitions, and we don't need to process it
+ // again
+
+ cteDefs.foreach { cteDef =>
+ cteMap(cteDef.id).container.foreach(c => cteMap(c).outgoingRefs -= cteDef.id)
Review Comment:
So what we really want to do is: the CTE refs inside the main query of this `WithCTE` should be excluded from container CTE relation's `outgoingRefs`, because this is a self-contained CTE. And because this is a self-contained CTE, the main query can only reference the CTE relations of this `WithCTE`, so we can just exclude all CTE relations of this `WithCTE`, instead of collecting the CTE refs from the main query.
--
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-50722][SQL] Detect self-contained WITH nodes [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #49379:
URL: https://github.com/apache/spark/pull/49379#discussion_r1905386993
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InlineCTE.scala:
##########
@@ -71,53 +71,64 @@ case class InlineCTE(
* @param plan The plan to collect the CTEs from
* @param cteMap A mutable map that accumulates the CTEs and their reference information by CTE
* ids.
- * @param collectCTERefs A function to collect CTE references so that the caller side can do some
- * bookkeeping work.
+ * @param outerCTEId While collecting the map we use this optional CTE id to identify the
+ * current outer CTE.
*/
private def buildCTEMap(
plan: LogicalPlan,
cteMap: mutable.Map[Long, CTEReferenceInfo],
- collectCTERefs: CTERelationRef => Unit = _ => ()): Unit = {
+ outerCTEId: Option[Long] = None): Unit = {
plan match {
case WithCTE(child, cteDefs) =>
- cteDefs.foreach { cteDef =>
- cteMap(cteDef.id) = CTEReferenceInfo(
- cteDef = cteDef,
- refCount = 0,
- outgoingRefs = mutable.Map.empty.withDefaultValue(0),
- shouldInline = true
- )
- }
- cteDefs.foreach { cteDef =>
- buildCTEMap(cteDef, cteMap, ref => {
- // A CTE relation can references CTE relations defined before it in the same `WithCTE`.
- // Here we update the out-going-ref-count for it, in case this CTE relation is not
- // referenced at all and can be optimized out, and we need to decrease the ref counts
- // for CTE relations that are referenced by it.
- if (cteDefs.exists(_.id == ref.cteId)) {
- cteMap(cteDef.id).increaseOutgoingRefCount(ref.cteId, 1)
- }
- // Similarly, a CTE relation can reference CTE relations defined in the outer `WithCTE`.
- // Here we call the `collectCTERefs` function so that the outer CTE can also update the
- // out-going-ref-count if needed.
- collectCTERefs(ref)
- })
+ val isDuplicated = cteDefs.forall(cteDef => cteMap.contains(cteDef.id))
+ if (isDuplicated) {
+ // If we have seen this `WithCTE` node then it must be self-contained so we can clear
+ // the references from containers to the definitions, and we don't need to process it
+ // again
+
+ cteDefs.foreach { cteDef =>
+ cteMap(cteDef.id).container.foreach(c => cteMap(c).outgoingRefs -= cteDef.id)
Review Comment:
oh you are right, when we hit the top-most duplicated `WithCTE`, we stop. Another trick we rely on to avoid issues...
--
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-50722][SQL] Detect self-contained WITH nodes [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #49379:
URL: https://github.com/apache/spark/pull/49379#discussion_r1905230125
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InlineCTE.scala:
##########
@@ -71,53 +71,61 @@ case class InlineCTE(
* @param plan The plan to collect the CTEs from
* @param cteMap A mutable map that accumulates the CTEs and their reference information by CTE
* ids.
- * @param collectCTERefs A function to collect CTE references so that the caller side can do some
- * bookkeeping work.
+ * @param outerCTEId While collecting the map we use this optional CTE id to identify the
+ * current outer CTE.
*/
private def buildCTEMap(
plan: LogicalPlan,
cteMap: mutable.Map[Long, CTEReferenceInfo],
- collectCTERefs: CTERelationRef => Unit = _ => ()): Unit = {
+ outerCTEId: Option[Long] = None): Unit = {
plan match {
case WithCTE(child, cteDefs) =>
+ var visit = true
cteDefs.foreach { cteDef =>
- cteMap(cteDef.id) = CTEReferenceInfo(
- cteDef = cteDef,
- refCount = 0,
- outgoingRefs = mutable.Map.empty.withDefaultValue(0),
- shouldInline = true
- )
+ cteMap.get(cteDef.id) match {
+ // If we have seen this CTE definition then it must be self-contained so we can clear
+ // the references from its container to the definition
+ case Some(refInfo) =>
+ refInfo.container.foreach(c => cteMap(c).outgoingRefs -= cteDef.id)
+
+ // We have processed this definition (and all others in this `WithCTE` node) so we
+ // don't need to do it again.
+ visit = false
+
+ case _ =>
+ cteMap(cteDef.id) = CTEReferenceInfo(
+ cteDef = cteDef,
+ refCount = 0,
+ outgoingRefs = mutable.Map.empty.withDefaultValue(0),
+ shouldInline = true,
+ container = outerCTEId
+ )
+ }
}
- cteDefs.foreach { cteDef =>
- buildCTEMap(cteDef, cteMap, ref => {
- // A CTE relation can references CTE relations defined before it in the same `WithCTE`.
- // Here we update the out-going-ref-count for it, in case this CTE relation is not
- // referenced at all and can be optimized out, and we need to decrease the ref counts
- // for CTE relations that are referenced by it.
- if (cteDefs.exists(_.id == ref.cteId)) {
- cteMap(cteDef.id).increaseOutgoingRefCount(ref.cteId, 1)
- }
- // Similarly, a CTE relation can reference CTE relations defined in the outer `WithCTE`.
- // Here we call the `collectCTERefs` function so that the outer CTE can also update the
- // out-going-ref-count if needed.
- collectCTERefs(ref)
- })
+
+ if (visit) {
+ cteDefs.foreach { cteDef =>
+ buildCTEMap(cteDef, cteMap, Some(cteDef.id))
+ }
+ buildCTEMap(child, cteMap, outerCTEId)
}
- buildCTEMap(child, cteMap, collectCTERefs)
case ref: CTERelationRef =>
cteMap(ref.cteId) = cteMap(ref.cteId).withRefCountIncreased(1)
- collectCTERefs(ref)
+ outerCTEId.foreach { cteId =>
+ cteMap(cteId).increaseOutgoingRefCount(ref.cteId, 1)
Review Comment:
Yeah, good idea. Done in https://github.com/apache/spark/pull/49379/commits/c62c6c4bf2ec4591fee1c5129315a2d9878159a9.
--
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-50722][SQL] Detect self-contained WITH nodes [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on PR #49379:
URL: https://github.com/apache/spark/pull/49379#issuecomment-2576986204
Thanks for the review @cloud-fan and @dongjoon-hyun!
--
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-50722][SQL] Detect self-contained WITH nodes [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #49379:
URL: https://github.com/apache/spark/pull/49379#discussion_r1904821667
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InlineCTE.scala:
##########
@@ -225,12 +233,15 @@ case class InlineCTE(
* from other CTE relations and regular places.
* @param outgoingRefs A mutable map that tracks outgoing reference counts to other CTE relations.
* @param shouldInline If true, this CTE relation should be inlined in the places that reference it.
+ * @param container The container of a CTE definition is another CTE definition in witch the
Review Comment:
```suggestion
* @param container The container of a CTE definition is another CTE definition in which the
```
--
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-50722][SQL] Detect self-contained WITH nodes [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #49379:
URL: https://github.com/apache/spark/pull/49379#discussion_r1905313289
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InlineCTE.scala:
##########
@@ -71,53 +71,64 @@ case class InlineCTE(
* @param plan The plan to collect the CTEs from
* @param cteMap A mutable map that accumulates the CTEs and their reference information by CTE
* ids.
- * @param collectCTERefs A function to collect CTE references so that the caller side can do some
- * bookkeeping work.
+ * @param outerCTEId While collecting the map we use this optional CTE id to identify the
+ * current outer CTE.
*/
private def buildCTEMap(
plan: LogicalPlan,
cteMap: mutable.Map[Long, CTEReferenceInfo],
- collectCTERefs: CTERelationRef => Unit = _ => ()): Unit = {
+ outerCTEId: Option[Long] = None): Unit = {
plan match {
case WithCTE(child, cteDefs) =>
- cteDefs.foreach { cteDef =>
- cteMap(cteDef.id) = CTEReferenceInfo(
- cteDef = cteDef,
- refCount = 0,
- outgoingRefs = mutable.Map.empty.withDefaultValue(0),
- shouldInline = true
- )
- }
- cteDefs.foreach { cteDef =>
- buildCTEMap(cteDef, cteMap, ref => {
- // A CTE relation can references CTE relations defined before it in the same `WithCTE`.
- // Here we update the out-going-ref-count for it, in case this CTE relation is not
- // referenced at all and can be optimized out, and we need to decrease the ref counts
- // for CTE relations that are referenced by it.
- if (cteDefs.exists(_.id == ref.cteId)) {
- cteMap(cteDef.id).increaseOutgoingRefCount(ref.cteId, 1)
- }
- // Similarly, a CTE relation can reference CTE relations defined in the outer `WithCTE`.
- // Here we call the `collectCTERefs` function so that the outer CTE can also update the
- // out-going-ref-count if needed.
- collectCTERefs(ref)
- })
+ val isDuplicated = cteDefs.forall(cteDef => cteMap.contains(cteDef.id))
+ if (isDuplicated) {
+ // If we have seen this `WithCTE` node then it must be self-contained so we can clear
+ // the references from containers to the definitions, and we don't need to process it
+ // again
+
+ cteDefs.foreach { cteDef =>
+ cteMap(cteDef.id).container.foreach(c => cteMap(c).outgoingRefs -= cteDef.id)
Review Comment:
yeah, I mentioned that in the PR description
--
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-50722][SQL] Detect self-contained WITH nodes [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #49379:
URL: https://github.com/apache/spark/pull/49379#discussion_r1904823374
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InlineCTE.scala:
##########
@@ -71,53 +71,61 @@ case class InlineCTE(
* @param plan The plan to collect the CTEs from
* @param cteMap A mutable map that accumulates the CTEs and their reference information by CTE
* ids.
- * @param collectCTERefs A function to collect CTE references so that the caller side can do some
- * bookkeeping work.
+ * @param outerCTEId While collecting the map we use this optional CTE id to identify the
+ * current outer CTE.
*/
private def buildCTEMap(
plan: LogicalPlan,
cteMap: mutable.Map[Long, CTEReferenceInfo],
- collectCTERefs: CTERelationRef => Unit = _ => ()): Unit = {
+ outerCTEId: Option[Long] = None): Unit = {
plan match {
case WithCTE(child, cteDefs) =>
+ var visit = true
cteDefs.foreach { cteDef =>
- cteMap(cteDef.id) = CTEReferenceInfo(
- cteDef = cteDef,
- refCount = 0,
- outgoingRefs = mutable.Map.empty.withDefaultValue(0),
- shouldInline = true
- )
+ cteMap.get(cteDef.id) match {
+ // If we have seen this CTE definition then it must be self-contained so we can clear
+ // the references from its container to the definition
+ case Some(refInfo) =>
+ refInfo.container.foreach(c => cteMap(c).outgoingRefs -= cteDef.id)
+
+ // We have processed this definition (and all others in this `WithCTE` node) so we
+ // don't need to do it again.
+ visit = false
+
+ case _ =>
+ cteMap(cteDef.id) = CTEReferenceInfo(
+ cteDef = cteDef,
+ refCount = 0,
+ outgoingRefs = mutable.Map.empty.withDefaultValue(0),
+ shouldInline = true,
+ container = outerCTEId
+ )
+ }
}
- cteDefs.foreach { cteDef =>
- buildCTEMap(cteDef, cteMap, ref => {
- // A CTE relation can references CTE relations defined before it in the same `WithCTE`.
- // Here we update the out-going-ref-count for it, in case this CTE relation is not
- // referenced at all and can be optimized out, and we need to decrease the ref counts
- // for CTE relations that are referenced by it.
- if (cteDefs.exists(_.id == ref.cteId)) {
- cteMap(cteDef.id).increaseOutgoingRefCount(ref.cteId, 1)
- }
- // Similarly, a CTE relation can reference CTE relations defined in the outer `WithCTE`.
- // Here we call the `collectCTERefs` function so that the outer CTE can also update the
- // out-going-ref-count if needed.
- collectCTERefs(ref)
- })
+
+ if (visit) {
+ cteDefs.foreach { cteDef =>
+ buildCTEMap(cteDef, cteMap, Some(cteDef.id))
+ }
+ buildCTEMap(child, cteMap, outerCTEId)
}
- buildCTEMap(child, cteMap, collectCTERefs)
case ref: CTERelationRef =>
cteMap(ref.cteId) = cteMap(ref.cteId).withRefCountIncreased(1)
- collectCTERefs(ref)
+ outerCTEId.foreach { cteId =>
+ cteMap(cteId).increaseOutgoingRefCount(ref.cteId, 1)
Review Comment:
Can we add some comments here? We should explain that the `outerCTEId` may not really reference this CTE relation, but contains. However, we treat them the same as there is no harm in updating the reference count even if it's a "contains" relationship.
--
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-50722][SQL] Detect self-contained WITH nodes [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on PR #49379:
URL: https://github.com/apache/spark/pull/49379#issuecomment-2575179429
That `Protobuf breaking change detection and Python CodeGen check` failure seems unrelated. Let me update this PR from `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