You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by GitBox <gi...@apache.org> on 2022/04/21 06:12:12 UTC
[GitHub] [spark] beliefer opened a new pull request, #36295: [SPARK-38978][SQL] Support push down limit and offset to JDBC data source V2
beliefer opened a new pull request, #36295:
URL: https://github.com/apache/spark/pull/36295
### What changes were proposed in this pull request?
Currently, Spark supports push down limit to data source.
https://github.com/apache/spark/pull/35975 supported limit work with offset.
In the user's scenario, limit must have the premise of offset. Because limit and offset are more valuable together.
### Why are the changes needed?
1. push down limit and offset is very useful for users scenario.
2. push down limit and offset could improves the performance.
### Does this PR introduce _any_ user-facing change?
'No'.
New feature.
### How was this patch tested?
New tests.
--
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
[GitHub] [spark] cloud-fan commented on a diff in pull request #36295: [SPARK-38978][SQL] DS V2 supports push down OFFSET operator
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #36295:
URL: https://github.com/apache/spark/pull/36295#discussion_r902316747
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala:
##########
@@ -407,8 +407,69 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper wit
case other => (other, false)
}
- def pushDownLimits(plan: LogicalPlan): LogicalPlan = plan.transform {
+ private def pushDownOffset(
+ plan: LogicalPlan,
+ offset: Int): Boolean = plan match {
+ case sHolder: ScanBuilderHolder =>
+ val isPushed = PushDownUtils.pushOffset(sHolder.builder, offset)
+ if (isPushed) {
+ sHolder.pushedOffset = Some(offset)
+ }
+ isPushed
+ case p: Project =>
+ pushDownOffset(p.child, offset)
+ case _ => false
+ }
+
+ def pushDownLimitAndOffset(plan: LogicalPlan): LogicalPlan = plan.transform {
+ case offset @ LimitAndOffset(limit, offsetValue, child) =>
+ val (newChild, canRemoveLimit) = pushDownLimit(child, limit)
+ if (canRemoveLimit) {
+ // If we can remove limit, it indicates data source only have one partition.
+ // For `dataset.limit(m).offset(n)`, try to push down `offset(n).limit(m - n)`.
+ // For example, `dataset.limit(5).offset(3)`, we can push down `offset(3).limit(2)`.
+ val isPushed = pushDownOffset(newChild, offsetValue)
+ if (isPushed) {
+ newChild
+ } else {
+ // For `dataset.limit(m).offset(n)`, only push down `limit(m)`.
+ // Spark still do `offset(n)`.
+ offset.withNewChildren(Seq(newChild))
+ }
+ } else {
+ // If we can't push down limit and offset, return `Offset`.
Review Comment:
why? limit can be pushed, right? We should return `Offset(offsetValue, newChild)` here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] cloud-fan commented on a diff in pull request #36295: [SPARK-38978][SQL] DS V2 supports push down OFFSET operator
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #36295:
URL: https://github.com/apache/spark/pull/36295#discussion_r903815058
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala:
##########
@@ -407,7 +407,66 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper wit
case other => (other, false)
}
- def pushDownLimits(plan: LogicalPlan): LogicalPlan = plan.transform {
+ private def pushDownOffset(
+ plan: LogicalPlan,
+ offset: Int): Boolean = plan match {
+ case sHolder: ScanBuilderHolder =>
+ val isPushed = PushDownUtils.pushOffset(sHolder.builder, offset)
+ if (isPushed) {
+ sHolder.pushedOffset = Some(offset)
+ }
+ isPushed
+ case p: Project =>
+ pushDownOffset(p.child, offset)
+ case _ => false
+ }
+
+ def pushDownLimitAndOffset(plan: LogicalPlan): LogicalPlan = plan.transform {
+ case offset @ LimitAndOffset(limit, offsetValue, child) =>
+ val (newChild, canRemoveLimit) = pushDownLimit(child, limit)
+ if (canRemoveLimit) {
+ // If we can remove limit, it indicates data source only have one partition.
+ // For `dataset.limit(m).offset(n)`, try to push down `limit(m).offset(n)`.
+ // For example, `dataset.limit(5).offset(3)`, we can push down `limit(5).offset(3)`.
+ val isPushed = pushDownOffset(newChild, offsetValue)
+ if (isPushed) {
+ newChild
+ } else {
+ // For `dataset.limit(m).offset(n)`, only push down `limit(m)`.
+ // Keep the OFFSET operator if we failed to push down OFFSET to the data source.
+ offset.withNewChildren(Seq(newChild))
+ }
+ } else {
+ // Keep the OFFSET operator if we can't remove LIMIT operator.
+ offset
+ }
+ case globalLimit @ OffsetAndLimit(offset, limit, child) =>
+ val (newChild, canRemoveLimit) = pushDownLimit(child, limit + offset)
+ if (canRemoveLimit) {
+ // If we can remove limit, it indicates data source only have one partition.
+ // For `dataset.offset(n).limit(m)`, try to push down `limit(m + n).offset(n)`.
+ // For example, `dataset.offset(3).limit(5)`, we can push down `limit(8).offset(3)`.
+ val isPushed = pushDownOffset(newChild, offset)
+ if (isPushed) {
+ newChild
+ } else {
+ // For `dataset.offset(n).limit(m)`, try to push down `limit(m + n)`.
+ // Spark still do `offset(n)`.
+ // For example, `dataset.offset(3).limit(5)`, we can push down `limit(8)`.
+ // Spark still do `offset(3)`.
+ Offset(Literal(offset), newChild)
+ }
+ } else {
+ // For `dataset.offset(n).limit(m)`, try to push down `offset(n)`.
+ // Spark still do `limit(m)`.
+ val isPushed = pushDownOffset(child, offset)
+ if (isPushed) {
+ globalLimit.withNewChildren(Seq(child))
Review Comment:
is this correct? seems we may lose the local limit operator?
--
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
[GitHub] [spark] beliefer commented on a diff in pull request #36295: [SPARK-38978][SQL] DS V2 supports push down OFFSET operator
Posted by GitBox <gi...@apache.org>.
beliefer commented on code in PR #36295:
URL: https://github.com/apache/spark/pull/36295#discussion_r904460364
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala:
##########
@@ -407,7 +407,66 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper wit
case other => (other, false)
}
- def pushDownLimits(plan: LogicalPlan): LogicalPlan = plan.transform {
+ private def pushDownOffset(
+ plan: LogicalPlan,
+ offset: Int): Boolean = plan match {
+ case sHolder: ScanBuilderHolder =>
+ val isPushed = PushDownUtils.pushOffset(sHolder.builder, offset)
+ if (isPushed) {
+ sHolder.pushedOffset = Some(offset)
+ }
+ isPushed
+ case p: Project =>
+ pushDownOffset(p.child, offset)
+ case _ => false
+ }
+
+ def pushDownLimitAndOffset(plan: LogicalPlan): LogicalPlan = plan.transform {
+ case offset @ LimitAndOffset(limit, offsetValue, child) =>
+ val (newChild, canRemoveLimit) = pushDownLimit(child, limit)
+ if (canRemoveLimit) {
+ // If we can remove limit, it indicates data source only have one partition.
Review Comment:
OK
--
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
[GitHub] [spark] beliefer commented on a diff in pull request #36295: [SPARK-38978][SQL] DS V2 supports push down OFFSET operator
Posted by GitBox <gi...@apache.org>.
beliefer commented on code in PR #36295:
URL: https://github.com/apache/spark/pull/36295#discussion_r904481912
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala:
##########
@@ -407,7 +407,66 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper wit
case other => (other, false)
}
- def pushDownLimits(plan: LogicalPlan): LogicalPlan = plan.transform {
+ private def pushDownOffset(
+ plan: LogicalPlan,
+ offset: Int): Boolean = plan match {
+ case sHolder: ScanBuilderHolder =>
+ val isPushed = PushDownUtils.pushOffset(sHolder.builder, offset)
+ if (isPushed) {
+ sHolder.pushedOffset = Some(offset)
+ }
+ isPushed
+ case p: Project =>
+ pushDownOffset(p.child, offset)
+ case _ => false
+ }
+
+ def pushDownLimitAndOffset(plan: LogicalPlan): LogicalPlan = plan.transform {
+ case offset @ LimitAndOffset(limit, offsetValue, child) =>
+ val (newChild, canRemoveLimit) = pushDownLimit(child, limit)
+ if (canRemoveLimit) {
+ // If we can remove limit, it indicates data source only have one partition.
+ // For `dataset.limit(m).offset(n)`, try to push down `limit(m).offset(n)`.
+ // For example, `dataset.limit(5).offset(3)`, we can push down `limit(5).offset(3)`.
+ val isPushed = pushDownOffset(newChild, offsetValue)
+ if (isPushed) {
+ newChild
+ } else {
+ // For `dataset.limit(m).offset(n)`, only push down `limit(m)`.
+ // Keep the OFFSET operator if we failed to push down OFFSET to the data source.
+ offset.withNewChildren(Seq(newChild))
+ }
+ } else {
+ // Keep the OFFSET operator if we can't remove LIMIT operator.
+ offset
+ }
+ case globalLimit @ OffsetAndLimit(offset, limit, child) =>
+ val (newChild, canRemoveLimit) = pushDownLimit(child, limit + offset)
Review Comment:
OK
--
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
[GitHub] [spark] beliefer commented on pull request #36295: [SPARK-38978][SQL] DS V2 supports push down OFFSET operator
Posted by GitBox <gi...@apache.org>.
beliefer commented on PR #36295:
URL: https://github.com/apache/spark/pull/36295#issuecomment-1165406326
@cloud-fan Thank you for the hard 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
[GitHub] [spark] beliefer commented on a diff in pull request #36295: [SPARK-38978][SQL] Support push down OFFSET to JDBC data source V2
Posted by GitBox <gi...@apache.org>.
beliefer commented on code in PR #36295:
URL: https://github.com/apache/spark/pull/36295#discussion_r881269803
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/DataSourceScanExec.scala:
##########
@@ -141,15 +141,23 @@ case class RowDataSourceScanExec(
handledFilters
}
- val topNOrLimitInfo =
+ val limitOrOffsetInfo =
if (pushedDownOperators.limit.isDefined && pushedDownOperators.sortValues.nonEmpty) {
- val pushedTopN =
- s"ORDER BY ${seqToString(pushedDownOperators.sortValues.map(_.describe()))}" +
- s" LIMIT ${pushedDownOperators.limit.get}"
- Some("PushedTopN" -> pushedTopN)
- } else {
- pushedDownOperators.limit.map(value => "PushedLimit" -> s"LIMIT $value")
- }
+ if (pushedDownOperators.offset.isDefined) {
Review Comment:
OK
--
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
[GitHub] [spark] cloud-fan commented on a diff in pull request #36295: [SPARK-38978][SQL] Support push down OFFSET to JDBC data source V2
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #36295:
URL: https://github.com/apache/spark/pull/36295#discussion_r874589600
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala:
##########
@@ -419,6 +420,72 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper wit
}
}
+ private def pushDownOffset(
Review Comment:
let's have a single `pushDownLimitAndOffset`. It should be simpler to the planner rule, and we need to match different kind of query plans: LIMIT only, OFFSET only, LIMIT + OFFSET, OFFSET + LIMIT, SORT + LIMT, SORT + LIMIT + OFFSET, etc.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala:
##########
@@ -419,6 +420,72 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper wit
}
}
+ private def pushDownOffset(
Review Comment:
let's have a single `pushDownLimitAndOffset`. It should be similar to the planner rule, and we need to match different kind of query plans: LIMIT only, OFFSET only, LIMIT + OFFSET, OFFSET + LIMIT, SORT + LIMT, SORT + LIMIT + OFFSET, etc.
--
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
[GitHub] [spark] cloud-fan commented on a diff in pull request #36295: [SPARK-38978][SQL] Support push down OFFSET to JDBC data source V2
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #36295:
URL: https://github.com/apache/spark/pull/36295#discussion_r882687268
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala:
##########
@@ -407,8 +407,65 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper wit
case other => (other, false)
}
- def pushDownLimits(plan: LogicalPlan): LogicalPlan = plan.transform {
+ private def pushDownOffset(
+ plan: LogicalPlan,
+ offset: Int): Boolean = plan match {
+ case sHolder: ScanBuilderHolder =>
+ val isPushed = PushDownUtils.pushOffset(sHolder.builder, offset)
+ if (isPushed) {
+ sHolder.pushedOffset = Some(offset)
+ }
+ isPushed
+ case p: Project =>
+ pushDownOffset(p.child, offset)
+ case _ => false
+ }
+
+ def pushDownLimitAndOffset(plan: LogicalPlan): LogicalPlan = plan.transform {
+ case offset @ LimitAndOffset(limit, offsetValue, child) =>
+ val (newChild, canRemoveLimit) = pushDownLimit(child, limit)
+ if (canRemoveLimit) {
+ // If we can remove limit, it indicates data source only have one partition.
+ // For `dataset.limit(m).offset(n)`, try to push down `LIMIT (m - n) OFFSET n`.
+ // For example, `dataset.limit(5).offset(3)`, we can push down `LIMIT 2 OFFSET 3`.
+ val isPushed = pushDownOffset(newChild, offsetValue)
+ if (isPushed) {
+ newChild
+ } else {
+ // For `dataset.limit(m).offset(n)`, only `LIMIT m` be pushed. Spark will do `OFFSET n`.
+ offset
+ }
+ } else {
+ // If we can't push down limit and offset, return `Offset`.
+ offset
+ }
+ case globalLimit @ OffsetAndLimit(offset, limitValue, child) =>
+ val isPushed = pushDownOffset(child, offset)
Review Comment:
no, we can't push OFFSET first. We need to follow the API doc and push LIMIT first. Although OFFSET appears first in the query plan, we can still push down LIMIT by updating the limit value to `limit + offset`
--
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
[GitHub] [spark] beliefer commented on a diff in pull request #36295: [SPARK-38978][SQL] Support push down OFFSET to JDBC data source V2
Posted by GitBox <gi...@apache.org>.
beliefer commented on code in PR #36295:
URL: https://github.com/apache/spark/pull/36295#discussion_r873551016
##########
sql/catalyst/src/main/java/org/apache/spark/sql/connector/read/ScanBuilder.java:
##########
@@ -23,7 +23,7 @@
* An interface for building the {@link Scan}. Implementations can mixin SupportsPushDownXYZ
* interfaces to do operator push down, and keep the operator push down result in the returned
* {@link Scan}. When pushing down operators, the push down order is:
- * sample -> filter -> aggregate -> limit -> column pruning.
+ * sample -> filter -> aggregate -> offset -> limit or top N -> column pruning.
Review Comment:
OK
--
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
[GitHub] [spark] cloud-fan commented on a diff in pull request #36295: [SPARK-38978][SQL] DS V2 supports push down OFFSET operator
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #36295:
URL: https://github.com/apache/spark/pull/36295#discussion_r903797786
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala:
##########
@@ -407,7 +407,66 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper wit
case other => (other, false)
}
- def pushDownLimits(plan: LogicalPlan): LogicalPlan = plan.transform {
+ private def pushDownOffset(
+ plan: LogicalPlan,
+ offset: Int): Boolean = plan match {
+ case sHolder: ScanBuilderHolder =>
+ val isPushed = PushDownUtils.pushOffset(sHolder.builder, offset)
+ if (isPushed) {
+ sHolder.pushedOffset = Some(offset)
+ }
+ isPushed
+ case p: Project =>
+ pushDownOffset(p.child, offset)
+ case _ => false
+ }
+
+ def pushDownLimitAndOffset(plan: LogicalPlan): LogicalPlan = plan.transform {
+ case offset @ LimitAndOffset(limit, offsetValue, child) =>
+ val (newChild, canRemoveLimit) = pushDownLimit(child, limit)
+ if (canRemoveLimit) {
+ // If we can remove limit, it indicates data source only have one partition.
+ // For `dataset.limit(m).offset(n)`, try to push down `limit(m).offset(n)`.
+ // For example, `dataset.limit(5).offset(3)`, we can push down `limit(5).offset(3)`.
+ val isPushed = pushDownOffset(newChild, offsetValue)
+ if (isPushed) {
+ newChild
+ } else {
+ // For `dataset.limit(m).offset(n)`, only push down `limit(m)`.
+ // Keep the OFFSET operator if we failed to push down OFFSET to the data source.
+ offset.withNewChildren(Seq(newChild))
+ }
+ } else {
+ // Keep the OFFSET operator if we can't remove LIMIT operator.
+ offset
+ }
+ case globalLimit @ OffsetAndLimit(offset, limit, child) =>
+ val (newChild, canRemoveLimit) = pushDownLimit(child, limit + offset)
+ if (canRemoveLimit) {
+ // If we can remove limit, it indicates data source only have one partition.
Review Comment:
ditto:
```
// Try to push down OFFSET only if the LIMIT operator has been pushed and can be removed.
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] cloud-fan commented on pull request #36295: [SPARK-38978][SQL] DS V2 supports push down OFFSET operator
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on PR #36295:
URL: https://github.com/apache/spark/pull/36295#issuecomment-1165391078
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
[GitHub] [spark] cloud-fan commented on a diff in pull request #36295: [SPARK-38978][SQL] DS V2 supports push down OFFSET operator
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #36295:
URL: https://github.com/apache/spark/pull/36295#discussion_r903813637
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala:
##########
@@ -407,7 +407,66 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper wit
case other => (other, false)
}
- def pushDownLimits(plan: LogicalPlan): LogicalPlan = plan.transform {
+ private def pushDownOffset(
+ plan: LogicalPlan,
+ offset: Int): Boolean = plan match {
+ case sHolder: ScanBuilderHolder =>
+ val isPushed = PushDownUtils.pushOffset(sHolder.builder, offset)
+ if (isPushed) {
+ sHolder.pushedOffset = Some(offset)
+ }
+ isPushed
+ case p: Project =>
+ pushDownOffset(p.child, offset)
+ case _ => false
+ }
+
+ def pushDownLimitAndOffset(plan: LogicalPlan): LogicalPlan = plan.transform {
+ case offset @ LimitAndOffset(limit, offsetValue, child) =>
+ val (newChild, canRemoveLimit) = pushDownLimit(child, limit)
+ if (canRemoveLimit) {
+ // If we can remove limit, it indicates data source only have one partition.
+ // For `dataset.limit(m).offset(n)`, try to push down `limit(m).offset(n)`.
+ // For example, `dataset.limit(5).offset(3)`, we can push down `limit(5).offset(3)`.
+ val isPushed = pushDownOffset(newChild, offsetValue)
+ if (isPushed) {
+ newChild
+ } else {
+ // For `dataset.limit(m).offset(n)`, only push down `limit(m)`.
+ // Keep the OFFSET operator if we failed to push down OFFSET to the data source.
+ offset.withNewChildren(Seq(newChild))
+ }
+ } else {
+ // Keep the OFFSET operator if we can't remove LIMIT operator.
+ offset
+ }
+ case globalLimit @ OffsetAndLimit(offset, limit, child) =>
+ val (newChild, canRemoveLimit) = pushDownLimit(child, limit + offset)
+ if (canRemoveLimit) {
+ // If we can remove limit, it indicates data source only have one partition.
+ // For `dataset.offset(n).limit(m)`, try to push down `limit(m + n).offset(n)`.
+ // For example, `dataset.offset(3).limit(5)`, we can push down `limit(8).offset(3)`.
+ val isPushed = pushDownOffset(newChild, offset)
+ if (isPushed) {
+ newChild
+ } else {
+ // For `dataset.offset(n).limit(m)`, try to push down `limit(m + n)`.
+ // Spark still do `offset(n)`.
+ // For example, `dataset.offset(3).limit(5)`, we can push down `limit(8)`.
+ // Spark still do `offset(3)`.
+ Offset(Literal(offset), newChild)
+ }
+ } else {
+ // For `dataset.offset(n).limit(m)`, try to push down `offset(n)`.
Review Comment:
```
// For `df.offset(n).limit(m)`, since we can't push down `limit(m+n)`, try to push down `offset(n)` here.
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] cloud-fan commented on a diff in pull request #36295: [SPARK-38978][SQL] Support push down limit and offset to JDBC data source V2
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #36295:
URL: https://github.com/apache/spark/pull/36295#discussion_r855351511
##########
sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala:
##########
@@ -2044,6 +2044,16 @@ class Dataset[T] private[sql](
Limit(Literal(n), logicalPlan)
}
+ /**
+ * Returns a new Dataset by skipping the first `m` rows and then taking the followed `n` rows.
+ *
+ * @group typedrel
+ * @since 3.4.0
+ */
+ def limitAndOffset(n: Int, m: Int): Dataset[T] = withTypedPlan {
Review Comment:
This does not match the SQL API. Shall we just have `def offset(...)`?
--
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
[GitHub] [spark] beliefer commented on pull request #36295: [SPARK-38978][SQL] Support push down OFFSET to JDBC data source V2
Posted by GitBox <gi...@apache.org>.
beliefer commented on PR #36295:
URL: https://github.com/apache/spark/pull/36295#issuecomment-1146967033
> Can we have some kind of performance numbers for "push down OFFSET could improves the performance."?
For most JDBC data source, push down OFFSET could improves the performance.
There are three reasons:
1. JDBC data source knows the data distribution.
2. JDBC data source constructs index for data.
3. JDBC data source may have the query cache.
4. Some JDBC data source running in native.
5. push down OFFSET could reduce the I/O, then may reduce the shuffle size.
--
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
[GitHub] [spark] cloud-fan commented on a diff in pull request #36295: [SPARK-38978][SQL] Support push down OFFSET to JDBC data source V2
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #36295:
URL: https://github.com/apache/spark/pull/36295#discussion_r874587041
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala:
##########
@@ -44,6 +44,7 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper wit
pushDownFilters,
pushDownAggregates,
pushDownLimits,
+ pushDownOffsets,
Review Comment:
according to the doc, OFFSET should be pushed after LIMIT now.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] cloud-fan commented on a diff in pull request #36295: [SPARK-38978][SQL] Support push down OFFSET to JDBC data source V2
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #36295:
URL: https://github.com/apache/spark/pull/36295#discussion_r874584082
##########
sql/catalyst/src/main/java/org/apache/spark/sql/connector/read/SupportsPushDownLimit.java:
##########
@@ -21,8 +21,8 @@
/**
* A mix-in interface for {@link ScanBuilder}. Data sources can implement this interface to
- * push down LIMIT. Please note that the combination of LIMIT with other operations
- * such as AGGREGATE, GROUP BY, SORT BY, CLUSTER BY, DISTRIBUTE BY, etc. is NOT pushed down.
+ * push down LIMIT. We can push down LIMIT with many other operations if they follow the
+ * operator order we defined in {@link ScanBuilder}'s class doc.
Review Comment:
let's open a backport PR for 3.3 to fix the classdoc later.
--
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
[GitHub] [spark] cloud-fan commented on a diff in pull request #36295: [SPARK-38978][SQL] Support push down OFFSET to JDBC data source V2
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #36295:
URL: https://github.com/apache/spark/pull/36295#discussion_r882304824
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala:
##########
@@ -44,6 +44,7 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper wit
pushDownFilters,
pushDownAggregates,
pushDownLimits,
Review Comment:
LIMIT should be pushed by `pushDownLimitAndOffset` 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
[GitHub] [spark] cloud-fan commented on a diff in pull request #36295: [SPARK-38978][SQL] DS V2 supports push down OFFSET operator
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #36295:
URL: https://github.com/apache/spark/pull/36295#discussion_r902313439
##########
sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCV2Suite.scala:
##########
@@ -206,6 +207,355 @@ class JDBCV2Suite extends QueryTest with SharedSparkSession with ExplainSuiteHel
checkAnswer(df5, Seq(Row(10000.00, 1000.0, "amy")))
}
+ private def checkOffsetRemoved(df: DataFrame, removed: Boolean = true): Unit = {
+ val offsets = df.queryExecution.optimizedPlan.collect {
+ case offset: Offset => offset
+ }
+ if (removed) {
+ assert(offsets.isEmpty)
+ } else {
+ assert(offsets.nonEmpty)
+ }
+ }
+
+ test("simple scan with OFFSET") {
+ val df1 = spark.read
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .offset(1)
+ checkOffsetRemoved(df1)
+ checkPushedInfo(df1,
+ "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], PushedOffset: OFFSET 1,")
+ checkAnswer(df1, Seq(Row(1, "cathy", 9000.00, 1200.0, false)))
+
+ val df2 = spark.read
+ .option("pushDownOffset", "false")
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .offset(1)
+ checkOffsetRemoved(df2, false)
+ checkPushedInfo(df2,
+ "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], ReadSchema:")
+ checkAnswer(df2, Seq(Row(1, "cathy", 9000.00, 1200.0, false)))
+
+ val df3 = spark.read
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .sort($"salary")
+ .offset(1)
+ checkOffsetRemoved(df3, false)
+ checkPushedInfo(df3,
+ "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], ReadSchema:")
+ checkAnswer(df3, Seq(Row(1, "amy", 10000.00, 1000.0, true)))
+
+ val df4 = spark.read
+ .option("partitionColumn", "dept")
+ .option("lowerBound", "0")
+ .option("upperBound", "2")
+ .option("numPartitions", "2")
+ .table("h2.test.employee")
+ .filter($"dept" > 1)
+ .offset(1)
+ checkOffsetRemoved(df4, false)
+ checkPushedInfo(df4, "PushedFilters: [DEPT IS NOT NULL, DEPT > 1], ReadSchema:")
+ checkAnswer(df4, Seq(Row(2, "david", 10000, 1300, true), Row(6, "jen", 12000, 1200, true)))
+
+ val df5 = spark.read
+ .table("h2.test.employee")
+ .groupBy("DEPT").sum("SALARY")
+ .offset(1)
+ checkOffsetRemoved(df5, false)
+ checkPushedInfo(df5,
+ "PushedAggregates: [SUM(SALARY)], PushedFilters: [], PushedGroupByExpressions: [DEPT], ")
+ checkAnswer(df5, Seq(Row(2, 22000.00), Row(6, 12000.00)))
+
+ val name = udf { (x: String) => x.matches("cat|dav|amy") }
+ val sub = udf { (x: String) => x.substring(0, 3) }
+ val df6 = spark.read
+ .table("h2.test.employee")
+ .select($"SALARY", $"BONUS", sub($"NAME").as("shortName"))
+ .filter(name($"shortName"))
+ .offset(1)
+ checkOffsetRemoved(df6, false)
+ // OFFSET is pushed down only if all the filters are pushed down
+ checkPushedInfo(df6, "PushedFilters: [], ")
+ checkAnswer(df6, Seq(Row(10000.00, 1300.0, "dav"), Row(9000.00, 1200.0, "cat")))
+ }
+
+ test("simple scan with LIMIT and OFFSET") {
+ val df1 = spark.read
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .limit(2)
+ .offset(1)
+ checkLimitRemoved(df1)
+ checkOffsetRemoved(df1)
+ checkPushedInfo(df1,
+ "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], PushedLimit: LIMIT 2, PushedOffset: OFFSET 1,")
+ checkAnswer(df1, Seq(Row(1, "cathy", 9000.00, 1200.0, false)))
+
+ val df2 = spark.read
+ .option("pushDownLimit", "false")
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .limit(2)
+ .offset(1)
+ checkLimitRemoved(df2, false)
+ checkOffsetRemoved(df2, false)
+ checkPushedInfo(df2,
+ "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], ReadSchema:")
+ checkAnswer(df2, Seq(Row(1, "cathy", 9000.00, 1200.0, false)))
+
+ val df3 = spark.read
+ .option("pushDownOffset", "false")
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .limit(2)
+ .offset(1)
+ checkLimitRemoved(df3)
+ checkOffsetRemoved(df3, false)
+ checkPushedInfo(df3,
+ "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], PushedLimit: LIMIT 2, ReadSchema:")
+ checkAnswer(df3, Seq(Row(1, "cathy", 9000.00, 1200.0, false)))
+
+ val df4 = spark.read
+ .option("pushDownLimit", "false")
+ .option("pushDownOffset", "false")
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .limit(2)
+ .offset(1)
+ checkLimitRemoved(df4, false)
+ checkOffsetRemoved(df4, false)
+ checkPushedInfo(df4,
+ "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], ReadSchema:")
+ checkAnswer(df4, Seq(Row(1, "cathy", 9000.00, 1200.0, false)))
+
+ val df5 = spark.read
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .sort($"salary")
+ .limit(2)
+ .offset(1)
+ checkLimitRemoved(df5)
+ checkOffsetRemoved(df5)
+ checkPushedInfo(df5, "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], " +
+ "PushedOffset: OFFSET 1, PushedTopN: ORDER BY [SALARY ASC NULLS FIRST] LIMIT 2, ReadSchema:")
+ checkAnswer(df5, Seq(Row(1, "amy", 10000.00, 1000.0, true)))
+
+ val df6 = spark.read
+ .option("pushDownLimit", "false")
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .sort($"salary")
+ .limit(2)
+ .offset(1)
+ checkLimitRemoved(df6, false)
+ checkOffsetRemoved(df6, false)
+ checkPushedInfo(df6, "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], ReadSchema:")
+ checkAnswer(df6, Seq(Row(1, "amy", 10000.00, 1000.0, true)))
+
+ val df7 = spark.read
+ .option("pushDownOffset", "false")
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .sort($"salary")
+ .limit(2)
+ .offset(1)
+ checkLimitRemoved(df7)
+ checkOffsetRemoved(df7, false)
+ checkPushedInfo(df7, "PushedFilters: [DEPT IS NOT NULL, DEPT = 1]," +
+ " PushedTopN: ORDER BY [SALARY ASC NULLS FIRST] LIMIT 2, ReadSchema:")
+ checkAnswer(df7, Seq(Row(1, "amy", 10000.00, 1000.0, true)))
+
+ val df8 = spark.read
+ .option("pushDownLimit", "false")
+ .option("pushDownOffset", "false")
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .sort($"salary")
+ .limit(2)
+ .offset(1)
+ checkLimitRemoved(df8, false)
+ checkOffsetRemoved(df8, false)
+ checkPushedInfo(df8, "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], ReadSchema:")
+ checkAnswer(df8, Seq(Row(1, "amy", 10000.00, 1000.0, true)))
+
+ val df9 = spark.read
+ .option("partitionColumn", "dept")
+ .option("lowerBound", "0")
+ .option("upperBound", "2")
+ .option("numPartitions", "2")
+ .table("h2.test.employee")
+ .filter($"dept" > 1)
+ .limit(2)
+ .offset(1)
+ checkLimitRemoved(df9, false)
+ checkOffsetRemoved(df9, false)
+ checkPushedInfo(df9,
+ "PushedFilters: [DEPT IS NOT NULL, DEPT > 1], PushedLimit: LIMIT 2, ReadSchema:")
+ checkAnswer(df9, Seq(Row(2, "david", 10000.00, 1300.0, true)))
+
+ val df10 = spark.read
+ .table("h2.test.employee")
+ .groupBy("DEPT").sum("SALARY")
+ .limit(2)
+ .offset(1)
+ checkLimitRemoved(df10, false)
+ checkOffsetRemoved(df10, false)
+ checkPushedInfo(df10,
+ "PushedAggregates: [SUM(SALARY)], PushedFilters: [], PushedGroupByExpressions: [DEPT], ")
+ checkAnswer(df10, Seq(Row(2, 22000.00)))
+
+ val name = udf { (x: String) => x.matches("cat|dav|amy") }
+ val sub = udf { (x: String) => x.substring(0, 3) }
+ val df11 = spark.read
+ .table("h2.test.employee")
+ .select($"SALARY", $"BONUS", sub($"NAME").as("shortName"))
+ .filter(name($"shortName"))
+ .limit(2)
+ .offset(1)
+ checkLimitRemoved(df11, false)
+ checkOffsetRemoved(df11, false)
+ checkPushedInfo(df11, "PushedFilters: [], ")
+ checkAnswer(df11, Seq(Row(9000.00, 1200.0, "cat")))
+ }
+
+ test("simple scan with OFFSET and LIMIT") {
+ val df1 = spark.read
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .offset(1)
+ .limit(1)
+ checkLimitRemoved(df1)
+ checkOffsetRemoved(df1)
+ checkPushedInfo(df1,
+ "[DEPT IS NOT NULL, DEPT = 1], PushedLimit: LIMIT 2, PushedOffset: OFFSET 1,")
+ checkAnswer(df1, Seq(Row(1, "cathy", 9000.00, 1200.0, false)))
+
+ val df2 = spark.read
+ .option("pushDownOffset", "false")
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .offset(1)
+ .limit(1)
+ checkLimitRemoved(df2, false)
+ checkOffsetRemoved(df2, false)
+ checkPushedInfo(df2,
+ "[DEPT IS NOT NULL, DEPT = 1], PushedLimit: LIMIT 2, ReadSchema:")
+ checkAnswer(df2, Seq(Row(1, "cathy", 9000.00, 1200.0, false)))
+
+ val df3 = spark.read
+ .option("pushDownLimit", "false")
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .offset(1)
+ .limit(1)
+ checkLimitRemoved(df3, false)
+ checkOffsetRemoved(df3)
Review Comment:
offset always goes after limit in pushdown, how can we support this case?
--
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
[GitHub] [spark] cloud-fan commented on a diff in pull request #36295: [SPARK-38978][SQL] DS V2 supports push down OFFSET operator
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #36295:
URL: https://github.com/apache/spark/pull/36295#discussion_r902320980
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala:
##########
@@ -407,8 +407,69 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper wit
case other => (other, false)
}
- def pushDownLimits(plan: LogicalPlan): LogicalPlan = plan.transform {
+ private def pushDownOffset(
+ plan: LogicalPlan,
+ offset: Int): Boolean = plan match {
+ case sHolder: ScanBuilderHolder =>
+ val isPushed = PushDownUtils.pushOffset(sHolder.builder, offset)
+ if (isPushed) {
+ sHolder.pushedOffset = Some(offset)
+ }
+ isPushed
+ case p: Project =>
+ pushDownOffset(p.child, offset)
+ case _ => false
+ }
+
+ def pushDownLimitAndOffset(plan: LogicalPlan): LogicalPlan = plan.transform {
+ case offset @ LimitAndOffset(limit, offsetValue, child) =>
+ val (newChild, canRemoveLimit) = pushDownLimit(child, limit)
+ if (canRemoveLimit) {
+ // If we can remove limit, it indicates data source only have one partition.
+ // For `dataset.limit(m).offset(n)`, try to push down `offset(n).limit(m - n)`.
+ // For example, `dataset.limit(5).offset(3)`, we can push down `offset(3).limit(2)`.
+ val isPushed = pushDownOffset(newChild, offsetValue)
+ if (isPushed) {
+ newChild
+ } else {
+ // For `dataset.limit(m).offset(n)`, only push down `limit(m)`.
+ // Spark still do `offset(n)`.
Review Comment:
The comment doesn't match the code
```
// Keep the OFFSET operator if we failed to push down OFFSET to the data source.
```
--
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
[GitHub] [spark] cloud-fan commented on a diff in pull request #36295: [SPARK-38978][SQL] DS V2 supports push down OFFSET operator
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #36295:
URL: https://github.com/apache/spark/pull/36295#discussion_r902324979
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala:
##########
@@ -417,6 +478,14 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper wit
globalLimit.child.asInstanceOf[LocalLimit].withNewChildren(Seq(newChild))
globalLimit.withNewChildren(Seq(newLocalLimit))
}
+ case offset @ Offset(IntegerLiteral(n), child) =>
+ // For `dataset.offset(n)`, try to push down `offset(n)`.
Review Comment:
ditto
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] cloud-fan commented on a diff in pull request #36295: [SPARK-38978][SQL] DS V2 supports push down OFFSET operator
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #36295:
URL: https://github.com/apache/spark/pull/36295#discussion_r902304551
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/jdbc/JDBCScanBuilder.scala:
##########
@@ -139,6 +142,25 @@ case class JDBCScanBuilder(
false
}
+ override def pushOffset(offset: Int): Boolean = {
+ if (jdbcOptions.pushDownOffset && !isPartiallyPushed) {
+ // We pushing down offset when data source have only one partition.
+ // There are some push down path.
Review Comment:
We can make the comment much simpler:
```
Spark pushes down LIMIT first, then OFFSET. In SQL statements, OFFSET is applied before
LIMIT. Here we need to adjust the LIMIT value to match SQL statements.
```
--
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
[GitHub] [spark] cloud-fan commented on a diff in pull request #36295: [SPARK-38978][SQL] DS V2 supports push down OFFSET operator
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #36295:
URL: https://github.com/apache/spark/pull/36295#discussion_r904486458
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala:
##########
@@ -407,7 +407,67 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper wit
case other => (other, false)
}
- def pushDownLimits(plan: LogicalPlan): LogicalPlan = plan.transform {
+ private def pushDownOffset(
+ plan: LogicalPlan,
+ offset: Int): Boolean = plan match {
+ case sHolder: ScanBuilderHolder =>
+ val isPushed = PushDownUtils.pushOffset(sHolder.builder, offset)
+ if (isPushed) {
+ sHolder.pushedOffset = Some(offset)
+ }
+ isPushed
+ case Project(projectList, child) if projectList.forall(_.deterministic) =>
+ pushDownOffset(child, offset)
+ case _ => false
+ }
+
+ def pushDownLimitAndOffset(plan: LogicalPlan): LogicalPlan = plan.transform {
+ case offset @ LimitAndOffset(limit, offsetValue, child) =>
+ val (newChild, canRemoveLimit) = pushDownLimit(child, limit)
+ if (canRemoveLimit) {
+ // Try to push down OFFSET only if the LIMIT operator has been pushed and can be removed.
+ // For `df.limit(m).offset(n)`, try to push down `limit(m).offset(n)`.
+ // For example, `df.limit(5).offset(3)`, we can push down `limit(5).offset(3)`.
+ val isPushed = pushDownOffset(newChild, offsetValue)
+ if (isPushed) {
+ newChild
+ } else {
+ // For `df.limit(m).offset(n)`, only push down `limit(m)`.
Review Comment:
```suggestion
```
--
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
[GitHub] [spark] cloud-fan commented on a diff in pull request #36295: [SPARK-38978][SQL] DS V2 supports push down OFFSET operator
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #36295:
URL: https://github.com/apache/spark/pull/36295#discussion_r904486283
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala:
##########
@@ -407,7 +407,67 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper wit
case other => (other, false)
}
- def pushDownLimits(plan: LogicalPlan): LogicalPlan = plan.transform {
+ private def pushDownOffset(
+ plan: LogicalPlan,
+ offset: Int): Boolean = plan match {
+ case sHolder: ScanBuilderHolder =>
+ val isPushed = PushDownUtils.pushOffset(sHolder.builder, offset)
+ if (isPushed) {
+ sHolder.pushedOffset = Some(offset)
+ }
+ isPushed
+ case Project(projectList, child) if projectList.forall(_.deterministic) =>
+ pushDownOffset(child, offset)
+ case _ => false
+ }
+
+ def pushDownLimitAndOffset(plan: LogicalPlan): LogicalPlan = plan.transform {
+ case offset @ LimitAndOffset(limit, offsetValue, child) =>
+ val (newChild, canRemoveLimit) = pushDownLimit(child, limit)
+ if (canRemoveLimit) {
+ // Try to push down OFFSET only if the LIMIT operator has been pushed and can be removed.
+ // For `df.limit(m).offset(n)`, try to push down `limit(m).offset(n)`.
+ // For example, `df.limit(5).offset(3)`, we can push down `limit(5).offset(3)`.
Review Comment:
```suggestion
```
--
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
[GitHub] [spark] cloud-fan commented on a diff in pull request #36295: [SPARK-38978][SQL] Support push down OFFSET to JDBC data source V2
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #36295:
URL: https://github.com/apache/spark/pull/36295#discussion_r873341127
##########
sql/catalyst/src/main/java/org/apache/spark/sql/connector/read/SupportsPushDownOffset.java:
##########
@@ -0,0 +1,36 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.connector.read;
+
+import org.apache.spark.annotation.Evolving;
+
+/**
+ * A mix-in interface for {@link ScanBuilder}. Data sources can implement this interface to
+ * push down OFFSET. Please note that the combination of OFFSET with other operations
+ * such as AGGREGATE, GROUP BY, SORT BY, CLUSTER BY, DISTRIBUTE BY, etc. is NOT pushed down.
Review Comment:
BTW we need to update `ScanBuider`'s classdoc for new pushdown support.
--
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
[GitHub] [spark] cloud-fan commented on a diff in pull request #36295: [SPARK-38978][SQL] Support push down OFFSET to JDBC data source V2
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #36295:
URL: https://github.com/apache/spark/pull/36295#discussion_r873515998
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala:
##########
@@ -419,6 +420,30 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper wit
}
}
+ private def pushDownOffset(plan: LogicalPlan, offset: Int): (LogicalPlan, Boolean) = plan match {
+ case operation @ ScanOperation(_, filter, sHolder: ScanBuilderHolder) if filter.isEmpty =>
+ val isPushed = PushDownUtils.pushOffset(sHolder.builder, offset)
+ if (isPushed) {
+ sHolder.pushedOffset = Some(offset)
+ }
+ (operation, isPushed)
+ case p: Project =>
+ val (newChild, isPushed) = pushDownOffset(p.child, offset)
+ (p.withNewChildren(Seq(newChild)), isPushed)
+ case other => (other, false)
+ }
+
+ def pushDownOffsets(plan: LogicalPlan): LogicalPlan = plan.transform {
+ // TODO supports push down Limit append Offset or Offset append Limit
+ case offset @ Offset(IntegerLiteral(n), child) =>
Review Comment:
We can match offset, limit + offset and offset + limit, similar to the planner rule (after https://github.com/apache/spark/pull/36541)
--
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
[GitHub] [spark] cloud-fan commented on a diff in pull request #36295: [SPARK-38978][SQL] Support push down OFFSET to JDBC data source V2
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #36295:
URL: https://github.com/apache/spark/pull/36295#discussion_r873511627
##########
sql/catalyst/src/main/java/org/apache/spark/sql/connector/read/ScanBuilder.java:
##########
@@ -23,7 +23,7 @@
* An interface for building the {@link Scan}. Implementations can mixin SupportsPushDownXYZ
* interfaces to do operator push down, and keep the operator push down result in the returned
* {@link Scan}. When pushing down operators, the push down order is:
- * sample -> filter -> aggregate -> limit -> column pruning.
+ * sample -> filter -> aggregate -> offset -> limit or top N -> column pruning.
Review Comment:
top n is a bit tricky as it's sort + limit. how about `aggregate -> limit/top-N(sort + limit) -> offset`? the order of limit and offset doesn't matter as we can always switch the order and adjust the value. And this order matches the physical plan more.
--
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
[GitHub] [spark] beliefer commented on a diff in pull request #36295: [SPARK-38978][SQL] DS V2 supports push down OFFSET operator
Posted by GitBox <gi...@apache.org>.
beliefer commented on code in PR #36295:
URL: https://github.com/apache/spark/pull/36295#discussion_r904473795
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/jdbc/JDBCScanBuilder.scala:
##########
@@ -139,6 +142,25 @@ case class JDBCScanBuilder(
false
}
+ override def pushOffset(offset: Int): Boolean = {
+ if (jdbcOptions.pushDownOffset && !isPartiallyPushed) {
+ // Spark pushes down LIMIT first, then OFFSET. In SQL statements, OFFSET is applied before
+ // LIMIT. Here we need to adjust the LIMIT value to match SQL statements.
+ // 1. For `dataset.limit(m).offset(n)`, try to push down `LIMIT (m - n) OFFSET n`.
+ // For example, `dataset.limit(5).offset(3)`, we can push down `LIMIT 2 OFFSET 3`.
+ // 2. For `dataset.offset(n).limit(m)`, try to push down `LIMIT m OFFSET n`.
+ // For example, `dataset.offset(3).limit(5)`, we can push down `LIMIT 5 OFFSET 3`.
+ // 3. For `dataset.offset(n)`, try to push down `OFFSET n`.
Review Comment:
OK
--
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
[GitHub] [spark] cloud-fan commented on a diff in pull request #36295: [SPARK-38978][SQL] DS V2 supports push down OFFSET operator
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #36295:
URL: https://github.com/apache/spark/pull/36295#discussion_r904487223
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala:
##########
@@ -407,7 +407,67 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper wit
case other => (other, false)
}
- def pushDownLimits(plan: LogicalPlan): LogicalPlan = plan.transform {
+ private def pushDownOffset(
+ plan: LogicalPlan,
+ offset: Int): Boolean = plan match {
+ case sHolder: ScanBuilderHolder =>
+ val isPushed = PushDownUtils.pushOffset(sHolder.builder, offset)
+ if (isPushed) {
+ sHolder.pushedOffset = Some(offset)
+ }
+ isPushed
+ case Project(projectList, child) if projectList.forall(_.deterministic) =>
+ pushDownOffset(child, offset)
+ case _ => false
+ }
+
+ def pushDownLimitAndOffset(plan: LogicalPlan): LogicalPlan = plan.transform {
+ case offset @ LimitAndOffset(limit, offsetValue, child) =>
+ val (newChild, canRemoveLimit) = pushDownLimit(child, limit)
+ if (canRemoveLimit) {
+ // Try to push down OFFSET only if the LIMIT operator has been pushed and can be removed.
+ // For `df.limit(m).offset(n)`, try to push down `limit(m).offset(n)`.
+ // For example, `df.limit(5).offset(3)`, we can push down `limit(5).offset(3)`.
+ val isPushed = pushDownOffset(newChild, offsetValue)
+ if (isPushed) {
+ newChild
+ } else {
+ // For `df.limit(m).offset(n)`, only push down `limit(m)`.
+ // Keep the OFFSET operator if we failed to push down OFFSET to the data source.
+ offset.withNewChildren(Seq(newChild))
+ }
+ } else {
+ // Keep the OFFSET operator if we can't remove LIMIT operator.
+ offset
+ }
+ case globalLimit @ OffsetAndLimit(offset, limit, child) =>
+ // For `df.offset(n).limit(m)`, we can push down `limit(m + n)` first.
+ val (newChild, canRemoveLimit) = pushDownLimit(child, limit + offset)
+ if (canRemoveLimit) {
+ // Try to push down OFFSET only if the LIMIT operator has been pushed and can be removed.
+ // For `df.offset(n).limit(m)`, try to push down `limit(m + n).offset(n)`.
+ // For example, `df.offset(3).limit(5)`, we can push down `limit(8).offset(3)`.
+ val isPushed = pushDownOffset(newChild, offset)
+ if (isPushed) {
+ newChild
+ } else {
+ // Still keep the OFFSET operator if we can't push it down.
+ // For example, `df.offset(3).limit(5)`, `limit(8)` has been pushed
+ // and can be removed, Spark still do `offset(3)`.
Review Comment:
```suggestion
```
--
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
[GitHub] [spark] cloud-fan commented on a diff in pull request #36295: [SPARK-38978][SQL] DS V2 supports push down OFFSET operator
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #36295:
URL: https://github.com/apache/spark/pull/36295#discussion_r902323385
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala:
##########
@@ -407,8 +407,69 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper wit
case other => (other, false)
}
- def pushDownLimits(plan: LogicalPlan): LogicalPlan = plan.transform {
+ private def pushDownOffset(
+ plan: LogicalPlan,
+ offset: Int): Boolean = plan match {
+ case sHolder: ScanBuilderHolder =>
+ val isPushed = PushDownUtils.pushOffset(sHolder.builder, offset)
+ if (isPushed) {
+ sHolder.pushedOffset = Some(offset)
+ }
+ isPushed
+ case p: Project =>
+ pushDownOffset(p.child, offset)
+ case _ => false
+ }
+
+ def pushDownLimitAndOffset(plan: LogicalPlan): LogicalPlan = plan.transform {
+ case offset @ LimitAndOffset(limit, offsetValue, child) =>
+ val (newChild, canRemoveLimit) = pushDownLimit(child, limit)
+ if (canRemoveLimit) {
+ // If we can remove limit, it indicates data source only have one partition.
+ // For `dataset.limit(m).offset(n)`, try to push down `offset(n).limit(m - n)`.
+ // For example, `dataset.limit(5).offset(3)`, we can push down `offset(3).limit(2)`.
+ val isPushed = pushDownOffset(newChild, offsetValue)
+ if (isPushed) {
+ newChild
+ } else {
+ // For `dataset.limit(m).offset(n)`, only push down `limit(m)`.
+ // Spark still do `offset(n)`.
+ offset.withNewChildren(Seq(newChild))
+ }
+ } else {
+ // If we can't push down limit and offset, return `Offset`.
+ offset
+ }
+ case globalLimit @ OffsetAndLimit(offset, limit, child) =>
+ val (newChild, canRemoveLimit) = pushDownLimit(child, limit + offset)
+ if (canRemoveLimit) {
+ // If we can remove limit, it indicates data source only have one partition.
+ // For `dataset.offset(n).limit(m)`, try to push down `limit(m + n).offset(n)`.
+ // For example, `dataset.offset(3).limit(5)`, we can push down `limit(8).offset(3)`.
+ val isPushed = pushDownOffset(newChild, offset)
+ if (isPushed) {
+ newChild
+ } else {
+ // For `dataset.offset(n).limit(m)`, try to push down `limit(m + n)`.
+ // Spark still do `offset(n).limit(m)`.
+ // For example, `dataset.offset(3).limit(5)`, we can push down `limit(8)`.
+ // Spark still do `offset(3).limit(5)`.
+ val newOffset = globalLimit.child.asInstanceOf[Offset].withNewChildren(Seq(newChild))
Review Comment:
I think it's just `Offset(offsetValue, newChild)`
--
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
[GitHub] [spark] beliefer commented on a diff in pull request #36295: [SPARK-38978][SQL] DS V2 supports push down OFFSET operator
Posted by GitBox <gi...@apache.org>.
beliefer commented on code in PR #36295:
URL: https://github.com/apache/spark/pull/36295#discussion_r902368213
##########
sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCV2Suite.scala:
##########
@@ -206,6 +207,355 @@ class JDBCV2Suite extends QueryTest with SharedSparkSession with ExplainSuiteHel
checkAnswer(df5, Seq(Row(10000.00, 1000.0, "amy")))
}
+ private def checkOffsetRemoved(df: DataFrame, removed: Boolean = true): Unit = {
+ val offsets = df.queryExecution.optimizedPlan.collect {
+ case offset: Offset => offset
+ }
+ if (removed) {
+ assert(offsets.isEmpty)
+ } else {
+ assert(offsets.nonEmpty)
+ }
+ }
+
+ test("simple scan with OFFSET") {
+ val df1 = spark.read
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .offset(1)
+ checkOffsetRemoved(df1)
+ checkPushedInfo(df1,
+ "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], PushedOffset: OFFSET 1,")
+ checkAnswer(df1, Seq(Row(1, "cathy", 9000.00, 1200.0, false)))
+
+ val df2 = spark.read
+ .option("pushDownOffset", "false")
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .offset(1)
+ checkOffsetRemoved(df2, false)
+ checkPushedInfo(df2,
+ "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], ReadSchema:")
+ checkAnswer(df2, Seq(Row(1, "cathy", 9000.00, 1200.0, false)))
+
+ val df3 = spark.read
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .sort($"salary")
+ .offset(1)
+ checkOffsetRemoved(df3, false)
+ checkPushedInfo(df3,
+ "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], ReadSchema:")
+ checkAnswer(df3, Seq(Row(1, "amy", 10000.00, 1000.0, true)))
+
+ val df4 = spark.read
+ .option("partitionColumn", "dept")
+ .option("lowerBound", "0")
+ .option("upperBound", "2")
+ .option("numPartitions", "2")
+ .table("h2.test.employee")
+ .filter($"dept" > 1)
+ .offset(1)
+ checkOffsetRemoved(df4, false)
+ checkPushedInfo(df4, "PushedFilters: [DEPT IS NOT NULL, DEPT > 1], ReadSchema:")
+ checkAnswer(df4, Seq(Row(2, "david", 10000, 1300, true), Row(6, "jen", 12000, 1200, true)))
+
+ val df5 = spark.read
+ .table("h2.test.employee")
+ .groupBy("DEPT").sum("SALARY")
+ .offset(1)
+ checkOffsetRemoved(df5, false)
+ checkPushedInfo(df5,
+ "PushedAggregates: [SUM(SALARY)], PushedFilters: [], PushedGroupByExpressions: [DEPT], ")
+ checkAnswer(df5, Seq(Row(2, 22000.00), Row(6, 12000.00)))
+
+ val name = udf { (x: String) => x.matches("cat|dav|amy") }
+ val sub = udf { (x: String) => x.substring(0, 3) }
+ val df6 = spark.read
+ .table("h2.test.employee")
+ .select($"SALARY", $"BONUS", sub($"NAME").as("shortName"))
+ .filter(name($"shortName"))
+ .offset(1)
+ checkOffsetRemoved(df6, false)
+ // OFFSET is pushed down only if all the filters are pushed down
+ checkPushedInfo(df6, "PushedFilters: [], ")
+ checkAnswer(df6, Seq(Row(10000.00, 1300.0, "dav"), Row(9000.00, 1200.0, "cat")))
+ }
+
+ test("simple scan with LIMIT and OFFSET") {
+ val df1 = spark.read
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .limit(2)
+ .offset(1)
+ checkLimitRemoved(df1)
+ checkOffsetRemoved(df1)
+ checkPushedInfo(df1,
+ "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], PushedLimit: LIMIT 2, PushedOffset: OFFSET 1,")
+ checkAnswer(df1, Seq(Row(1, "cathy", 9000.00, 1200.0, false)))
+
+ val df2 = spark.read
+ .option("pushDownLimit", "false")
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .limit(2)
+ .offset(1)
+ checkLimitRemoved(df2, false)
+ checkOffsetRemoved(df2, false)
+ checkPushedInfo(df2,
+ "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], ReadSchema:")
+ checkAnswer(df2, Seq(Row(1, "cathy", 9000.00, 1200.0, false)))
+
+ val df3 = spark.read
+ .option("pushDownOffset", "false")
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .limit(2)
+ .offset(1)
+ checkLimitRemoved(df3)
+ checkOffsetRemoved(df3, false)
+ checkPushedInfo(df3,
+ "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], PushedLimit: LIMIT 2, ReadSchema:")
+ checkAnswer(df3, Seq(Row(1, "cathy", 9000.00, 1200.0, false)))
+
+ val df4 = spark.read
+ .option("pushDownLimit", "false")
+ .option("pushDownOffset", "false")
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .limit(2)
+ .offset(1)
+ checkLimitRemoved(df4, false)
+ checkOffsetRemoved(df4, false)
+ checkPushedInfo(df4,
+ "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], ReadSchema:")
+ checkAnswer(df4, Seq(Row(1, "cathy", 9000.00, 1200.0, false)))
+
+ val df5 = spark.read
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .sort($"salary")
+ .limit(2)
+ .offset(1)
+ checkLimitRemoved(df5)
+ checkOffsetRemoved(df5)
+ checkPushedInfo(df5, "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], " +
+ "PushedOffset: OFFSET 1, PushedTopN: ORDER BY [SALARY ASC NULLS FIRST] LIMIT 2, ReadSchema:")
+ checkAnswer(df5, Seq(Row(1, "amy", 10000.00, 1000.0, true)))
+
+ val df6 = spark.read
+ .option("pushDownLimit", "false")
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .sort($"salary")
+ .limit(2)
+ .offset(1)
+ checkLimitRemoved(df6, false)
+ checkOffsetRemoved(df6, false)
+ checkPushedInfo(df6, "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], ReadSchema:")
+ checkAnswer(df6, Seq(Row(1, "amy", 10000.00, 1000.0, true)))
+
+ val df7 = spark.read
+ .option("pushDownOffset", "false")
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .sort($"salary")
+ .limit(2)
+ .offset(1)
+ checkLimitRemoved(df7)
+ checkOffsetRemoved(df7, false)
+ checkPushedInfo(df7, "PushedFilters: [DEPT IS NOT NULL, DEPT = 1]," +
+ " PushedTopN: ORDER BY [SALARY ASC NULLS FIRST] LIMIT 2, ReadSchema:")
+ checkAnswer(df7, Seq(Row(1, "amy", 10000.00, 1000.0, true)))
+
+ val df8 = spark.read
+ .option("pushDownLimit", "false")
+ .option("pushDownOffset", "false")
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .sort($"salary")
+ .limit(2)
+ .offset(1)
+ checkLimitRemoved(df8, false)
+ checkOffsetRemoved(df8, false)
+ checkPushedInfo(df8, "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], ReadSchema:")
+ checkAnswer(df8, Seq(Row(1, "amy", 10000.00, 1000.0, true)))
+
+ val df9 = spark.read
+ .option("partitionColumn", "dept")
+ .option("lowerBound", "0")
+ .option("upperBound", "2")
+ .option("numPartitions", "2")
+ .table("h2.test.employee")
+ .filter($"dept" > 1)
+ .limit(2)
+ .offset(1)
+ checkLimitRemoved(df9, false)
+ checkOffsetRemoved(df9, false)
+ checkPushedInfo(df9,
+ "PushedFilters: [DEPT IS NOT NULL, DEPT > 1], PushedLimit: LIMIT 2, ReadSchema:")
+ checkAnswer(df9, Seq(Row(2, "david", 10000.00, 1300.0, true)))
+
+ val df10 = spark.read
+ .table("h2.test.employee")
+ .groupBy("DEPT").sum("SALARY")
+ .limit(2)
+ .offset(1)
+ checkLimitRemoved(df10, false)
+ checkOffsetRemoved(df10, false)
+ checkPushedInfo(df10,
+ "PushedAggregates: [SUM(SALARY)], PushedFilters: [], PushedGroupByExpressions: [DEPT], ")
+ checkAnswer(df10, Seq(Row(2, 22000.00)))
+
+ val name = udf { (x: String) => x.matches("cat|dav|amy") }
+ val sub = udf { (x: String) => x.substring(0, 3) }
+ val df11 = spark.read
+ .table("h2.test.employee")
+ .select($"SALARY", $"BONUS", sub($"NAME").as("shortName"))
+ .filter(name($"shortName"))
+ .limit(2)
+ .offset(1)
+ checkLimitRemoved(df11, false)
+ checkOffsetRemoved(df11, false)
+ checkPushedInfo(df11, "PushedFilters: [], ")
+ checkAnswer(df11, Seq(Row(9000.00, 1200.0, "cat")))
+ }
+
+ test("simple scan with OFFSET and LIMIT") {
+ val df1 = spark.read
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .offset(1)
+ .limit(1)
+ checkLimitRemoved(df1)
+ checkOffsetRemoved(df1)
+ checkPushedInfo(df1,
+ "[DEPT IS NOT NULL, DEPT = 1], PushedLimit: LIMIT 2, PushedOffset: OFFSET 1,")
+ checkAnswer(df1, Seq(Row(1, "cathy", 9000.00, 1200.0, false)))
+
+ val df2 = spark.read
+ .option("pushDownOffset", "false")
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .offset(1)
+ .limit(1)
+ checkLimitRemoved(df2, false)
+ checkOffsetRemoved(df2, false)
+ checkPushedInfo(df2,
+ "[DEPT IS NOT NULL, DEPT = 1], PushedLimit: LIMIT 2, ReadSchema:")
+ checkAnswer(df2, Seq(Row(1, "cathy", 9000.00, 1200.0, false)))
+
+ val df3 = spark.read
+ .option("pushDownLimit", "false")
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .offset(1)
+ .limit(1)
+ checkLimitRemoved(df3, false)
+ checkOffsetRemoved(df3)
Review Comment:
Try to push limit 2. If you cannot push, but there is only one partition, you can still push offset 1
--
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
[GitHub] [spark] cloud-fan commented on a diff in pull request #36295: [SPARK-38978][SQL] Support push down OFFSET to JDBC data source V2
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #36295:
URL: https://github.com/apache/spark/pull/36295#discussion_r873511627
##########
sql/catalyst/src/main/java/org/apache/spark/sql/connector/read/ScanBuilder.java:
##########
@@ -23,7 +23,7 @@
* An interface for building the {@link Scan}. Implementations can mixin SupportsPushDownXYZ
* interfaces to do operator push down, and keep the operator push down result in the returned
* {@link Scan}. When pushing down operators, the push down order is:
- * sample -> filter -> aggregate -> limit -> column pruning.
+ * sample -> filter -> aggregate -> offset -> limit or top N -> column pruning.
Review Comment:
top n is a bit tricky as it's sort + limit. how about `aggregate -> limit/top-N(sort + limit) -> offset`? the order of limit and offset doesn't matter as we can always switch the order and adjust the value.
--
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
[GitHub] [spark] cloud-fan commented on a diff in pull request #36295: [SPARK-38978][SQL] Support push down OFFSET to JDBC data source V2
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #36295:
URL: https://github.com/apache/spark/pull/36295#discussion_r873340929
##########
sql/catalyst/src/main/java/org/apache/spark/sql/connector/read/SupportsPushDownOffset.java:
##########
@@ -0,0 +1,36 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.connector.read;
+
+import org.apache.spark.annotation.Evolving;
+
+/**
+ * A mix-in interface for {@link ScanBuilder}. Data sources can implement this interface to
+ * push down OFFSET. Please note that the combination of OFFSET with other operations
+ * such as AGGREGATE, GROUP BY, SORT BY, CLUSTER BY, DISTRIBUTE BY, etc. is NOT pushed down.
Review Comment:
I understand that this is copied from other pushdown interfaces, but I find it really hard to follow. We can push down OFFSET with many other operators if they follow the operator order we defined in `ScanBuilder`'s class doc.
--
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
[GitHub] [spark] beliefer commented on a diff in pull request #36295: [SPARK-38978][SQL] Support push down OFFSET to JDBC data source V2
Posted by GitBox <gi...@apache.org>.
beliefer commented on code in PR #36295:
URL: https://github.com/apache/spark/pull/36295#discussion_r874600195
##########
sql/catalyst/src/main/java/org/apache/spark/sql/connector/read/SupportsPushDownLimit.java:
##########
@@ -21,8 +21,8 @@
/**
* A mix-in interface for {@link ScanBuilder}. Data sources can implement this interface to
- * push down LIMIT. Please note that the combination of LIMIT with other operations
- * such as AGGREGATE, GROUP BY, SORT BY, CLUSTER BY, DISTRIBUTE BY, etc. is NOT pushed down.
+ * push down LIMIT. We can push down LIMIT with many other operations if they follow the
+ * operator order we defined in {@link ScanBuilder}'s class doc.
Review Comment:
OK
--
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
[GitHub] [spark] cloud-fan commented on a diff in pull request #36295: [SPARK-38978][SQL] Support push down OFFSET to JDBC data source V2
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #36295:
URL: https://github.com/apache/spark/pull/36295#discussion_r882692211
##########
sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCV2Suite.scala:
##########
@@ -203,6 +204,355 @@ class JDBCV2Suite extends QueryTest with SharedSparkSession with ExplainSuiteHel
checkAnswer(df5, Seq(Row(10000.00, 1000.0, "amy")))
}
+ private def checkOffsetRemoved(df: DataFrame, removed: Boolean = true): Unit = {
+ val offsets = df.queryExecution.optimizedPlan.collect {
+ case offset: Offset => offset
+ }
+ if (removed) {
+ assert(offsets.isEmpty)
+ } else {
+ assert(offsets.nonEmpty)
+ }
+ }
+
+ test("simple scan with OFFSET") {
+ val df1 = spark.read
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .offset(1)
+ checkOffsetRemoved(df1)
+ checkPushedInfo(df1,
+ "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], PushedOffset: OFFSET 1,")
+ checkAnswer(df1, Seq(Row(1, "cathy", 9000.00, 1200.0, false)))
+
+ val df2 = spark.read
+ .option("pushDownOffset", "false")
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .offset(1)
+ checkOffsetRemoved(df2, false)
+ checkPushedInfo(df2,
+ "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], ReadSchema:")
+ checkAnswer(df2, Seq(Row(1, "cathy", 9000.00, 1200.0, false)))
+
+ val df3 = spark.read
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .sort($"salary")
+ .offset(1)
+ checkOffsetRemoved(df3, false)
+ checkPushedInfo(df3,
+ "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], ReadSchema:")
+ checkAnswer(df3, Seq(Row(1, "amy", 10000.00, 1000.0, true)))
+
+ val df4 = spark.read
+ .option("partitionColumn", "dept")
+ .option("lowerBound", "0")
+ .option("upperBound", "2")
+ .option("numPartitions", "2")
+ .table("h2.test.employee")
+ .filter($"dept" > 1)
+ .offset(1)
+ checkOffsetRemoved(df4, false)
+ checkPushedInfo(df4, "PushedFilters: [DEPT IS NOT NULL, DEPT > 1], ReadSchema:")
+ checkAnswer(df4, Seq(Row(2, "david", 10000, 1300, true), Row(6, "jen", 12000, 1200, true)))
+
+ val df5 = spark.read
+ .table("h2.test.employee")
+ .groupBy("DEPT").sum("SALARY")
+ .offset(1)
+ checkOffsetRemoved(df5, false)
+ checkPushedInfo(df5,
+ "PushedAggregates: [SUM(SALARY)], PushedFilters: [], PushedGroupByExpressions: [DEPT], ")
+ checkAnswer(df5, Seq(Row(2, 22000.00), Row(6, 12000.00)))
+
+ val name = udf { (x: String) => x.matches("cat|dav|amy") }
+ val sub = udf { (x: String) => x.substring(0, 3) }
+ val df6 = spark.read
+ .table("h2.test.employee")
+ .select($"SALARY", $"BONUS", sub($"NAME").as("shortName"))
+ .filter(name($"shortName"))
+ .offset(1)
+ checkOffsetRemoved(df6, false)
+ // OFFSET is pushed down only if all the filters are pushed down
+ checkPushedInfo(df6, "PushedFilters: [], ")
+ checkAnswer(df6, Seq(Row(10000.00, 1300.0, "dav"), Row(9000.00, 1200.0, "cat")))
+ }
+
+ test("simple scan with LIMIT and OFFSET") {
+ val df1 = spark.read
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .limit(2)
+ .offset(1)
+ checkLimitRemoved(df1)
+ checkOffsetRemoved(df1)
+ checkPushedInfo(df1,
+ "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], PushedLimit: LIMIT 2, PushedOffset: OFFSET 1,")
+ checkAnswer(df1, Seq(Row(1, "cathy", 9000.00, 1200.0, false)))
+
+ val df2 = spark.read
+ .option("pushDownLimit", "false")
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .limit(2)
+ .offset(1)
+ checkLimitRemoved(df2, false)
+ checkOffsetRemoved(df2, false)
+ checkPushedInfo(df2,
+ "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], ReadSchema:")
+ checkAnswer(df2, Seq(Row(1, "cathy", 9000.00, 1200.0, false)))
+
+ val df3 = spark.read
+ .option("pushDownOffset", "false")
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .limit(2)
+ .offset(1)
+ checkLimitRemoved(df3)
+ checkOffsetRemoved(df3, false)
+ checkPushedInfo(df3,
+ "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], PushedLimit: LIMIT 2, ReadSchema:")
+ checkAnswer(df3, Seq(Row(1, "cathy", 9000.00, 1200.0, false)))
+
+ val df4 = spark.read
+ .option("pushDownLimit", "false")
+ .option("pushDownOffset", "false")
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .limit(2)
+ .offset(1)
+ checkLimitRemoved(df4, false)
+ checkOffsetRemoved(df4, false)
+ checkPushedInfo(df4,
+ "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], ReadSchema:")
+ checkAnswer(df4, Seq(Row(1, "cathy", 9000.00, 1200.0, false)))
+
+ val df5 = spark.read
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .sort($"salary")
+ .limit(2)
+ .offset(1)
+ checkLimitRemoved(df5)
+ checkOffsetRemoved(df5)
+ checkPushedInfo(df5, "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], " +
+ "PushedOffset: OFFSET 1, PushedTopN: ORDER BY [SALARY ASC NULLS FIRST] LIMIT 2, ReadSchema:")
+ checkAnswer(df5, Seq(Row(1, "amy", 10000.00, 1000.0, true)))
+
+ val df6 = spark.read
+ .option("pushDownLimit", "false")
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .sort($"salary")
+ .limit(2)
+ .offset(1)
+ checkLimitRemoved(df6, false)
+ checkOffsetRemoved(df6, false)
+ checkPushedInfo(df6, "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], ReadSchema:")
+ checkAnswer(df6, Seq(Row(1, "amy", 10000.00, 1000.0, true)))
+
+ val df7 = spark.read
+ .option("pushDownOffset", "false")
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .sort($"salary")
+ .limit(2)
+ .offset(1)
+ checkLimitRemoved(df7)
+ checkOffsetRemoved(df7, false)
+ checkPushedInfo(df7, "PushedFilters: [DEPT IS NOT NULL, DEPT = 1]," +
+ " PushedTopN: ORDER BY [SALARY ASC NULLS FIRST] LIMIT 2, ReadSchema:")
+ checkAnswer(df7, Seq(Row(1, "amy", 10000.00, 1000.0, true)))
+
+ val df8 = spark.read
+ .option("pushDownLimit", "false")
+ .option("pushDownOffset", "false")
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .sort($"salary")
+ .limit(2)
+ .offset(1)
+ checkLimitRemoved(df8, false)
+ checkOffsetRemoved(df8, false)
+ checkPushedInfo(df8, "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], ReadSchema:")
+ checkAnswer(df8, Seq(Row(1, "amy", 10000.00, 1000.0, true)))
+
+ val df9 = spark.read
+ .option("partitionColumn", "dept")
+ .option("lowerBound", "0")
+ .option("upperBound", "2")
+ .option("numPartitions", "2")
+ .table("h2.test.employee")
+ .filter($"dept" > 1)
+ .limit(2)
+ .offset(1)
+ checkLimitRemoved(df9, false)
+ checkOffsetRemoved(df9, false)
+ checkPushedInfo(df9,
+ "PushedFilters: [DEPT IS NOT NULL, DEPT > 1], PushedLimit: LIMIT 2, ReadSchema:")
+ checkAnswer(df9, Seq(Row(2, "david", 10000.00, 1300.0, true)))
+
+ val df10 = spark.read
+ .table("h2.test.employee")
+ .groupBy("DEPT").sum("SALARY")
Review Comment:
ditto
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] beliefer commented on a diff in pull request #36295: [SPARK-38978][SQL] DS V2 supports push down OFFSET operator
Posted by GitBox <gi...@apache.org>.
beliefer commented on code in PR #36295:
URL: https://github.com/apache/spark/pull/36295#discussion_r902351005
##########
sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCV2Suite.scala:
##########
@@ -206,6 +207,355 @@ class JDBCV2Suite extends QueryTest with SharedSparkSession with ExplainSuiteHel
checkAnswer(df5, Seq(Row(10000.00, 1000.0, "amy")))
}
+ private def checkOffsetRemoved(df: DataFrame, removed: Boolean = true): Unit = {
+ val offsets = df.queryExecution.optimizedPlan.collect {
+ case offset: Offset => offset
+ }
+ if (removed) {
+ assert(offsets.isEmpty)
+ } else {
+ assert(offsets.nonEmpty)
+ }
+ }
+
+ test("simple scan with OFFSET") {
+ val df1 = spark.read
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .offset(1)
+ checkOffsetRemoved(df1)
+ checkPushedInfo(df1,
+ "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], PushedOffset: OFFSET 1,")
+ checkAnswer(df1, Seq(Row(1, "cathy", 9000.00, 1200.0, false)))
+
+ val df2 = spark.read
+ .option("pushDownOffset", "false")
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .offset(1)
+ checkOffsetRemoved(df2, false)
+ checkPushedInfo(df2,
+ "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], ReadSchema:")
+ checkAnswer(df2, Seq(Row(1, "cathy", 9000.00, 1200.0, false)))
+
+ val df3 = spark.read
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .sort($"salary")
+ .offset(1)
+ checkOffsetRemoved(df3, false)
+ checkPushedInfo(df3,
+ "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], ReadSchema:")
+ checkAnswer(df3, Seq(Row(1, "amy", 10000.00, 1000.0, true)))
+
+ val df4 = spark.read
+ .option("partitionColumn", "dept")
+ .option("lowerBound", "0")
+ .option("upperBound", "2")
+ .option("numPartitions", "2")
+ .table("h2.test.employee")
+ .filter($"dept" > 1)
+ .offset(1)
+ checkOffsetRemoved(df4, false)
+ checkPushedInfo(df4, "PushedFilters: [DEPT IS NOT NULL, DEPT > 1], ReadSchema:")
+ checkAnswer(df4, Seq(Row(2, "david", 10000, 1300, true), Row(6, "jen", 12000, 1200, true)))
+
+ val df5 = spark.read
+ .table("h2.test.employee")
+ .groupBy("DEPT").sum("SALARY")
+ .offset(1)
+ checkOffsetRemoved(df5, false)
+ checkPushedInfo(df5,
+ "PushedAggregates: [SUM(SALARY)], PushedFilters: [], PushedGroupByExpressions: [DEPT], ")
+ checkAnswer(df5, Seq(Row(2, 22000.00), Row(6, 12000.00)))
+
+ val name = udf { (x: String) => x.matches("cat|dav|amy") }
+ val sub = udf { (x: String) => x.substring(0, 3) }
+ val df6 = spark.read
+ .table("h2.test.employee")
+ .select($"SALARY", $"BONUS", sub($"NAME").as("shortName"))
+ .filter(name($"shortName"))
+ .offset(1)
+ checkOffsetRemoved(df6, false)
+ // OFFSET is pushed down only if all the filters are pushed down
+ checkPushedInfo(df6, "PushedFilters: [], ")
+ checkAnswer(df6, Seq(Row(10000.00, 1300.0, "dav"), Row(9000.00, 1200.0, "cat")))
+ }
+
+ test("simple scan with LIMIT and OFFSET") {
+ val df1 = spark.read
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .limit(2)
+ .offset(1)
+ checkLimitRemoved(df1)
+ checkOffsetRemoved(df1)
+ checkPushedInfo(df1,
+ "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], PushedLimit: LIMIT 2, PushedOffset: OFFSET 1,")
+ checkAnswer(df1, Seq(Row(1, "cathy", 9000.00, 1200.0, false)))
+
+ val df2 = spark.read
+ .option("pushDownLimit", "false")
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .limit(2)
+ .offset(1)
+ checkLimitRemoved(df2, false)
+ checkOffsetRemoved(df2, false)
+ checkPushedInfo(df2,
+ "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], ReadSchema:")
+ checkAnswer(df2, Seq(Row(1, "cathy", 9000.00, 1200.0, false)))
+
+ val df3 = spark.read
+ .option("pushDownOffset", "false")
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .limit(2)
+ .offset(1)
+ checkLimitRemoved(df3)
+ checkOffsetRemoved(df3, false)
+ checkPushedInfo(df3,
+ "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], PushedLimit: LIMIT 2, ReadSchema:")
+ checkAnswer(df3, Seq(Row(1, "cathy", 9000.00, 1200.0, false)))
+
+ val df4 = spark.read
+ .option("pushDownLimit", "false")
+ .option("pushDownOffset", "false")
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .limit(2)
+ .offset(1)
+ checkLimitRemoved(df4, false)
+ checkOffsetRemoved(df4, false)
+ checkPushedInfo(df4,
+ "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], ReadSchema:")
+ checkAnswer(df4, Seq(Row(1, "cathy", 9000.00, 1200.0, false)))
+
+ val df5 = spark.read
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .sort($"salary")
+ .limit(2)
+ .offset(1)
+ checkLimitRemoved(df5)
+ checkOffsetRemoved(df5)
+ checkPushedInfo(df5, "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], " +
+ "PushedOffset: OFFSET 1, PushedTopN: ORDER BY [SALARY ASC NULLS FIRST] LIMIT 2, ReadSchema:")
+ checkAnswer(df5, Seq(Row(1, "amy", 10000.00, 1000.0, true)))
+
+ val df6 = spark.read
+ .option("pushDownLimit", "false")
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .sort($"salary")
+ .limit(2)
+ .offset(1)
+ checkLimitRemoved(df6, false)
+ checkOffsetRemoved(df6, false)
+ checkPushedInfo(df6, "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], ReadSchema:")
+ checkAnswer(df6, Seq(Row(1, "amy", 10000.00, 1000.0, true)))
+
+ val df7 = spark.read
+ .option("pushDownOffset", "false")
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .sort($"salary")
+ .limit(2)
+ .offset(1)
+ checkLimitRemoved(df7)
+ checkOffsetRemoved(df7, false)
+ checkPushedInfo(df7, "PushedFilters: [DEPT IS NOT NULL, DEPT = 1]," +
+ " PushedTopN: ORDER BY [SALARY ASC NULLS FIRST] LIMIT 2, ReadSchema:")
+ checkAnswer(df7, Seq(Row(1, "amy", 10000.00, 1000.0, true)))
+
+ val df8 = spark.read
+ .option("pushDownLimit", "false")
+ .option("pushDownOffset", "false")
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .sort($"salary")
+ .limit(2)
+ .offset(1)
+ checkLimitRemoved(df8, false)
+ checkOffsetRemoved(df8, false)
+ checkPushedInfo(df8, "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], ReadSchema:")
+ checkAnswer(df8, Seq(Row(1, "amy", 10000.00, 1000.0, true)))
+
+ val df9 = spark.read
+ .option("partitionColumn", "dept")
+ .option("lowerBound", "0")
+ .option("upperBound", "2")
+ .option("numPartitions", "2")
+ .table("h2.test.employee")
+ .filter($"dept" > 1)
+ .limit(2)
+ .offset(1)
+ checkLimitRemoved(df9, false)
+ checkOffsetRemoved(df9, false)
+ checkPushedInfo(df9,
+ "PushedFilters: [DEPT IS NOT NULL, DEPT > 1], PushedLimit: LIMIT 2, ReadSchema:")
+ checkAnswer(df9, Seq(Row(2, "david", 10000.00, 1300.0, true)))
+
+ val df10 = spark.read
+ .table("h2.test.employee")
+ .groupBy("DEPT").sum("SALARY")
+ .limit(2)
+ .offset(1)
+ checkLimitRemoved(df10, false)
+ checkOffsetRemoved(df10, false)
+ checkPushedInfo(df10,
+ "PushedAggregates: [SUM(SALARY)], PushedFilters: [], PushedGroupByExpressions: [DEPT], ")
+ checkAnswer(df10, Seq(Row(2, 22000.00)))
+
+ val name = udf { (x: String) => x.matches("cat|dav|amy") }
+ val sub = udf { (x: String) => x.substring(0, 3) }
+ val df11 = spark.read
+ .table("h2.test.employee")
+ .select($"SALARY", $"BONUS", sub($"NAME").as("shortName"))
+ .filter(name($"shortName"))
+ .limit(2)
+ .offset(1)
+ checkLimitRemoved(df11, false)
+ checkOffsetRemoved(df11, false)
+ checkPushedInfo(df11, "PushedFilters: [], ")
+ checkAnswer(df11, Seq(Row(9000.00, 1200.0, "cat")))
+ }
+
+ test("simple scan with OFFSET and LIMIT") {
+ val df1 = spark.read
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .offset(1)
+ .limit(1)
+ checkLimitRemoved(df1)
+ checkOffsetRemoved(df1)
+ checkPushedInfo(df1,
+ "[DEPT IS NOT NULL, DEPT = 1], PushedLimit: LIMIT 2, PushedOffset: OFFSET 1,")
+ checkAnswer(df1, Seq(Row(1, "cathy", 9000.00, 1200.0, false)))
+
+ val df2 = spark.read
+ .option("pushDownOffset", "false")
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .offset(1)
+ .limit(1)
+ checkLimitRemoved(df2, false)
Review Comment:
Yeah. Push down `limit 2` completely and then Spark only need do `offset(1)`.
--
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
[GitHub] [spark] cloud-fan closed pull request #36295: [SPARK-38978][SQL] DS V2 supports push down OFFSET operator
Posted by GitBox <gi...@apache.org>.
cloud-fan closed pull request #36295: [SPARK-38978][SQL] DS V2 supports push down OFFSET operator
URL: https://github.com/apache/spark/pull/36295
--
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
[GitHub] [spark] cloud-fan commented on a diff in pull request #36295: [SPARK-38978][SQL] DS V2 supports push down OFFSET operator
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #36295:
URL: https://github.com/apache/spark/pull/36295#discussion_r903805958
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala:
##########
@@ -407,7 +407,66 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper wit
case other => (other, false)
}
- def pushDownLimits(plan: LogicalPlan): LogicalPlan = plan.transform {
+ private def pushDownOffset(
+ plan: LogicalPlan,
+ offset: Int): Boolean = plan match {
+ case sHolder: ScanBuilderHolder =>
+ val isPushed = PushDownUtils.pushOffset(sHolder.builder, offset)
+ if (isPushed) {
+ sHolder.pushedOffset = Some(offset)
+ }
+ isPushed
+ case p: Project =>
+ pushDownOffset(p.child, offset)
+ case _ => false
+ }
+
+ def pushDownLimitAndOffset(plan: LogicalPlan): LogicalPlan = plan.transform {
+ case offset @ LimitAndOffset(limit, offsetValue, child) =>
+ val (newChild, canRemoveLimit) = pushDownLimit(child, limit)
+ if (canRemoveLimit) {
+ // If we can remove limit, it indicates data source only have one partition.
+ // For `dataset.limit(m).offset(n)`, try to push down `limit(m).offset(n)`.
+ // For example, `dataset.limit(5).offset(3)`, we can push down `limit(5).offset(3)`.
+ val isPushed = pushDownOffset(newChild, offsetValue)
+ if (isPushed) {
+ newChild
+ } else {
+ // For `dataset.limit(m).offset(n)`, only push down `limit(m)`.
+ // Keep the OFFSET operator if we failed to push down OFFSET to the data source.
+ offset.withNewChildren(Seq(newChild))
+ }
+ } else {
+ // Keep the OFFSET operator if we can't remove LIMIT operator.
+ offset
+ }
+ case globalLimit @ OffsetAndLimit(offset, limit, child) =>
+ val (newChild, canRemoveLimit) = pushDownLimit(child, limit + offset)
Review Comment:
add a comment
```
// For `df.offset(n).limit(m)`, we can push down `limit(m + n)` first.
```
--
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
[GitHub] [spark] cloud-fan commented on a diff in pull request #36295: [SPARK-38978][SQL] DS V2 supports push down OFFSET operator
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #36295:
URL: https://github.com/apache/spark/pull/36295#discussion_r904486692
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala:
##########
@@ -407,7 +407,67 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper wit
case other => (other, false)
}
- def pushDownLimits(plan: LogicalPlan): LogicalPlan = plan.transform {
+ private def pushDownOffset(
+ plan: LogicalPlan,
+ offset: Int): Boolean = plan match {
+ case sHolder: ScanBuilderHolder =>
+ val isPushed = PushDownUtils.pushOffset(sHolder.builder, offset)
+ if (isPushed) {
+ sHolder.pushedOffset = Some(offset)
+ }
+ isPushed
+ case Project(projectList, child) if projectList.forall(_.deterministic) =>
+ pushDownOffset(child, offset)
+ case _ => false
+ }
+
+ def pushDownLimitAndOffset(plan: LogicalPlan): LogicalPlan = plan.transform {
+ case offset @ LimitAndOffset(limit, offsetValue, child) =>
+ val (newChild, canRemoveLimit) = pushDownLimit(child, limit)
+ if (canRemoveLimit) {
+ // Try to push down OFFSET only if the LIMIT operator has been pushed and can be removed.
+ // For `df.limit(m).offset(n)`, try to push down `limit(m).offset(n)`.
+ // For example, `df.limit(5).offset(3)`, we can push down `limit(5).offset(3)`.
+ val isPushed = pushDownOffset(newChild, offsetValue)
+ if (isPushed) {
+ newChild
+ } else {
+ // For `df.limit(m).offset(n)`, only push down `limit(m)`.
+ // Keep the OFFSET operator if we failed to push down OFFSET to the data source.
+ offset.withNewChildren(Seq(newChild))
+ }
+ } else {
+ // Keep the OFFSET operator if we can't remove LIMIT operator.
+ offset
+ }
+ case globalLimit @ OffsetAndLimit(offset, limit, child) =>
+ // For `df.offset(n).limit(m)`, we can push down `limit(m + n)` first.
+ val (newChild, canRemoveLimit) = pushDownLimit(child, limit + offset)
+ if (canRemoveLimit) {
+ // Try to push down OFFSET only if the LIMIT operator has been pushed and can be removed.
+ // For `df.offset(n).limit(m)`, try to push down `limit(m + n).offset(n)`.
+ // For example, `df.offset(3).limit(5)`, we can push down `limit(8).offset(3)`.
Review Comment:
```suggestion
```
--
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
[GitHub] [spark] cloud-fan commented on a diff in pull request #36295: [SPARK-38978][SQL] DS V2 supports push down OFFSET operator
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #36295:
URL: https://github.com/apache/spark/pull/36295#discussion_r903765555
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala:
##########
@@ -407,7 +407,66 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper wit
case other => (other, false)
}
- def pushDownLimits(plan: LogicalPlan): LogicalPlan = plan.transform {
+ private def pushDownOffset(
+ plan: LogicalPlan,
+ offset: Int): Boolean = plan match {
+ case sHolder: ScanBuilderHolder =>
+ val isPushed = PushDownUtils.pushOffset(sHolder.builder, offset)
+ if (isPushed) {
+ sHolder.pushedOffset = Some(offset)
+ }
+ isPushed
+ case p: Project =>
+ pushDownOffset(p.child, offset)
+ case _ => false
+ }
+
+ def pushDownLimitAndOffset(plan: LogicalPlan): LogicalPlan = plan.transform {
+ case offset @ LimitAndOffset(limit, offsetValue, child) =>
+ val (newChild, canRemoveLimit) = pushDownLimit(child, limit)
+ if (canRemoveLimit) {
+ // If we can remove limit, it indicates data source only have one partition.
Review Comment:
I don't think it applies to all data sources. Let's be more general here:
```
// Try to push down OFFSET only if the LIMIT operator has been pushed and can be removed.
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] cloud-fan commented on a diff in pull request #36295: [SPARK-38978][SQL] Support push down OFFSET to JDBC data source V2
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #36295:
URL: https://github.com/apache/spark/pull/36295#discussion_r877141680
##########
sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCV2Suite.scala:
##########
@@ -203,6 +204,245 @@ class JDBCV2Suite extends QueryTest with SharedSparkSession with ExplainSuiteHel
checkAnswer(df5, Seq(Row(10000.00, 1000.0, "amy")))
}
+ private def checkOffsetRemoved(df: DataFrame, removed: Boolean = true): Unit = {
+ val offsets = df.queryExecution.optimizedPlan.collect {
+ case offset: Offset => offset
+ }
+ if (removed) {
+ assert(offsets.isEmpty)
+ } else {
+ assert(offsets.nonEmpty)
+ }
+ }
+
+ test("simple scan with OFFSET") {
+ val df1 = spark.read
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .offset(1)
+ checkOffsetRemoved(df1)
+ checkPushedInfo(df1,
+ "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], PushedOffset: OFFSET 1,")
+ checkAnswer(df1, Seq(Row(1, "cathy", 9000.00, 1200.0, false)))
+
+ val df2 = spark.read
+ .option("pushDownOffset", "false")
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .offset(1)
+ checkOffsetRemoved(df2, false)
+ checkPushedInfo(df2,
+ "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], ReadSchema:")
+ checkAnswer(df2, Seq(Row(1, "cathy", 9000.00, 1200.0, false)))
+
+ val df3 = spark.read
+ .option("partitionColumn", "dept")
+ .option("lowerBound", "0")
+ .option("upperBound", "2")
+ .option("numPartitions", "2")
+ .table("h2.test.employee")
+ .filter($"dept" > 1)
+ .offset(1)
+ checkOffsetRemoved(df3, false)
+ checkPushedInfo(df3, "PushedFilters: [DEPT IS NOT NULL, DEPT > 1], ReadSchema:")
+ checkAnswer(df3, Seq(Row(2, "david", 10000, 1300, true), Row(6, "jen", 12000, 1200, true)))
+
+ val df4 = spark.read
+ .table("h2.test.employee")
+ .groupBy("DEPT").sum("SALARY")
+ .offset(1)
+ checkOffsetRemoved(df4, false)
+ checkPushedInfo(df4,
+ "PushedAggregates: [SUM(SALARY)], PushedFilters: [], PushedGroupByExpressions: [DEPT], ")
+ checkAnswer(df4, Seq(Row(2, 22000.00), Row(6, 12000.00)))
+
+ val name = udf { (x: String) => x.matches("cat|dav|amy") }
+ val sub = udf { (x: String) => x.substring(0, 3) }
+ val df5 = spark.read
+ .table("h2.test.employee")
+ .select($"SALARY", $"BONUS", sub($"NAME").as("shortName"))
+ .filter(name($"shortName"))
+ .offset(1)
+ checkOffsetRemoved(df5, false)
+ // OFFSET is pushed down only if all the filters are pushed down
+ checkPushedInfo(df5, "PushedFilters: [], ")
+ checkAnswer(df5, Seq(Row(10000.00, 1300.0, "dav"), Row(9000.00, 1200.0, "cat")))
+ }
+
+ test("simple scan with LIMIT and OFFSET") {
+ val df1 = spark.read
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .limit(2)
+ .offset(1)
+ checkLimitRemoved(df1)
+ checkOffsetRemoved(df1)
+ checkPushedInfo(df1,
+ "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], PushedLimit: LIMIT 1, PushedOffset: OFFSET 1,")
Review Comment:
This does not match https://github.com/apache/spark/pull/36295/files#diff-85c754089fc8e0db142a16714e92b127001bab9e6433684d1e3a15af04cb219aR26
Assume that I have a local array data source. According to the API doc, Spark pushes down LIMIT first. For this query, I'll do `array.take(1).drop(1)`. This is wrong and doesn't match the query `df.limit(2).offset(1)`.
we should either fix the API doc, or fix the pushdown logic
--
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
[GitHub] [spark] cloud-fan commented on a diff in pull request #36295: [SPARK-38978][SQL] Support push down OFFSET to JDBC data source V2
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #36295:
URL: https://github.com/apache/spark/pull/36295#discussion_r879216237
##########
sql/catalyst/src/main/java/org/apache/spark/sql/connector/read/ScanBuilder.java:
##########
@@ -23,7 +23,8 @@
* An interface for building the {@link Scan}. Implementations can mixin SupportsPushDownXYZ
* interfaces to do operator push down, and keep the operator push down result in the returned
* {@link Scan}. When pushing down operators, the push down order is:
- * sample -> filter -> aggregate -> limit -> column pruning.
+ * sample -> filter -> aggregate -> offset -> limit/top-n(sort + limit) ->
Review Comment:
We can switch the order of LIMIT and OFFSET to match this defined order, but we can't switch the order of ORDER BY and OFFSET. What if the SQL query is `ORDER BY ... LIMIT ... OFFSET ...`? Nothing can be pushed down?
--
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
[GitHub] [spark] beliefer commented on a diff in pull request #36295: [SPARK-38978][SQL] Support push down OFFSET to JDBC data source V2
Posted by GitBox <gi...@apache.org>.
beliefer commented on code in PR #36295:
URL: https://github.com/apache/spark/pull/36295#discussion_r883245824
##########
sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCV2Suite.scala:
##########
@@ -203,6 +204,355 @@ class JDBCV2Suite extends QueryTest with SharedSparkSession with ExplainSuiteHel
checkAnswer(df5, Seq(Row(10000.00, 1000.0, "amy")))
}
+ private def checkOffsetRemoved(df: DataFrame, removed: Boolean = true): Unit = {
+ val offsets = df.queryExecution.optimizedPlan.collect {
+ case offset: Offset => offset
+ }
+ if (removed) {
+ assert(offsets.isEmpty)
+ } else {
+ assert(offsets.nonEmpty)
+ }
+ }
+
+ test("simple scan with OFFSET") {
+ val df1 = spark.read
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .offset(1)
+ checkOffsetRemoved(df1)
+ checkPushedInfo(df1,
+ "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], PushedOffset: OFFSET 1,")
+ checkAnswer(df1, Seq(Row(1, "cathy", 9000.00, 1200.0, false)))
+
+ val df2 = spark.read
+ .option("pushDownOffset", "false")
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .offset(1)
+ checkOffsetRemoved(df2, false)
+ checkPushedInfo(df2,
+ "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], ReadSchema:")
+ checkAnswer(df2, Seq(Row(1, "cathy", 9000.00, 1200.0, false)))
+
+ val df3 = spark.read
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .sort($"salary")
+ .offset(1)
+ checkOffsetRemoved(df3, false)
+ checkPushedInfo(df3,
+ "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], ReadSchema:")
+ checkAnswer(df3, Seq(Row(1, "amy", 10000.00, 1000.0, true)))
+
+ val df4 = spark.read
+ .option("partitionColumn", "dept")
+ .option("lowerBound", "0")
+ .option("upperBound", "2")
+ .option("numPartitions", "2")
+ .table("h2.test.employee")
+ .filter($"dept" > 1)
+ .offset(1)
+ checkOffsetRemoved(df4, false)
+ checkPushedInfo(df4, "PushedFilters: [DEPT IS NOT NULL, DEPT > 1], ReadSchema:")
+ checkAnswer(df4, Seq(Row(2, "david", 10000, 1300, true), Row(6, "jen", 12000, 1200, true)))
+
+ val df5 = spark.read
+ .table("h2.test.employee")
+ .groupBy("DEPT").sum("SALARY")
Review Comment:
In fact, limit have the same issue. I think we can support them in a single PR.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] cloud-fan commented on a diff in pull request #36295: [SPARK-38978][SQL] Support push down OFFSET to JDBC data source V2
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #36295:
URL: https://github.com/apache/spark/pull/36295#discussion_r873511627
##########
sql/catalyst/src/main/java/org/apache/spark/sql/connector/read/ScanBuilder.java:
##########
@@ -23,7 +23,7 @@
* An interface for building the {@link Scan}. Implementations can mixin SupportsPushDownXYZ
* interfaces to do operator push down, and keep the operator push down result in the returned
* {@link Scan}. When pushing down operators, the push down order is:
- * sample -> filter -> aggregate -> limit -> column pruning.
+ * sample -> filter -> aggregate -> offset -> limit or top N -> column pruning.
Review Comment:
top n is a bit tricky as it's sort + limit. how about `aggregate -> limit/top-n(sort + limit) -> offset`? the order of limit and offset doesn't matter as we can always switch the order and adjust the value.
--
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
[GitHub] [spark] beliefer commented on a diff in pull request #36295: [SPARK-38978][SQL] Support push down OFFSET to JDBC data source V2
Posted by GitBox <gi...@apache.org>.
beliefer commented on code in PR #36295:
URL: https://github.com/apache/spark/pull/36295#discussion_r874727085
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JDBCRDD.scala:
##########
@@ -304,10 +307,11 @@ private[jdbc] class JDBCRDD(
}
val myLimitClause: String = dialect.getLimitClause(limit)
+ val myOffsetClause: String = dialect.getOffsetClause(offset)
val sqlText = options.prepareQuery +
s"SELECT $columnList FROM ${options.tableOrQuery} $myTableSampleClause" +
- s" $myWhereClause $getGroupByClause $getOrderByClause $myLimitClause"
+ s" $myWhereClause $getGroupByClause $getOrderByClause $myLimitClause $myOffsetClause"
Review Comment:
Good idea.
--
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
[GitHub] [spark] cloud-fan commented on a diff in pull request #36295: [SPARK-38978][SQL] Support push down OFFSET to JDBC data source V2
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #36295:
URL: https://github.com/apache/spark/pull/36295#discussion_r874586232
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JDBCRDD.scala:
##########
@@ -304,10 +307,11 @@ private[jdbc] class JDBCRDD(
}
val myLimitClause: String = dialect.getLimitClause(limit)
+ val myOffsetClause: String = dialect.getOffsetClause(offset)
val sqlText = options.prepareQuery +
s"SELECT $columnList FROM ${options.tableOrQuery} $myTableSampleClause" +
- s" $myWhereClause $getGroupByClause $getOrderByClause $myLimitClause"
+ s" $myWhereClause $getGroupByClause $getOrderByClause $myLimitClause $myOffsetClause"
Review Comment:
We need to be careful here. The semantic of pushdown is LIMIT first, then OFFSET, while `LIMIT a OFFSET b` means OFFSET first in SQL spec.
--
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
[GitHub] [spark] beliefer commented on a diff in pull request #36295: [SPARK-38978][SQL] Support push down OFFSET to JDBC data source V2
Posted by GitBox <gi...@apache.org>.
beliefer commented on code in PR #36295:
URL: https://github.com/apache/spark/pull/36295#discussion_r871033582
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala:
##########
@@ -44,6 +44,7 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper wit
pushDownFilters,
pushDownAggregates,
pushDownLimits,
+ pushDownOffsets,
Review Comment:
OK
--
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
[GitHub] [spark] cloud-fan commented on a diff in pull request #36295: [SPARK-38978][SQL] Support push down OFFSET to JDBC data source V2
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #36295:
URL: https://github.com/apache/spark/pull/36295#discussion_r877141680
##########
sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCV2Suite.scala:
##########
@@ -203,6 +204,245 @@ class JDBCV2Suite extends QueryTest with SharedSparkSession with ExplainSuiteHel
checkAnswer(df5, Seq(Row(10000.00, 1000.0, "amy")))
}
+ private def checkOffsetRemoved(df: DataFrame, removed: Boolean = true): Unit = {
+ val offsets = df.queryExecution.optimizedPlan.collect {
+ case offset: Offset => offset
+ }
+ if (removed) {
+ assert(offsets.isEmpty)
+ } else {
+ assert(offsets.nonEmpty)
+ }
+ }
+
+ test("simple scan with OFFSET") {
+ val df1 = spark.read
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .offset(1)
+ checkOffsetRemoved(df1)
+ checkPushedInfo(df1,
+ "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], PushedOffset: OFFSET 1,")
+ checkAnswer(df1, Seq(Row(1, "cathy", 9000.00, 1200.0, false)))
+
+ val df2 = spark.read
+ .option("pushDownOffset", "false")
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .offset(1)
+ checkOffsetRemoved(df2, false)
+ checkPushedInfo(df2,
+ "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], ReadSchema:")
+ checkAnswer(df2, Seq(Row(1, "cathy", 9000.00, 1200.0, false)))
+
+ val df3 = spark.read
+ .option("partitionColumn", "dept")
+ .option("lowerBound", "0")
+ .option("upperBound", "2")
+ .option("numPartitions", "2")
+ .table("h2.test.employee")
+ .filter($"dept" > 1)
+ .offset(1)
+ checkOffsetRemoved(df3, false)
+ checkPushedInfo(df3, "PushedFilters: [DEPT IS NOT NULL, DEPT > 1], ReadSchema:")
+ checkAnswer(df3, Seq(Row(2, "david", 10000, 1300, true), Row(6, "jen", 12000, 1200, true)))
+
+ val df4 = spark.read
+ .table("h2.test.employee")
+ .groupBy("DEPT").sum("SALARY")
+ .offset(1)
+ checkOffsetRemoved(df4, false)
+ checkPushedInfo(df4,
+ "PushedAggregates: [SUM(SALARY)], PushedFilters: [], PushedGroupByExpressions: [DEPT], ")
+ checkAnswer(df4, Seq(Row(2, 22000.00), Row(6, 12000.00)))
+
+ val name = udf { (x: String) => x.matches("cat|dav|amy") }
+ val sub = udf { (x: String) => x.substring(0, 3) }
+ val df5 = spark.read
+ .table("h2.test.employee")
+ .select($"SALARY", $"BONUS", sub($"NAME").as("shortName"))
+ .filter(name($"shortName"))
+ .offset(1)
+ checkOffsetRemoved(df5, false)
+ // OFFSET is pushed down only if all the filters are pushed down
+ checkPushedInfo(df5, "PushedFilters: [], ")
+ checkAnswer(df5, Seq(Row(10000.00, 1300.0, "dav"), Row(9000.00, 1200.0, "cat")))
+ }
+
+ test("simple scan with LIMIT and OFFSET") {
+ val df1 = spark.read
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .limit(2)
+ .offset(1)
+ checkLimitRemoved(df1)
+ checkOffsetRemoved(df1)
+ checkPushedInfo(df1,
+ "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], PushedLimit: LIMIT 1, PushedOffset: OFFSET 1,")
Review Comment:
This does not match https://github.com/apache/spark/pull/36295/files#diff-85c754089fc8e0db142a16714e92b127001bab9e6433684d1e3a15af04cb219aR26
Assume that I have a local array data source. According to the API doc, Spark pushes down LIMIT first. For this query, I'll do `array.limit(1).drop(1)`. This is wrong and doesn't match the query `df.limit(2).offset(1)`.
we should either fix the API doc, or fix the pushdown logic
--
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
[GitHub] [spark] beliefer commented on a diff in pull request #36295: [SPARK-38978][SQL] Support push down limit and offset to JDBC data source V2
Posted by GitBox <gi...@apache.org>.
beliefer commented on code in PR #36295:
URL: https://github.com/apache/spark/pull/36295#discussion_r856159262
##########
sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala:
##########
@@ -2044,6 +2044,16 @@ class Dataset[T] private[sql](
Limit(Literal(n), logicalPlan)
}
+ /**
+ * Returns a new Dataset by skipping the first `m` rows and then taking the followed `n` rows.
+ *
+ * @group typedrel
+ * @since 3.4.0
+ */
+ def limitAndOffset(n: Int, m: Int): Dataset[T] = withTypedPlan {
Review Comment:
https://github.com/apache/spark/pull/36319 move the check of Offset from analysis to finsh analysis, so we can add `def offset(...)`.
##########
sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala:
##########
@@ -2044,6 +2044,16 @@ class Dataset[T] private[sql](
Limit(Literal(n), logicalPlan)
}
+ /**
+ * Returns a new Dataset by skipping the first `m` rows and then taking the followed `n` rows.
+ *
+ * @group typedrel
+ * @since 3.4.0
+ */
+ def limitAndOffset(n: Int, m: Int): Dataset[T] = withTypedPlan {
Review Comment:
https://github.com/apache/spark/pull/36319 moves the check of Offset from analysis to finsh analysis, so we can add `def offset(...)`.
--
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
[GitHub] [spark] cloud-fan commented on a diff in pull request #36295: [SPARK-38978][SQL] DS V2 supports push down OFFSET operator
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #36295:
URL: https://github.com/apache/spark/pull/36295#discussion_r903820510
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/jdbc/JDBCScanBuilder.scala:
##########
@@ -139,6 +142,25 @@ case class JDBCScanBuilder(
false
}
+ override def pushOffset(offset: Int): Boolean = {
+ if (jdbcOptions.pushDownOffset && !isPartiallyPushed) {
+ // Spark pushes down LIMIT first, then OFFSET. In SQL statements, OFFSET is applied before
+ // LIMIT. Here we need to adjust the LIMIT value to match SQL statements.
+ // 1. For `dataset.limit(m).offset(n)`, try to push down `LIMIT (m - n) OFFSET n`.
+ // For example, `dataset.limit(5).offset(3)`, we can push down `LIMIT 2 OFFSET 3`.
+ // 2. For `dataset.offset(n).limit(m)`, try to push down `LIMIT m OFFSET n`.
+ // For example, `dataset.offset(3).limit(5)`, we can push down `LIMIT 5 OFFSET 3`.
+ // 3. For `dataset.offset(n)`, try to push down `OFFSET n`.
Review Comment:
I don't think the 1,2,3 comments make sense here. It's about the internal details of the spark pushdown rule, and data source shouldn't care about it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] cloud-fan commented on a diff in pull request #36295: [SPARK-38978][SQL] DS V2 supports push down OFFSET operator
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #36295:
URL: https://github.com/apache/spark/pull/36295#discussion_r902324439
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala:
##########
@@ -407,8 +407,69 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper wit
case other => (other, false)
}
- def pushDownLimits(plan: LogicalPlan): LogicalPlan = plan.transform {
+ private def pushDownOffset(
+ plan: LogicalPlan,
+ offset: Int): Boolean = plan match {
+ case sHolder: ScanBuilderHolder =>
+ val isPushed = PushDownUtils.pushOffset(sHolder.builder, offset)
+ if (isPushed) {
+ sHolder.pushedOffset = Some(offset)
+ }
+ isPushed
+ case p: Project =>
+ pushDownOffset(p.child, offset)
+ case _ => false
+ }
+
+ def pushDownLimitAndOffset(plan: LogicalPlan): LogicalPlan = plan.transform {
+ case offset @ LimitAndOffset(limit, offsetValue, child) =>
+ val (newChild, canRemoveLimit) = pushDownLimit(child, limit)
+ if (canRemoveLimit) {
+ // If we can remove limit, it indicates data source only have one partition.
+ // For `dataset.limit(m).offset(n)`, try to push down `offset(n).limit(m - n)`.
+ // For example, `dataset.limit(5).offset(3)`, we can push down `offset(3).limit(2)`.
+ val isPushed = pushDownOffset(newChild, offsetValue)
+ if (isPushed) {
+ newChild
+ } else {
+ // For `dataset.limit(m).offset(n)`, only push down `limit(m)`.
+ // Spark still do `offset(n)`.
+ offset.withNewChildren(Seq(newChild))
+ }
+ } else {
+ // If we can't push down limit and offset, return `Offset`.
+ offset
+ }
+ case globalLimit @ OffsetAndLimit(offset, limit, child) =>
+ val (newChild, canRemoveLimit) = pushDownLimit(child, limit + offset)
+ if (canRemoveLimit) {
+ // If we can remove limit, it indicates data source only have one partition.
+ // For `dataset.offset(n).limit(m)`, try to push down `limit(m + n).offset(n)`.
+ // For example, `dataset.offset(3).limit(5)`, we can push down `limit(8).offset(3)`.
+ val isPushed = pushDownOffset(newChild, offset)
+ if (isPushed) {
+ newChild
+ } else {
+ // For `dataset.offset(n).limit(m)`, try to push down `limit(m + n)`.
+ // Spark still do `offset(n).limit(m)`.
+ // For example, `dataset.offset(3).limit(5)`, we can push down `limit(8)`.
+ // Spark still do `offset(3).limit(5)`.
+ val newOffset = globalLimit.child.asInstanceOf[Offset].withNewChildren(Seq(newChild))
+ globalLimit.withNewChildren(Seq(newOffset))
+ }
+ } else {
+ // For `dataset.offset(n).limit(m)`, try to push down `offset(n)`.
+ // Spark still do `limit(m)`.
+ val isPushed = pushDownOffset(child, offset)
+ if (isPushed) {
+ globalLimit.withNewChildren(Seq(child))
+ } else {
+ // If we can't push down limit and offset, return `GlobalLimit`.
+ globalLimit
+ }
+ }
case globalLimit @ Limit(IntegerLiteral(limitValue), child) =>
+ // For `dataset.limit(m)`, try to push down `limit(m)`.
Review Comment:
this comment doesn't seem to provide any value...
--
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
[GitHub] [spark] beliefer commented on a diff in pull request #36295: [SPARK-38978][SQL] DS V2 supports push down OFFSET operator
Posted by GitBox <gi...@apache.org>.
beliefer commented on code in PR #36295:
URL: https://github.com/apache/spark/pull/36295#discussion_r902375146
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala:
##########
@@ -407,8 +407,69 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper wit
case other => (other, false)
}
- def pushDownLimits(plan: LogicalPlan): LogicalPlan = plan.transform {
+ private def pushDownOffset(
+ plan: LogicalPlan,
+ offset: Int): Boolean = plan match {
+ case sHolder: ScanBuilderHolder =>
+ val isPushed = PushDownUtils.pushOffset(sHolder.builder, offset)
+ if (isPushed) {
+ sHolder.pushedOffset = Some(offset)
+ }
+ isPushed
+ case p: Project =>
+ pushDownOffset(p.child, offset)
+ case _ => false
+ }
+
+ def pushDownLimitAndOffset(plan: LogicalPlan): LogicalPlan = plan.transform {
+ case offset @ LimitAndOffset(limit, offsetValue, child) =>
+ val (newChild, canRemoveLimit) = pushDownLimit(child, limit)
+ if (canRemoveLimit) {
+ // If we can remove limit, it indicates data source only have one partition.
+ // For `dataset.limit(m).offset(n)`, try to push down `offset(n).limit(m - n)`.
+ // For example, `dataset.limit(5).offset(3)`, we can push down `offset(3).limit(2)`.
+ val isPushed = pushDownOffset(newChild, offsetValue)
+ if (isPushed) {
+ newChild
+ } else {
+ // For `dataset.limit(m).offset(n)`, only push down `limit(m)`.
+ // Spark still do `offset(n)`.
+ offset.withNewChildren(Seq(newChild))
+ }
+ } else {
+ // If we can't push down limit and offset, return `Offset`.
Review Comment:
How about `If we can't remove Limit and push down offset, return Offset.` ?
--
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
[GitHub] [spark] cloud-fan commented on a diff in pull request #36295: [SPARK-38978][SQL] Support push down OFFSET to JDBC data source V2
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #36295:
URL: https://github.com/apache/spark/pull/36295#discussion_r880689282
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala:
##########
@@ -419,6 +420,60 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper wit
}
}
+ private def pushDownOffset(
+ plan: LogicalPlan,
+ offset: Int): (LogicalPlan, Boolean) = plan match {
+ case operation @ ScanOperation(_, filter, sHolder: ScanBuilderHolder) if filter.isEmpty =>
+ val isPushed = PushDownUtils.pushOffset(sHolder.builder, offset)
+ if (isPushed) {
+ sHolder.pushedOffset = Some(offset)
+ }
+ (operation, isPushed)
+ case p: Project =>
+ val (newChild, isPushed) = pushDownOffset(p.child, offset)
+ (p.withNewChildren(Seq(newChild)), isPushed)
+ case other => (other, false)
+ }
+
+ def pushDownOffsets(plan: LogicalPlan): LogicalPlan = plan.transform {
Review Comment:
I think the code can be much easier to read if we push down limit and offset together.
```
def pushDownLimitAndOffset(plan: LogicalPlan): LogicalPlan = plan.transform {
case Limit...
case Offset...
case LimitAndOffset ...
...
}
```
the code should be similar to the plann rule `SpecialLimits`
--
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
[GitHub] [spark] beliefer commented on a diff in pull request #36295: [SPARK-38978][SQL] DS V2 supports push down OFFSET operator
Posted by GitBox <gi...@apache.org>.
beliefer commented on code in PR #36295:
URL: https://github.com/apache/spark/pull/36295#discussion_r902375146
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala:
##########
@@ -407,8 +407,69 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper wit
case other => (other, false)
}
- def pushDownLimits(plan: LogicalPlan): LogicalPlan = plan.transform {
+ private def pushDownOffset(
+ plan: LogicalPlan,
+ offset: Int): Boolean = plan match {
+ case sHolder: ScanBuilderHolder =>
+ val isPushed = PushDownUtils.pushOffset(sHolder.builder, offset)
+ if (isPushed) {
+ sHolder.pushedOffset = Some(offset)
+ }
+ isPushed
+ case p: Project =>
+ pushDownOffset(p.child, offset)
+ case _ => false
+ }
+
+ def pushDownLimitAndOffset(plan: LogicalPlan): LogicalPlan = plan.transform {
+ case offset @ LimitAndOffset(limit, offsetValue, child) =>
+ val (newChild, canRemoveLimit) = pushDownLimit(child, limit)
+ if (canRemoveLimit) {
+ // If we can remove limit, it indicates data source only have one partition.
+ // For `dataset.limit(m).offset(n)`, try to push down `offset(n).limit(m - n)`.
+ // For example, `dataset.limit(5).offset(3)`, we can push down `offset(3).limit(2)`.
+ val isPushed = pushDownOffset(newChild, offsetValue)
+ if (isPushed) {
+ newChild
+ } else {
+ // For `dataset.limit(m).offset(n)`, only push down `limit(m)`.
+ // Spark still do `offset(n)`.
+ offset.withNewChildren(Seq(newChild))
+ }
+ } else {
+ // If we can't push down limit and offset, return `Offset`.
Review Comment:
How about `If we can't remove `Limit` and push down offset, return `Offset`.` ?
--
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
[GitHub] [spark] cloud-fan commented on a diff in pull request #36295: [SPARK-38978][SQL] DS V2 supports push down OFFSET operator
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #36295:
URL: https://github.com/apache/spark/pull/36295#discussion_r902312467
##########
sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCV2Suite.scala:
##########
@@ -206,6 +207,355 @@ class JDBCV2Suite extends QueryTest with SharedSparkSession with ExplainSuiteHel
checkAnswer(df5, Seq(Row(10000.00, 1000.0, "amy")))
}
+ private def checkOffsetRemoved(df: DataFrame, removed: Boolean = true): Unit = {
+ val offsets = df.queryExecution.optimizedPlan.collect {
+ case offset: Offset => offset
+ }
+ if (removed) {
+ assert(offsets.isEmpty)
+ } else {
+ assert(offsets.nonEmpty)
+ }
+ }
+
+ test("simple scan with OFFSET") {
+ val df1 = spark.read
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .offset(1)
+ checkOffsetRemoved(df1)
+ checkPushedInfo(df1,
+ "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], PushedOffset: OFFSET 1,")
+ checkAnswer(df1, Seq(Row(1, "cathy", 9000.00, 1200.0, false)))
+
+ val df2 = spark.read
+ .option("pushDownOffset", "false")
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .offset(1)
+ checkOffsetRemoved(df2, false)
+ checkPushedInfo(df2,
+ "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], ReadSchema:")
+ checkAnswer(df2, Seq(Row(1, "cathy", 9000.00, 1200.0, false)))
+
+ val df3 = spark.read
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .sort($"salary")
+ .offset(1)
+ checkOffsetRemoved(df3, false)
+ checkPushedInfo(df3,
+ "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], ReadSchema:")
+ checkAnswer(df3, Seq(Row(1, "amy", 10000.00, 1000.0, true)))
+
+ val df4 = spark.read
+ .option("partitionColumn", "dept")
+ .option("lowerBound", "0")
+ .option("upperBound", "2")
+ .option("numPartitions", "2")
+ .table("h2.test.employee")
+ .filter($"dept" > 1)
+ .offset(1)
+ checkOffsetRemoved(df4, false)
+ checkPushedInfo(df4, "PushedFilters: [DEPT IS NOT NULL, DEPT > 1], ReadSchema:")
+ checkAnswer(df4, Seq(Row(2, "david", 10000, 1300, true), Row(6, "jen", 12000, 1200, true)))
+
+ val df5 = spark.read
+ .table("h2.test.employee")
+ .groupBy("DEPT").sum("SALARY")
+ .offset(1)
+ checkOffsetRemoved(df5, false)
+ checkPushedInfo(df5,
+ "PushedAggregates: [SUM(SALARY)], PushedFilters: [], PushedGroupByExpressions: [DEPT], ")
+ checkAnswer(df5, Seq(Row(2, 22000.00), Row(6, 12000.00)))
+
+ val name = udf { (x: String) => x.matches("cat|dav|amy") }
+ val sub = udf { (x: String) => x.substring(0, 3) }
+ val df6 = spark.read
+ .table("h2.test.employee")
+ .select($"SALARY", $"BONUS", sub($"NAME").as("shortName"))
+ .filter(name($"shortName"))
+ .offset(1)
+ checkOffsetRemoved(df6, false)
+ // OFFSET is pushed down only if all the filters are pushed down
+ checkPushedInfo(df6, "PushedFilters: [], ")
+ checkAnswer(df6, Seq(Row(10000.00, 1300.0, "dav"), Row(9000.00, 1200.0, "cat")))
+ }
+
+ test("simple scan with LIMIT and OFFSET") {
+ val df1 = spark.read
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .limit(2)
+ .offset(1)
+ checkLimitRemoved(df1)
+ checkOffsetRemoved(df1)
+ checkPushedInfo(df1,
+ "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], PushedLimit: LIMIT 2, PushedOffset: OFFSET 1,")
+ checkAnswer(df1, Seq(Row(1, "cathy", 9000.00, 1200.0, false)))
+
+ val df2 = spark.read
+ .option("pushDownLimit", "false")
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .limit(2)
+ .offset(1)
+ checkLimitRemoved(df2, false)
+ checkOffsetRemoved(df2, false)
+ checkPushedInfo(df2,
+ "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], ReadSchema:")
+ checkAnswer(df2, Seq(Row(1, "cathy", 9000.00, 1200.0, false)))
+
+ val df3 = spark.read
+ .option("pushDownOffset", "false")
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .limit(2)
+ .offset(1)
+ checkLimitRemoved(df3)
+ checkOffsetRemoved(df3, false)
+ checkPushedInfo(df3,
+ "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], PushedLimit: LIMIT 2, ReadSchema:")
+ checkAnswer(df3, Seq(Row(1, "cathy", 9000.00, 1200.0, false)))
+
+ val df4 = spark.read
+ .option("pushDownLimit", "false")
+ .option("pushDownOffset", "false")
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .limit(2)
+ .offset(1)
+ checkLimitRemoved(df4, false)
+ checkOffsetRemoved(df4, false)
+ checkPushedInfo(df4,
+ "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], ReadSchema:")
+ checkAnswer(df4, Seq(Row(1, "cathy", 9000.00, 1200.0, false)))
+
+ val df5 = spark.read
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .sort($"salary")
+ .limit(2)
+ .offset(1)
+ checkLimitRemoved(df5)
+ checkOffsetRemoved(df5)
+ checkPushedInfo(df5, "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], " +
+ "PushedOffset: OFFSET 1, PushedTopN: ORDER BY [SALARY ASC NULLS FIRST] LIMIT 2, ReadSchema:")
+ checkAnswer(df5, Seq(Row(1, "amy", 10000.00, 1000.0, true)))
+
+ val df6 = spark.read
+ .option("pushDownLimit", "false")
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .sort($"salary")
+ .limit(2)
+ .offset(1)
+ checkLimitRemoved(df6, false)
+ checkOffsetRemoved(df6, false)
+ checkPushedInfo(df6, "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], ReadSchema:")
+ checkAnswer(df6, Seq(Row(1, "amy", 10000.00, 1000.0, true)))
+
+ val df7 = spark.read
+ .option("pushDownOffset", "false")
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .sort($"salary")
+ .limit(2)
+ .offset(1)
+ checkLimitRemoved(df7)
+ checkOffsetRemoved(df7, false)
+ checkPushedInfo(df7, "PushedFilters: [DEPT IS NOT NULL, DEPT = 1]," +
+ " PushedTopN: ORDER BY [SALARY ASC NULLS FIRST] LIMIT 2, ReadSchema:")
+ checkAnswer(df7, Seq(Row(1, "amy", 10000.00, 1000.0, true)))
+
+ val df8 = spark.read
+ .option("pushDownLimit", "false")
+ .option("pushDownOffset", "false")
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .sort($"salary")
+ .limit(2)
+ .offset(1)
+ checkLimitRemoved(df8, false)
+ checkOffsetRemoved(df8, false)
+ checkPushedInfo(df8, "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], ReadSchema:")
+ checkAnswer(df8, Seq(Row(1, "amy", 10000.00, 1000.0, true)))
+
+ val df9 = spark.read
+ .option("partitionColumn", "dept")
+ .option("lowerBound", "0")
+ .option("upperBound", "2")
+ .option("numPartitions", "2")
+ .table("h2.test.employee")
+ .filter($"dept" > 1)
+ .limit(2)
+ .offset(1)
+ checkLimitRemoved(df9, false)
+ checkOffsetRemoved(df9, false)
+ checkPushedInfo(df9,
+ "PushedFilters: [DEPT IS NOT NULL, DEPT > 1], PushedLimit: LIMIT 2, ReadSchema:")
+ checkAnswer(df9, Seq(Row(2, "david", 10000.00, 1300.0, true)))
+
+ val df10 = spark.read
+ .table("h2.test.employee")
+ .groupBy("DEPT").sum("SALARY")
+ .limit(2)
+ .offset(1)
+ checkLimitRemoved(df10, false)
+ checkOffsetRemoved(df10, false)
+ checkPushedInfo(df10,
+ "PushedAggregates: [SUM(SALARY)], PushedFilters: [], PushedGroupByExpressions: [DEPT], ")
+ checkAnswer(df10, Seq(Row(2, 22000.00)))
+
+ val name = udf { (x: String) => x.matches("cat|dav|amy") }
+ val sub = udf { (x: String) => x.substring(0, 3) }
+ val df11 = spark.read
+ .table("h2.test.employee")
+ .select($"SALARY", $"BONUS", sub($"NAME").as("shortName"))
+ .filter(name($"shortName"))
+ .limit(2)
+ .offset(1)
+ checkLimitRemoved(df11, false)
+ checkOffsetRemoved(df11, false)
+ checkPushedInfo(df11, "PushedFilters: [], ")
+ checkAnswer(df11, Seq(Row(9000.00, 1200.0, "cat")))
+ }
+
+ test("simple scan with OFFSET and LIMIT") {
+ val df1 = spark.read
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .offset(1)
+ .limit(1)
+ checkLimitRemoved(df1)
+ checkOffsetRemoved(df1)
+ checkPushedInfo(df1,
+ "[DEPT IS NOT NULL, DEPT = 1], PushedLimit: LIMIT 2, PushedOffset: OFFSET 1,")
+ checkAnswer(df1, Seq(Row(1, "cathy", 9000.00, 1200.0, false)))
+
+ val df2 = spark.read
+ .option("pushDownOffset", "false")
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .offset(1)
+ .limit(1)
+ checkLimitRemoved(df2, false)
Review Comment:
I think we should still push down limit in this case.
--
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
[GitHub] [spark] beliefer commented on a diff in pull request #36295: [SPARK-38978][SQL] Support push down OFFSET to JDBC data source V2
Posted by GitBox <gi...@apache.org>.
beliefer commented on code in PR #36295:
URL: https://github.com/apache/spark/pull/36295#discussion_r870998281
##########
sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala:
##########
@@ -2102,6 +2102,16 @@ class Dataset[T] private[sql](
Limit(Literal(n), logicalPlan)
}
+ /**
+ * Returns a new Dataset by skipping the first `m` rows.
+ *
+ * @group typedrel
+ * @since 3.4.0
+ */
+ def offset(n: Int): Dataset[T] = withTypedPlan {
Review Comment:
OK
--
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
[GitHub] [spark] beliefer commented on a diff in pull request #36295: [SPARK-38978][SQL] Support push down OFFSET to JDBC data source V2
Posted by GitBox <gi...@apache.org>.
beliefer commented on code in PR #36295:
URL: https://github.com/apache/spark/pull/36295#discussion_r883240658
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/DataSourceScanExec.scala:
##########
@@ -143,13 +143,15 @@ case class RowDataSourceScanExec(
val topNOrLimitInfo =
Review Comment:
I see.
--
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
[GitHub] [spark] cloud-fan commented on a diff in pull request #36295: [SPARK-38978][SQL] Support push down OFFSET to JDBC data source V2
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #36295:
URL: https://github.com/apache/spark/pull/36295#discussion_r882685206
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala:
##########
@@ -407,8 +407,65 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper wit
case other => (other, false)
}
- def pushDownLimits(plan: LogicalPlan): LogicalPlan = plan.transform {
+ private def pushDownOffset(
+ plan: LogicalPlan,
+ offset: Int): Boolean = plan match {
+ case sHolder: ScanBuilderHolder =>
+ val isPushed = PushDownUtils.pushOffset(sHolder.builder, offset)
+ if (isPushed) {
+ sHolder.pushedOffset = Some(offset)
+ }
+ isPushed
+ case p: Project =>
+ pushDownOffset(p.child, offset)
+ case _ => false
+ }
+
+ def pushDownLimitAndOffset(plan: LogicalPlan): LogicalPlan = plan.transform {
+ case offset @ LimitAndOffset(limit, offsetValue, child) =>
+ val (newChild, canRemoveLimit) = pushDownLimit(child, limit)
+ if (canRemoveLimit) {
+ // If we can remove limit, it indicates data source only have one partition.
+ // For `dataset.limit(m).offset(n)`, try to push down `LIMIT (m - n) OFFSET n`.
+ // For example, `dataset.limit(5).offset(3)`, we can push down `LIMIT 2 OFFSET 3`.
Review Comment:
This comment makes no sense here. We are pushing down operators, not pushing down a SQL query to JDBC. It's only a problem in JDBC that `LIMIT x OFFSET y` means doing OFFSET first, then LIMIT
--
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
[GitHub] [spark] cloud-fan commented on a diff in pull request #36295: [SPARK-38978][SQL] Support push down OFFSET to JDBC data source V2
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #36295:
URL: https://github.com/apache/spark/pull/36295#discussion_r882304055
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/DataSourceScanExec.scala:
##########
@@ -141,15 +141,20 @@ case class RowDataSourceScanExec(
handledFilters
}
- val topNOrLimitInfo =
+ val limitOrOffsetInfo =
if (pushedDownOperators.limit.isDefined && pushedDownOperators.sortValues.nonEmpty) {
- val pushedTopN =
+ val topNStr =
s"ORDER BY ${seqToString(pushedDownOperators.sortValues.map(_.describe()))}" +
s" LIMIT ${pushedDownOperators.limit.get}"
- Some("PushedTopN" -> pushedTopN)
- } else {
- pushedDownOperators.limit.map(value => "PushedLimit" -> s"LIMIT $value")
- }
+ if (pushedDownOperators.offset.isDefined) {
+ Map("PushedPaging" -> s"$topNStr OFFSET ${pushedDownOperators.offset.get}")
Review Comment:
do we need to combine them into one metrics? I think we can have `PushedTopN` and `PushedOffset` together in the metrics.
--
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
[GitHub] [spark] cloud-fan commented on a diff in pull request #36295: [SPARK-38978][SQL] Support push down OFFSET to JDBC data source V2
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #36295:
URL: https://github.com/apache/spark/pull/36295#discussion_r882695556
##########
sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCV2Suite.scala:
##########
@@ -203,6 +204,355 @@ class JDBCV2Suite extends QueryTest with SharedSparkSession with ExplainSuiteHel
checkAnswer(df5, Seq(Row(10000.00, 1000.0, "amy")))
}
+ private def checkOffsetRemoved(df: DataFrame, removed: Boolean = true): Unit = {
+ val offsets = df.queryExecution.optimizedPlan.collect {
+ case offset: Offset => offset
+ }
+ if (removed) {
+ assert(offsets.isEmpty)
+ } else {
+ assert(offsets.nonEmpty)
+ }
+ }
+
+ test("simple scan with OFFSET") {
+ val df1 = spark.read
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .offset(1)
+ checkOffsetRemoved(df1)
+ checkPushedInfo(df1,
+ "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], PushedOffset: OFFSET 1,")
+ checkAnswer(df1, Seq(Row(1, "cathy", 9000.00, 1200.0, false)))
+
+ val df2 = spark.read
+ .option("pushDownOffset", "false")
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .offset(1)
+ checkOffsetRemoved(df2, false)
+ checkPushedInfo(df2,
+ "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], ReadSchema:")
+ checkAnswer(df2, Seq(Row(1, "cathy", 9000.00, 1200.0, false)))
+
+ val df3 = spark.read
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .sort($"salary")
+ .offset(1)
+ checkOffsetRemoved(df3, false)
+ checkPushedInfo(df3,
+ "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], ReadSchema:")
+ checkAnswer(df3, Seq(Row(1, "amy", 10000.00, 1000.0, true)))
+
+ val df4 = spark.read
+ .option("partitionColumn", "dept")
+ .option("lowerBound", "0")
+ .option("upperBound", "2")
+ .option("numPartitions", "2")
+ .table("h2.test.employee")
+ .filter($"dept" > 1)
+ .offset(1)
+ checkOffsetRemoved(df4, false)
+ checkPushedInfo(df4, "PushedFilters: [DEPT IS NOT NULL, DEPT > 1], ReadSchema:")
+ checkAnswer(df4, Seq(Row(2, "david", 10000, 1300, true), Row(6, "jen", 12000, 1200, true)))
+
+ val df5 = spark.read
+ .table("h2.test.employee")
+ .groupBy("DEPT").sum("SALARY")
+ .offset(1)
+ checkOffsetRemoved(df5, false)
+ checkPushedInfo(df5,
+ "PushedAggregates: [SUM(SALARY)], PushedFilters: [], PushedGroupByExpressions: [DEPT], ")
+ checkAnswer(df5, Seq(Row(2, 22000.00), Row(6, 12000.00)))
+
+ val name = udf { (x: String) => x.matches("cat|dav|amy") }
+ val sub = udf { (x: String) => x.substring(0, 3) }
+ val df6 = spark.read
+ .table("h2.test.employee")
+ .select($"SALARY", $"BONUS", sub($"NAME").as("shortName"))
+ .filter(name($"shortName"))
+ .offset(1)
+ checkOffsetRemoved(df6, false)
+ // OFFSET is pushed down only if all the filters are pushed down
+ checkPushedInfo(df6, "PushedFilters: [], ")
+ checkAnswer(df6, Seq(Row(10000.00, 1300.0, "dav"), Row(9000.00, 1200.0, "cat")))
+ }
+
+ test("simple scan with LIMIT and OFFSET") {
+ val df1 = spark.read
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .limit(2)
+ .offset(1)
+ checkLimitRemoved(df1)
+ checkOffsetRemoved(df1)
+ checkPushedInfo(df1,
+ "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], PushedLimit: LIMIT 2, PushedOffset: OFFSET 1,")
+ checkAnswer(df1, Seq(Row(1, "cathy", 9000.00, 1200.0, false)))
+
+ val df2 = spark.read
+ .option("pushDownLimit", "false")
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .limit(2)
+ .offset(1)
+ checkLimitRemoved(df2, false)
+ checkOffsetRemoved(df2, false)
+ checkPushedInfo(df2,
+ "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], ReadSchema:")
+ checkAnswer(df2, Seq(Row(1, "cathy", 9000.00, 1200.0, false)))
+
+ val df3 = spark.read
+ .option("pushDownOffset", "false")
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .limit(2)
+ .offset(1)
+ checkLimitRemoved(df3)
+ checkOffsetRemoved(df3, false)
+ checkPushedInfo(df3,
+ "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], PushedLimit: LIMIT 2, ReadSchema:")
+ checkAnswer(df3, Seq(Row(1, "cathy", 9000.00, 1200.0, false)))
+
+ val df4 = spark.read
+ .option("pushDownLimit", "false")
+ .option("pushDownOffset", "false")
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .limit(2)
+ .offset(1)
+ checkLimitRemoved(df4, false)
+ checkOffsetRemoved(df4, false)
+ checkPushedInfo(df4,
+ "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], ReadSchema:")
+ checkAnswer(df4, Seq(Row(1, "cathy", 9000.00, 1200.0, false)))
+
+ val df5 = spark.read
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .sort($"salary")
+ .limit(2)
+ .offset(1)
+ checkLimitRemoved(df5)
+ checkOffsetRemoved(df5)
+ checkPushedInfo(df5, "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], " +
+ "PushedOffset: OFFSET 1, PushedTopN: ORDER BY [SALARY ASC NULLS FIRST] LIMIT 2, ReadSchema:")
+ checkAnswer(df5, Seq(Row(1, "amy", 10000.00, 1000.0, true)))
+
+ val df6 = spark.read
+ .option("pushDownLimit", "false")
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .sort($"salary")
+ .limit(2)
+ .offset(1)
+ checkLimitRemoved(df6, false)
+ checkOffsetRemoved(df6, false)
+ checkPushedInfo(df6, "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], ReadSchema:")
+ checkAnswer(df6, Seq(Row(1, "amy", 10000.00, 1000.0, true)))
+
+ val df7 = spark.read
+ .option("pushDownOffset", "false")
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .sort($"salary")
+ .limit(2)
+ .offset(1)
+ checkLimitRemoved(df7)
+ checkOffsetRemoved(df7, false)
+ checkPushedInfo(df7, "PushedFilters: [DEPT IS NOT NULL, DEPT = 1]," +
+ " PushedTopN: ORDER BY [SALARY ASC NULLS FIRST] LIMIT 2, ReadSchema:")
+ checkAnswer(df7, Seq(Row(1, "amy", 10000.00, 1000.0, true)))
+
+ val df8 = spark.read
+ .option("pushDownLimit", "false")
+ .option("pushDownOffset", "false")
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .sort($"salary")
+ .limit(2)
+ .offset(1)
+ checkLimitRemoved(df8, false)
+ checkOffsetRemoved(df8, false)
+ checkPushedInfo(df8, "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], ReadSchema:")
+ checkAnswer(df8, Seq(Row(1, "amy", 10000.00, 1000.0, true)))
+
+ val df9 = spark.read
+ .option("partitionColumn", "dept")
+ .option("lowerBound", "0")
+ .option("upperBound", "2")
+ .option("numPartitions", "2")
+ .table("h2.test.employee")
+ .filter($"dept" > 1)
+ .limit(2)
+ .offset(1)
+ checkLimitRemoved(df9, false)
+ checkOffsetRemoved(df9, false)
+ checkPushedInfo(df9,
+ "PushedFilters: [DEPT IS NOT NULL, DEPT > 1], PushedLimit: LIMIT 2, ReadSchema:")
+ checkAnswer(df9, Seq(Row(2, "david", 10000.00, 1300.0, true)))
+
+ val df10 = spark.read
+ .table("h2.test.employee")
+ .groupBy("DEPT").sum("SALARY")
+ .limit(2)
+ .offset(1)
+ checkLimitRemoved(df10, false)
+ checkOffsetRemoved(df10, false)
+ checkPushedInfo(df10,
+ "PushedAggregates: [SUM(SALARY)], PushedFilters: [], PushedGroupByExpressions: [DEPT], ")
+ checkAnswer(df10, Seq(Row(2, 22000.00)))
+
+ val name = udf { (x: String) => x.matches("cat|dav|amy") }
+ val sub = udf { (x: String) => x.substring(0, 3) }
+ val df11 = spark.read
+ .table("h2.test.employee")
+ .select($"SALARY", $"BONUS", sub($"NAME").as("shortName"))
+ .filter(name($"shortName"))
+ .limit(2)
+ .offset(1)
+ checkLimitRemoved(df11, false)
+ checkOffsetRemoved(df11, false)
+ checkPushedInfo(df11, "PushedFilters: [], ")
+ checkAnswer(df11, Seq(Row(9000.00, 1200.0, "cat")))
+ }
+
+ test("simple scan with OFFSET and LIMIT") {
+ val df1 = spark.read
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .offset(1)
+ .limit(1)
+ checkLimitRemoved(df1)
+ checkOffsetRemoved(df1)
+ checkPushedInfo(df1,
+ "[DEPT IS NOT NULL, DEPT = 1], PushedLimit: LIMIT 1, PushedOffset: OFFSET 1,")
+ checkAnswer(df1, Seq(Row(1, "cathy", 9000.00, 1200.0, false)))
+
+ val df2 = spark.read
+ .option("pushDownOffset", "false")
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .offset(1)
+ .limit(1)
+ checkLimitRemoved(df2, false)
+ checkOffsetRemoved(df2, false)
+ checkPushedInfo(df2,
+ "[DEPT IS NOT NULL, DEPT = 1], PushedLimit: LIMIT 2, ReadSchema:")
+ checkAnswer(df2, Seq(Row(1, "cathy", 9000.00, 1200.0, false)))
+
+ val df3 = spark.read
+ .option("pushDownLimit", "false")
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .offset(1)
+ .limit(1)
+ checkLimitRemoved(df3, false)
+ checkOffsetRemoved(df3)
+ checkPushedInfo(df3,
+ "[DEPT IS NOT NULL, DEPT = 1], PushedOffset: OFFSET 1, ReadSchema:")
+ checkAnswer(df3, Seq(Row(1, "cathy", 9000.00, 1200.0, false)))
+
+ val df4 = spark.read
+ .option("pushDownOffset", "false")
+ .option("pushDownLimit", "false")
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .offset(1)
+ .limit(1)
+ checkLimitRemoved(df4, false)
+ checkOffsetRemoved(df4, false)
+ checkPushedInfo(df4,
+ "[DEPT IS NOT NULL, DEPT = 1], ReadSchema:")
+ checkAnswer(df4, Seq(Row(1, "cathy", 9000.00, 1200.0, false)))
+
+ val df5 = spark.read
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .sort($"salary")
+ .offset(1)
+ .limit(1)
+ checkLimitRemoved(df5, false)
+ checkOffsetRemoved(df5)
+ checkPushedInfo(df5, "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], " +
+ "PushedOffset: OFFSET 1, PushedTopN: ORDER BY [SALARY ASC NULLS FIRST] LIMIT 2, ReadSchema:")
+ checkAnswer(df5, Seq(Row(1, "amy", 10000.00, 1000.0, true)))
+
+ val df6 = spark.read
+ .option("pushDownOffset", "false")
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .sort($"salary")
+ .offset(1)
+ .limit(1)
+ checkLimitRemoved(df6, false)
+ checkOffsetRemoved(df6, false)
+ checkPushedInfo(df6, "[DEPT IS NOT NULL, DEPT = 1]," +
+ " PushedTopN: ORDER BY [SALARY ASC NULLS FIRST] LIMIT 2, ReadSchema:")
+ checkAnswer(df6, Seq(Row(1, "amy", 10000.00, 1000.0, true)))
+
+ val df7 = spark.read
+ .option("pushDownLimit", "false")
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .sort($"salary")
+ .offset(1)
+ .limit(1)
+ checkLimitRemoved(df7, false)
+ checkOffsetRemoved(df7, false)
+ checkPushedInfo(df7, "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], ReadSchema:")
+ checkAnswer(df7, Seq(Row(1, "amy", 10000.00, 1000.0, true)))
+
+ val df8 = spark.read
+ .option("pushDownOffset", "false")
+ .option("pushDownLimit", "false")
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .sort($"salary")
+ .offset(1)
+ .limit(1)
+ checkLimitRemoved(df8, false)
+ checkOffsetRemoved(df8, false)
+ checkPushedInfo(df8, "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], ReadSchema:")
+ checkAnswer(df8, Seq(Row(1, "amy", 10000.00, 1000.0, true)))
+
+ val df9 = spark.read
+ .option("partitionColumn", "dept")
+ .option("lowerBound", "0")
+ .option("upperBound", "2")
+ .option("numPartitions", "2")
+ .table("h2.test.employee")
+ .filter($"dept" > 1)
+ .offset(1)
+ .limit(1)
+ checkLimitRemoved(df9, false)
+ checkOffsetRemoved(df9, false)
+ checkPushedInfo(df9,
+ "PushedFilters: [DEPT IS NOT NULL, DEPT > 1], PushedLimit: LIMIT 2, ReadSchema:")
+ checkAnswer(df9, Seq(Row(2, "david", 10000.00, 1300.0, true)))
+
+ val df10 = sql("SELECT dept, sum(salary) FROM h2.test.employee group by dept LIMIT 1 OFFSET 1")
Review Comment:
ditto
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] cloud-fan commented on a diff in pull request #36295: [SPARK-38978][SQL] Support push down OFFSET to JDBC data source V2
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #36295:
URL: https://github.com/apache/spark/pull/36295#discussion_r880684191
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/DataSourceScanExec.scala:
##########
@@ -141,15 +141,23 @@ case class RowDataSourceScanExec(
handledFilters
}
- val topNOrLimitInfo =
+ val limitOrOffsetInfo =
if (pushedDownOperators.limit.isDefined && pushedDownOperators.sortValues.nonEmpty) {
- val pushedTopN =
- s"ORDER BY ${seqToString(pushedDownOperators.sortValues.map(_.describe()))}" +
- s" LIMIT ${pushedDownOperators.limit.get}"
- Some("PushedTopN" -> pushedTopN)
- } else {
- pushedDownOperators.limit.map(value => "PushedLimit" -> s"LIMIT $value")
- }
+ if (pushedDownOperators.offset.isDefined) {
Review Comment:
nit:
```
val topNStr = "ORDER BY ... LIMIT ..."
if (pushedDownOperators.offset.isDefined) {
$"topNStr OFFSET ..."
} else {
topNStr
}
```
--
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
[GitHub] [spark] cloud-fan commented on a diff in pull request #36295: [SPARK-38978][SQL] Support push down limit and offset to JDBC data source V2
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #36295:
URL: https://github.com/apache/spark/pull/36295#discussion_r855354481
##########
sql/catalyst/src/main/java/org/apache/spark/sql/connector/read/SupportsPushDownLimitAndOffset.java:
##########
@@ -0,0 +1,43 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.connector.read;
+
+import org.apache.spark.annotation.Evolving;
+
+/**
+ * A mix-in interface for {@link ScanBuilder}. Data sources can implement this interface to
+ * push down LIMIT and OFFSET. Please note that the combination of LIMIT and OFFSET with other
+ * operations such as AGGREGATE, GROUP BY, SORT BY, CLUSTER BY, DISTRIBUTE BY, etc.
+ * is NOT pushed down.
+ *
+ * @since 3.4.0
+ */
+@Evolving
+public interface SupportsPushDownLimitAndOffset {
Review Comment:
I have a different opinion about this API design. For top n pushdown, the rationale is that we shouldn't pushdown sort, but the data source should report its ordering, and pushing top n is more reasonable. For offset, the current limitation is it must be used together with limit, but this can be changed in the future. I don't see why can't we pushdown offset along.
--
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
[GitHub] [spark] beliefer commented on pull request #36295: [SPARK-38978][SQL] Support push down OFFSET to JDBC data source V2
Posted by GitBox <gi...@apache.org>.
beliefer commented on PR #36295:
URL: https://github.com/apache/spark/pull/36295#issuecomment-1123506746
ping @huaxingao cc @cloud-fan
--
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
[GitHub] [spark] beliefer commented on pull request #36295: [SPARK-38978][SQL] Support push down limit and offset to JDBC data source V2
Posted by GitBox <gi...@apache.org>.
beliefer commented on PR #36295:
URL: https://github.com/apache/spark/pull/36295#issuecomment-1104929922
ping @huaxingao cc @cloud-fan
--
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
[GitHub] [spark] beliefer commented on a diff in pull request #36295: [SPARK-38978][SQL] DS V2 supports push down OFFSET operator
Posted by GitBox <gi...@apache.org>.
beliefer commented on code in PR #36295:
URL: https://github.com/apache/spark/pull/36295#discussion_r902391544
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala:
##########
@@ -407,8 +407,69 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper wit
case other => (other, false)
}
- def pushDownLimits(plan: LogicalPlan): LogicalPlan = plan.transform {
+ private def pushDownOffset(
+ plan: LogicalPlan,
+ offset: Int): Boolean = plan match {
+ case sHolder: ScanBuilderHolder =>
+ val isPushed = PushDownUtils.pushOffset(sHolder.builder, offset)
+ if (isPushed) {
+ sHolder.pushedOffset = Some(offset)
+ }
+ isPushed
+ case p: Project =>
+ pushDownOffset(p.child, offset)
+ case _ => false
+ }
+
+ def pushDownLimitAndOffset(plan: LogicalPlan): LogicalPlan = plan.transform {
+ case offset @ LimitAndOffset(limit, offsetValue, child) =>
+ val (newChild, canRemoveLimit) = pushDownLimit(child, limit)
+ if (canRemoveLimit) {
+ // If we can remove limit, it indicates data source only have one partition.
+ // For `dataset.limit(m).offset(n)`, try to push down `offset(n).limit(m - n)`.
+ // For example, `dataset.limit(5).offset(3)`, we can push down `offset(3).limit(2)`.
+ val isPushed = pushDownOffset(newChild, offsetValue)
+ if (isPushed) {
+ newChild
+ } else {
+ // For `dataset.limit(m).offset(n)`, only push down `limit(m)`.
+ // Spark still do `offset(n)`.
Review Comment:
OK
--
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
[GitHub] [spark] cloud-fan commented on a diff in pull request #36295: [SPARK-38978][SQL] Support push down OFFSET to JDBC data source V2
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #36295:
URL: https://github.com/apache/spark/pull/36295#discussion_r870959171
##########
sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala:
##########
@@ -2102,6 +2102,16 @@ class Dataset[T] private[sql](
Limit(Literal(n), logicalPlan)
}
+ /**
+ * Returns a new Dataset by skipping the first `m` rows.
+ *
+ * @group typedrel
+ * @since 3.4.0
+ */
+ def offset(n: Int): Dataset[T] = withTypedPlan {
Review Comment:
please open a separate PR for it. This is adding a new user-facing API and needs more visibility.
--
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
[GitHub] [spark] cloud-fan commented on a diff in pull request #36295: [SPARK-38978][SQL] Support push down OFFSET to JDBC data source V2
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #36295:
URL: https://github.com/apache/spark/pull/36295#discussion_r870959719
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala:
##########
@@ -44,6 +44,7 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper wit
pushDownFilters,
pushDownAggregates,
pushDownLimits,
+ pushDownOffsets,
Review Comment:
shall we push down offset before limit? The natural operators order in SQL is offset then limit.
--
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
[GitHub] [spark] cloud-fan commented on a diff in pull request #36295: [SPARK-38978][SQL] Support push down OFFSET to JDBC data source V2
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #36295:
URL: https://github.com/apache/spark/pull/36295#discussion_r870958697
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/CheckAnalysis.scala:
##########
@@ -433,13 +433,6 @@ trait CheckAnalysis extends PredicateHelper with LookupCatalog {
case Offset(offsetExpr, _) => checkLimitLikeClause("offset", offsetExpr)
- case o if !o.isInstanceOf[GlobalLimit] && !o.isInstanceOf[LocalLimit]
Review Comment:
can we have a separate PR for this?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] beliefer commented on a diff in pull request #36295: [SPARK-38978][SQL] Support push down OFFSET to JDBC data source V2
Posted by GitBox <gi...@apache.org>.
beliefer commented on code in PR #36295:
URL: https://github.com/apache/spark/pull/36295#discussion_r882541698
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala:
##########
@@ -44,6 +44,7 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper wit
pushDownFilters,
pushDownAggregates,
pushDownLimits,
Review Comment:
OK
--
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
[GitHub] [spark] cloud-fan commented on a diff in pull request #36295: [SPARK-38978][SQL] Support push down OFFSET to JDBC data source V2
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #36295:
URL: https://github.com/apache/spark/pull/36295#discussion_r882306593
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala:
##########
@@ -419,6 +420,60 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper wit
}
}
+ private def pushDownOffset(
+ plan: LogicalPlan,
+ offset: Int): (LogicalPlan, Boolean) = plan match {
+ case operation @ ScanOperation(_, filter, sHolder: ScanBuilderHolder) if filter.isEmpty =>
+ val isPushed = PushDownUtils.pushOffset(sHolder.builder, offset)
+ if (isPushed) {
+ sHolder.pushedOffset = Some(offset)
+ }
+ (operation, isPushed)
+ case p: Project =>
+ val (newChild, isPushed) = pushDownOffset(p.child, offset)
+ (p.withNewChildren(Seq(newChild)), isPushed)
+ case other => (other, false)
+ }
+
+ def pushDownLimitAndOffset(plan: LogicalPlan): LogicalPlan = plan.transform {
+ case offset @ Offset(IntegerLiteral(n), child) =>
+ // For `dataset.offset(n)`, try to push down `OFFSET n`.
+ val (newChild, isPushed) = pushDownOffset(child, n)
+ if (isPushed) {
+ newChild
+ } else {
+ offset
+ }
Review Comment:
```suggestion
private def pushDownOffset(
plan: LogicalPlan,
offset: Int): Boolean = plan match {
case sHolder: ScanBuilderHolder =>
val isPushed = PushDownUtils.pushOffset(sHolder.builder, offset)
if (isPushed) {
sHolder.pushedOffset = Some(offset)
}
isPushed
case p: Project =>
pushDownOffset(p.child, offset)
case _ => false
}
def pushDownLimitAndOffset(plan: LogicalPlan): LogicalPlan = plan.transform {
case offset @ Offset(IntegerLiteral(n), child) =>
// For `dataset.offset(n)`, try to push down `OFFSET n`.
val isPushed = pushDownOffset(child, n)
if (isPushed) {
child
} else {
offset
}
```
--
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
[GitHub] [spark] cloud-fan commented on a diff in pull request #36295: [SPARK-38978][SQL] Support push down OFFSET to JDBC data source V2
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #36295:
URL: https://github.com/apache/spark/pull/36295#discussion_r882304320
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala:
##########
@@ -44,6 +44,7 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper wit
pushDownFilters,
pushDownAggregates,
pushDownLimits,
Review Comment:
why do we still keep it?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] cloud-fan commented on a diff in pull request #36295: [SPARK-38978][SQL] DS V2 supports push down OFFSET operator
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #36295:
URL: https://github.com/apache/spark/pull/36295#discussion_r903769471
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala:
##########
@@ -407,7 +407,66 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper wit
case other => (other, false)
}
- def pushDownLimits(plan: LogicalPlan): LogicalPlan = plan.transform {
+ private def pushDownOffset(
+ plan: LogicalPlan,
+ offset: Int): Boolean = plan match {
+ case sHolder: ScanBuilderHolder =>
+ val isPushed = PushDownUtils.pushOffset(sHolder.builder, offset)
+ if (isPushed) {
+ sHolder.pushedOffset = Some(offset)
+ }
+ isPushed
+ case p: Project =>
Review Comment:
If the project has non-deterministic expressions, we should not push down offset as it may change the query result. e.g. producing 10 rows, doing a project and skip the first 5 rows can be different from producing 5 rows and doing a project.
--
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
[GitHub] [spark] cloud-fan commented on a diff in pull request #36295: [SPARK-38978][SQL] DS V2 supports push down OFFSET operator
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #36295:
URL: https://github.com/apache/spark/pull/36295#discussion_r903807231
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala:
##########
@@ -407,7 +407,66 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper wit
case other => (other, false)
}
- def pushDownLimits(plan: LogicalPlan): LogicalPlan = plan.transform {
+ private def pushDownOffset(
+ plan: LogicalPlan,
+ offset: Int): Boolean = plan match {
+ case sHolder: ScanBuilderHolder =>
+ val isPushed = PushDownUtils.pushOffset(sHolder.builder, offset)
+ if (isPushed) {
+ sHolder.pushedOffset = Some(offset)
+ }
+ isPushed
+ case p: Project =>
+ pushDownOffset(p.child, offset)
+ case _ => false
+ }
+
+ def pushDownLimitAndOffset(plan: LogicalPlan): LogicalPlan = plan.transform {
+ case offset @ LimitAndOffset(limit, offsetValue, child) =>
+ val (newChild, canRemoveLimit) = pushDownLimit(child, limit)
+ if (canRemoveLimit) {
+ // If we can remove limit, it indicates data source only have one partition.
+ // For `dataset.limit(m).offset(n)`, try to push down `limit(m).offset(n)`.
+ // For example, `dataset.limit(5).offset(3)`, we can push down `limit(5).offset(3)`.
+ val isPushed = pushDownOffset(newChild, offsetValue)
+ if (isPushed) {
+ newChild
+ } else {
+ // For `dataset.limit(m).offset(n)`, only push down `limit(m)`.
+ // Keep the OFFSET operator if we failed to push down OFFSET to the data source.
+ offset.withNewChildren(Seq(newChild))
+ }
+ } else {
+ // Keep the OFFSET operator if we can't remove LIMIT operator.
+ offset
+ }
+ case globalLimit @ OffsetAndLimit(offset, limit, child) =>
+ val (newChild, canRemoveLimit) = pushDownLimit(child, limit + offset)
+ if (canRemoveLimit) {
+ // If we can remove limit, it indicates data source only have one partition.
+ // For `dataset.offset(n).limit(m)`, try to push down `limit(m + n).offset(n)`.
+ // For example, `dataset.offset(3).limit(5)`, we can push down `limit(8).offset(3)`.
+ val isPushed = pushDownOffset(newChild, offset)
+ if (isPushed) {
+ newChild
+ } else {
+ // For `dataset.offset(n).limit(m)`, try to push down `limit(m + n)`.
+ // Spark still do `offset(n)`.
+ // For example, `dataset.offset(3).limit(5)`, we can push down `limit(8)`.
+ // Spark still do `offset(3)`.
+ Offset(Literal(offset), newChild)
Review Comment:
The comment can be
```
// Still keep the OFFSET operator if we can't push it down.
```
--
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
[GitHub] [spark] cloud-fan commented on a diff in pull request #36295: [SPARK-38978][SQL] DS V2 supports push down OFFSET operator
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #36295:
URL: https://github.com/apache/spark/pull/36295#discussion_r903816073
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala:
##########
@@ -407,7 +407,66 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper wit
case other => (other, false)
}
- def pushDownLimits(plan: LogicalPlan): LogicalPlan = plan.transform {
+ private def pushDownOffset(
+ plan: LogicalPlan,
+ offset: Int): Boolean = plan match {
+ case sHolder: ScanBuilderHolder =>
+ val isPushed = PushDownUtils.pushOffset(sHolder.builder, offset)
+ if (isPushed) {
+ sHolder.pushedOffset = Some(offset)
+ }
+ isPushed
+ case p: Project =>
+ pushDownOffset(p.child, offset)
+ case _ => false
+ }
+
+ def pushDownLimitAndOffset(plan: LogicalPlan): LogicalPlan = plan.transform {
+ case offset @ LimitAndOffset(limit, offsetValue, child) =>
+ val (newChild, canRemoveLimit) = pushDownLimit(child, limit)
+ if (canRemoveLimit) {
+ // If we can remove limit, it indicates data source only have one partition.
+ // For `dataset.limit(m).offset(n)`, try to push down `limit(m).offset(n)`.
+ // For example, `dataset.limit(5).offset(3)`, we can push down `limit(5).offset(3)`.
+ val isPushed = pushDownOffset(newChild, offsetValue)
+ if (isPushed) {
+ newChild
+ } else {
+ // For `dataset.limit(m).offset(n)`, only push down `limit(m)`.
+ // Keep the OFFSET operator if we failed to push down OFFSET to the data source.
+ offset.withNewChildren(Seq(newChild))
+ }
+ } else {
+ // Keep the OFFSET operator if we can't remove LIMIT operator.
+ offset
+ }
+ case globalLimit @ OffsetAndLimit(offset, limit, child) =>
+ val (newChild, canRemoveLimit) = pushDownLimit(child, limit + offset)
+ if (canRemoveLimit) {
+ // If we can remove limit, it indicates data source only have one partition.
+ // For `dataset.offset(n).limit(m)`, try to push down `limit(m + n).offset(n)`.
+ // For example, `dataset.offset(3).limit(5)`, we can push down `limit(8).offset(3)`.
+ val isPushed = pushDownOffset(newChild, offset)
+ if (isPushed) {
+ newChild
+ } else {
+ // For `dataset.offset(n).limit(m)`, try to push down `limit(m + n)`.
+ // Spark still do `offset(n)`.
+ // For example, `dataset.offset(3).limit(5)`, we can push down `limit(8)`.
+ // Spark still do `offset(3)`.
+ Offset(Literal(offset), newChild)
+ }
+ } else {
+ // For `dataset.offset(n).limit(m)`, try to push down `offset(n)`.
+ // Spark still do `limit(m)`.
+ val isPushed = pushDownOffset(child, offset)
+ if (isPushed) {
+ globalLimit.withNewChildren(Seq(child))
Review Comment:
Maybe `Limit(limit, child)`?
--
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
[GitHub] [spark] cloud-fan commented on a diff in pull request #36295: [SPARK-38978][SQL] Support push down OFFSET to JDBC data source V2
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #36295:
URL: https://github.com/apache/spark/pull/36295#discussion_r874587459
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala:
##########
@@ -44,6 +44,7 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper wit
pushDownFilters,
pushDownAggregates,
pushDownLimits,
+ pushDownOffsets,
Review Comment:
Or can we push them down in one step?
--
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
[GitHub] [spark] beliefer commented on a diff in pull request #36295: [SPARK-38978][SQL] Support push down OFFSET to JDBC data source V2
Posted by GitBox <gi...@apache.org>.
beliefer commented on code in PR #36295:
URL: https://github.com/apache/spark/pull/36295#discussion_r873554764
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala:
##########
@@ -419,6 +420,30 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper wit
}
}
+ private def pushDownOffset(plan: LogicalPlan, offset: Int): (LogicalPlan, Boolean) = plan match {
+ case operation @ ScanOperation(_, filter, sHolder: ScanBuilderHolder) if filter.isEmpty =>
+ val isPushed = PushDownUtils.pushOffset(sHolder.builder, offset)
+ if (isPushed) {
+ sHolder.pushedOffset = Some(offset)
+ }
+ (operation, isPushed)
+ case p: Project =>
+ val (newChild, isPushed) = pushDownOffset(p.child, offset)
+ (p.withNewChildren(Seq(newChild)), isPushed)
+ case other => (other, false)
+ }
+
+ def pushDownOffsets(plan: LogicalPlan): LogicalPlan = plan.transform {
+ // TODO supports push down Limit append Offset or Offset append Limit
+ case offset @ Offset(IntegerLiteral(n), child) =>
Review Comment:
OK
--
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
[GitHub] [spark] beliefer commented on a diff in pull request #36295: [SPARK-38978][SQL] Support push down OFFSET to JDBC data source V2
Posted by GitBox <gi...@apache.org>.
beliefer commented on code in PR #36295:
URL: https://github.com/apache/spark/pull/36295#discussion_r878954574
##########
sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCV2Suite.scala:
##########
@@ -203,6 +204,245 @@ class JDBCV2Suite extends QueryTest with SharedSparkSession with ExplainSuiteHel
checkAnswer(df5, Seq(Row(10000.00, 1000.0, "amy")))
}
+ private def checkOffsetRemoved(df: DataFrame, removed: Boolean = true): Unit = {
+ val offsets = df.queryExecution.optimizedPlan.collect {
+ case offset: Offset => offset
+ }
+ if (removed) {
+ assert(offsets.isEmpty)
+ } else {
+ assert(offsets.nonEmpty)
+ }
+ }
+
+ test("simple scan with OFFSET") {
+ val df1 = spark.read
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .offset(1)
+ checkOffsetRemoved(df1)
+ checkPushedInfo(df1,
+ "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], PushedOffset: OFFSET 1,")
+ checkAnswer(df1, Seq(Row(1, "cathy", 9000.00, 1200.0, false)))
+
+ val df2 = spark.read
+ .option("pushDownOffset", "false")
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .offset(1)
+ checkOffsetRemoved(df2, false)
+ checkPushedInfo(df2,
+ "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], ReadSchema:")
+ checkAnswer(df2, Seq(Row(1, "cathy", 9000.00, 1200.0, false)))
+
+ val df3 = spark.read
+ .option("partitionColumn", "dept")
+ .option("lowerBound", "0")
+ .option("upperBound", "2")
+ .option("numPartitions", "2")
+ .table("h2.test.employee")
+ .filter($"dept" > 1)
+ .offset(1)
+ checkOffsetRemoved(df3, false)
+ checkPushedInfo(df3, "PushedFilters: [DEPT IS NOT NULL, DEPT > 1], ReadSchema:")
+ checkAnswer(df3, Seq(Row(2, "david", 10000, 1300, true), Row(6, "jen", 12000, 1200, true)))
+
+ val df4 = spark.read
+ .table("h2.test.employee")
+ .groupBy("DEPT").sum("SALARY")
+ .offset(1)
+ checkOffsetRemoved(df4, false)
+ checkPushedInfo(df4,
+ "PushedAggregates: [SUM(SALARY)], PushedFilters: [], PushedGroupByExpressions: [DEPT], ")
+ checkAnswer(df4, Seq(Row(2, 22000.00), Row(6, 12000.00)))
+
+ val name = udf { (x: String) => x.matches("cat|dav|amy") }
+ val sub = udf { (x: String) => x.substring(0, 3) }
+ val df5 = spark.read
+ .table("h2.test.employee")
+ .select($"SALARY", $"BONUS", sub($"NAME").as("shortName"))
+ .filter(name($"shortName"))
+ .offset(1)
+ checkOffsetRemoved(df5, false)
+ // OFFSET is pushed down only if all the filters are pushed down
+ checkPushedInfo(df5, "PushedFilters: [], ")
+ checkAnswer(df5, Seq(Row(10000.00, 1300.0, "dav"), Row(9000.00, 1200.0, "cat")))
+ }
+
+ test("simple scan with LIMIT and OFFSET") {
+ val df1 = spark.read
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .limit(2)
+ .offset(1)
+ checkLimitRemoved(df1)
+ checkOffsetRemoved(df1)
+ checkPushedInfo(df1,
+ "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], PushedLimit: LIMIT 1, PushedOffset: OFFSET 1,")
Review Comment:
OK
--
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
[GitHub] [spark] sadikovi commented on pull request #36295: [SPARK-38978][SQL] Support push down OFFSET to JDBC data source V2
Posted by GitBox <gi...@apache.org>.
sadikovi commented on PR #36295:
URL: https://github.com/apache/spark/pull/36295#issuecomment-1146883974
Can we have some kind of performance numbers for "push down OFFSET could improves the performance."?
--
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
[GitHub] [spark] cloud-fan commented on a diff in pull request #36295: [SPARK-38978][SQL] Support push down OFFSET to JDBC data source V2
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #36295:
URL: https://github.com/apache/spark/pull/36295#discussion_r882693128
##########
sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCV2Suite.scala:
##########
@@ -203,6 +204,355 @@ class JDBCV2Suite extends QueryTest with SharedSparkSession with ExplainSuiteHel
checkAnswer(df5, Seq(Row(10000.00, 1000.0, "amy")))
}
+ private def checkOffsetRemoved(df: DataFrame, removed: Boolean = true): Unit = {
+ val offsets = df.queryExecution.optimizedPlan.collect {
+ case offset: Offset => offset
+ }
+ if (removed) {
+ assert(offsets.isEmpty)
+ } else {
+ assert(offsets.nonEmpty)
+ }
+ }
+
+ test("simple scan with OFFSET") {
+ val df1 = spark.read
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .offset(1)
+ checkOffsetRemoved(df1)
+ checkPushedInfo(df1,
+ "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], PushedOffset: OFFSET 1,")
+ checkAnswer(df1, Seq(Row(1, "cathy", 9000.00, 1200.0, false)))
+
+ val df2 = spark.read
+ .option("pushDownOffset", "false")
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .offset(1)
+ checkOffsetRemoved(df2, false)
+ checkPushedInfo(df2,
+ "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], ReadSchema:")
+ checkAnswer(df2, Seq(Row(1, "cathy", 9000.00, 1200.0, false)))
+
+ val df3 = spark.read
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .sort($"salary")
+ .offset(1)
+ checkOffsetRemoved(df3, false)
+ checkPushedInfo(df3,
+ "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], ReadSchema:")
+ checkAnswer(df3, Seq(Row(1, "amy", 10000.00, 1000.0, true)))
+
+ val df4 = spark.read
+ .option("partitionColumn", "dept")
+ .option("lowerBound", "0")
+ .option("upperBound", "2")
+ .option("numPartitions", "2")
+ .table("h2.test.employee")
+ .filter($"dept" > 1)
+ .offset(1)
+ checkOffsetRemoved(df4, false)
+ checkPushedInfo(df4, "PushedFilters: [DEPT IS NOT NULL, DEPT > 1], ReadSchema:")
+ checkAnswer(df4, Seq(Row(2, "david", 10000, 1300, true), Row(6, "jen", 12000, 1200, true)))
+
+ val df5 = spark.read
+ .table("h2.test.employee")
+ .groupBy("DEPT").sum("SALARY")
+ .offset(1)
+ checkOffsetRemoved(df5, false)
+ checkPushedInfo(df5,
+ "PushedAggregates: [SUM(SALARY)], PushedFilters: [], PushedGroupByExpressions: [DEPT], ")
+ checkAnswer(df5, Seq(Row(2, 22000.00), Row(6, 12000.00)))
+
+ val name = udf { (x: String) => x.matches("cat|dav|amy") }
+ val sub = udf { (x: String) => x.substring(0, 3) }
+ val df6 = spark.read
+ .table("h2.test.employee")
+ .select($"SALARY", $"BONUS", sub($"NAME").as("shortName"))
+ .filter(name($"shortName"))
+ .offset(1)
+ checkOffsetRemoved(df6, false)
+ // OFFSET is pushed down only if all the filters are pushed down
+ checkPushedInfo(df6, "PushedFilters: [], ")
+ checkAnswer(df6, Seq(Row(10000.00, 1300.0, "dav"), Row(9000.00, 1200.0, "cat")))
+ }
+
+ test("simple scan with LIMIT and OFFSET") {
+ val df1 = spark.read
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .limit(2)
+ .offset(1)
+ checkLimitRemoved(df1)
+ checkOffsetRemoved(df1)
+ checkPushedInfo(df1,
+ "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], PushedLimit: LIMIT 2, PushedOffset: OFFSET 1,")
+ checkAnswer(df1, Seq(Row(1, "cathy", 9000.00, 1200.0, false)))
+
+ val df2 = spark.read
+ .option("pushDownLimit", "false")
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .limit(2)
+ .offset(1)
+ checkLimitRemoved(df2, false)
+ checkOffsetRemoved(df2, false)
+ checkPushedInfo(df2,
+ "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], ReadSchema:")
+ checkAnswer(df2, Seq(Row(1, "cathy", 9000.00, 1200.0, false)))
+
+ val df3 = spark.read
+ .option("pushDownOffset", "false")
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .limit(2)
+ .offset(1)
+ checkLimitRemoved(df3)
+ checkOffsetRemoved(df3, false)
+ checkPushedInfo(df3,
+ "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], PushedLimit: LIMIT 2, ReadSchema:")
+ checkAnswer(df3, Seq(Row(1, "cathy", 9000.00, 1200.0, false)))
+
+ val df4 = spark.read
+ .option("pushDownLimit", "false")
+ .option("pushDownOffset", "false")
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .limit(2)
+ .offset(1)
+ checkLimitRemoved(df4, false)
+ checkOffsetRemoved(df4, false)
+ checkPushedInfo(df4,
+ "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], ReadSchema:")
+ checkAnswer(df4, Seq(Row(1, "cathy", 9000.00, 1200.0, false)))
+
+ val df5 = spark.read
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .sort($"salary")
+ .limit(2)
+ .offset(1)
+ checkLimitRemoved(df5)
+ checkOffsetRemoved(df5)
+ checkPushedInfo(df5, "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], " +
+ "PushedOffset: OFFSET 1, PushedTopN: ORDER BY [SALARY ASC NULLS FIRST] LIMIT 2, ReadSchema:")
+ checkAnswer(df5, Seq(Row(1, "amy", 10000.00, 1000.0, true)))
+
+ val df6 = spark.read
+ .option("pushDownLimit", "false")
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .sort($"salary")
+ .limit(2)
+ .offset(1)
+ checkLimitRemoved(df6, false)
+ checkOffsetRemoved(df6, false)
+ checkPushedInfo(df6, "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], ReadSchema:")
+ checkAnswer(df6, Seq(Row(1, "amy", 10000.00, 1000.0, true)))
+
+ val df7 = spark.read
+ .option("pushDownOffset", "false")
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .sort($"salary")
+ .limit(2)
+ .offset(1)
+ checkLimitRemoved(df7)
+ checkOffsetRemoved(df7, false)
+ checkPushedInfo(df7, "PushedFilters: [DEPT IS NOT NULL, DEPT = 1]," +
+ " PushedTopN: ORDER BY [SALARY ASC NULLS FIRST] LIMIT 2, ReadSchema:")
+ checkAnswer(df7, Seq(Row(1, "amy", 10000.00, 1000.0, true)))
+
+ val df8 = spark.read
+ .option("pushDownLimit", "false")
+ .option("pushDownOffset", "false")
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .sort($"salary")
+ .limit(2)
+ .offset(1)
+ checkLimitRemoved(df8, false)
+ checkOffsetRemoved(df8, false)
+ checkPushedInfo(df8, "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], ReadSchema:")
+ checkAnswer(df8, Seq(Row(1, "amy", 10000.00, 1000.0, true)))
+
+ val df9 = spark.read
+ .option("partitionColumn", "dept")
+ .option("lowerBound", "0")
+ .option("upperBound", "2")
+ .option("numPartitions", "2")
+ .table("h2.test.employee")
+ .filter($"dept" > 1)
+ .limit(2)
+ .offset(1)
+ checkLimitRemoved(df9, false)
+ checkOffsetRemoved(df9, false)
+ checkPushedInfo(df9,
+ "PushedFilters: [DEPT IS NOT NULL, DEPT > 1], PushedLimit: LIMIT 2, ReadSchema:")
+ checkAnswer(df9, Seq(Row(2, "david", 10000.00, 1300.0, true)))
+
+ val df10 = spark.read
+ .table("h2.test.employee")
+ .groupBy("DEPT").sum("SALARY")
+ .limit(2)
+ .offset(1)
+ checkLimitRemoved(df10, false)
+ checkOffsetRemoved(df10, false)
+ checkPushedInfo(df10,
+ "PushedAggregates: [SUM(SALARY)], PushedFilters: [], PushedGroupByExpressions: [DEPT], ")
+ checkAnswer(df10, Seq(Row(2, 22000.00)))
+
+ val name = udf { (x: String) => x.matches("cat|dav|amy") }
+ val sub = udf { (x: String) => x.substring(0, 3) }
+ val df11 = spark.read
+ .table("h2.test.employee")
+ .select($"SALARY", $"BONUS", sub($"NAME").as("shortName"))
+ .filter(name($"shortName"))
+ .limit(2)
+ .offset(1)
+ checkLimitRemoved(df11, false)
+ checkOffsetRemoved(df11, false)
+ checkPushedInfo(df11, "PushedFilters: [], ")
+ checkAnswer(df11, Seq(Row(9000.00, 1200.0, "cat")))
+ }
+
+ test("simple scan with OFFSET and LIMIT") {
+ val df1 = spark.read
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .offset(1)
+ .limit(1)
+ checkLimitRemoved(df1)
+ checkOffsetRemoved(df1)
+ checkPushedInfo(df1,
+ "[DEPT IS NOT NULL, DEPT = 1], PushedLimit: LIMIT 1, PushedOffset: OFFSET 1,")
Review Comment:
This is obviously wrong. The pushdown order is LIMIT then OFFSET, so the limit value should be 2.
--
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
[GitHub] [spark] cloud-fan commented on a diff in pull request #36295: [SPARK-38978][SQL] Support push down OFFSET to JDBC data source V2
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #36295:
URL: https://github.com/apache/spark/pull/36295#discussion_r882690656
##########
sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCV2Suite.scala:
##########
@@ -203,6 +204,355 @@ class JDBCV2Suite extends QueryTest with SharedSparkSession with ExplainSuiteHel
checkAnswer(df5, Seq(Row(10000.00, 1000.0, "amy")))
}
+ private def checkOffsetRemoved(df: DataFrame, removed: Boolean = true): Unit = {
+ val offsets = df.queryExecution.optimizedPlan.collect {
+ case offset: Offset => offset
+ }
+ if (removed) {
+ assert(offsets.isEmpty)
+ } else {
+ assert(offsets.nonEmpty)
+ }
+ }
+
+ test("simple scan with OFFSET") {
+ val df1 = spark.read
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .offset(1)
+ checkOffsetRemoved(df1)
+ checkPushedInfo(df1,
+ "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], PushedOffset: OFFSET 1,")
+ checkAnswer(df1, Seq(Row(1, "cathy", 9000.00, 1200.0, false)))
+
+ val df2 = spark.read
+ .option("pushDownOffset", "false")
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .offset(1)
+ checkOffsetRemoved(df2, false)
+ checkPushedInfo(df2,
+ "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], ReadSchema:")
+ checkAnswer(df2, Seq(Row(1, "cathy", 9000.00, 1200.0, false)))
+
+ val df3 = spark.read
+ .table("h2.test.employee")
+ .where($"dept" === 1)
+ .sort($"salary")
+ .offset(1)
+ checkOffsetRemoved(df3, false)
+ checkPushedInfo(df3,
+ "PushedFilters: [DEPT IS NOT NULL, DEPT = 1], ReadSchema:")
+ checkAnswer(df3, Seq(Row(1, "amy", 10000.00, 1000.0, true)))
+
+ val df4 = spark.read
+ .option("partitionColumn", "dept")
+ .option("lowerBound", "0")
+ .option("upperBound", "2")
+ .option("numPartitions", "2")
+ .table("h2.test.employee")
+ .filter($"dept" > 1)
+ .offset(1)
+ checkOffsetRemoved(df4, false)
+ checkPushedInfo(df4, "PushedFilters: [DEPT IS NOT NULL, DEPT > 1], ReadSchema:")
+ checkAnswer(df4, Seq(Row(2, "david", 10000, 1300, true), Row(6, "jen", 12000, 1200, true)))
+
+ val df5 = spark.read
+ .table("h2.test.employee")
+ .groupBy("DEPT").sum("SALARY")
Review Comment:
can we test the same query while we can fully push down agg? then offset can be pushed 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
[GitHub] [spark] cloud-fan commented on a diff in pull request #36295: [SPARK-38978][SQL] Support push down OFFSET to JDBC data source V2
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #36295:
URL: https://github.com/apache/spark/pull/36295#discussion_r882663613
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/DataSourceScanExec.scala:
##########
@@ -143,13 +143,15 @@ case class RowDataSourceScanExec(
val topNOrLimitInfo =
Review Comment:
seems we don't need to change anything in `topNOrLimitInfo`
--
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
[GitHub] [spark] beliefer commented on a diff in pull request #36295: [SPARK-38978][SQL] DS V2 supports push down OFFSET operator
Posted by GitBox <gi...@apache.org>.
beliefer commented on code in PR #36295:
URL: https://github.com/apache/spark/pull/36295#discussion_r902379735
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala:
##########
@@ -407,8 +407,69 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper wit
case other => (other, false)
}
- def pushDownLimits(plan: LogicalPlan): LogicalPlan = plan.transform {
+ private def pushDownOffset(
+ plan: LogicalPlan,
+ offset: Int): Boolean = plan match {
+ case sHolder: ScanBuilderHolder =>
+ val isPushed = PushDownUtils.pushOffset(sHolder.builder, offset)
+ if (isPushed) {
+ sHolder.pushedOffset = Some(offset)
+ }
+ isPushed
+ case p: Project =>
+ pushDownOffset(p.child, offset)
+ case _ => false
+ }
+
+ def pushDownLimitAndOffset(plan: LogicalPlan): LogicalPlan = plan.transform {
+ case offset @ LimitAndOffset(limit, offsetValue, child) =>
+ val (newChild, canRemoveLimit) = pushDownLimit(child, limit)
+ if (canRemoveLimit) {
+ // If we can remove limit, it indicates data source only have one partition.
+ // For `dataset.limit(m).offset(n)`, try to push down `offset(n).limit(m - n)`.
+ // For example, `dataset.limit(5).offset(3)`, we can push down `offset(3).limit(2)`.
Review Comment:
In fact, we can push down LIMIT and OFFSET 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
[GitHub] [spark] beliefer commented on a diff in pull request #36295: [SPARK-38978][SQL] DS V2 supports push down OFFSET operator
Posted by GitBox <gi...@apache.org>.
beliefer commented on code in PR #36295:
URL: https://github.com/apache/spark/pull/36295#discussion_r902375146
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala:
##########
@@ -407,8 +407,69 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper wit
case other => (other, false)
}
- def pushDownLimits(plan: LogicalPlan): LogicalPlan = plan.transform {
+ private def pushDownOffset(
+ plan: LogicalPlan,
+ offset: Int): Boolean = plan match {
+ case sHolder: ScanBuilderHolder =>
+ val isPushed = PushDownUtils.pushOffset(sHolder.builder, offset)
+ if (isPushed) {
+ sHolder.pushedOffset = Some(offset)
+ }
+ isPushed
+ case p: Project =>
+ pushDownOffset(p.child, offset)
+ case _ => false
+ }
+
+ def pushDownLimitAndOffset(plan: LogicalPlan): LogicalPlan = plan.transform {
+ case offset @ LimitAndOffset(limit, offsetValue, child) =>
+ val (newChild, canRemoveLimit) = pushDownLimit(child, limit)
+ if (canRemoveLimit) {
+ // If we can remove limit, it indicates data source only have one partition.
+ // For `dataset.limit(m).offset(n)`, try to push down `offset(n).limit(m - n)`.
+ // For example, `dataset.limit(5).offset(3)`, we can push down `offset(3).limit(2)`.
+ val isPushed = pushDownOffset(newChild, offsetValue)
+ if (isPushed) {
+ newChild
+ } else {
+ // For `dataset.limit(m).offset(n)`, only push down `limit(m)`.
+ // Spark still do `offset(n)`.
+ offset.withNewChildren(Seq(newChild))
+ }
+ } else {
+ // If we can't push down limit and offset, return `Offset`.
Review Comment:
How about `If we can't remove `Limit` and push down offset, return Offset.` ?
--
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
[GitHub] [spark] cloud-fan commented on a diff in pull request #36295: [SPARK-38978][SQL] DS V2 supports push down OFFSET operator
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #36295:
URL: https://github.com/apache/spark/pull/36295#discussion_r903804573
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala:
##########
@@ -407,7 +407,66 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper wit
case other => (other, false)
}
- def pushDownLimits(plan: LogicalPlan): LogicalPlan = plan.transform {
+ private def pushDownOffset(
+ plan: LogicalPlan,
+ offset: Int): Boolean = plan match {
+ case sHolder: ScanBuilderHolder =>
+ val isPushed = PushDownUtils.pushOffset(sHolder.builder, offset)
+ if (isPushed) {
+ sHolder.pushedOffset = Some(offset)
+ }
+ isPushed
+ case p: Project =>
+ pushDownOffset(p.child, offset)
+ case _ => false
+ }
+
+ def pushDownLimitAndOffset(plan: LogicalPlan): LogicalPlan = plan.transform {
+ case offset @ LimitAndOffset(limit, offsetValue, child) =>
+ val (newChild, canRemoveLimit) = pushDownLimit(child, limit)
+ if (canRemoveLimit) {
+ // If we can remove limit, it indicates data source only have one partition.
+ // For `dataset.limit(m).offset(n)`, try to push down `limit(m).offset(n)`.
+ // For example, `dataset.limit(5).offset(3)`, we can push down `limit(5).offset(3)`.
+ val isPushed = pushDownOffset(newChild, offsetValue)
+ if (isPushed) {
+ newChild
+ } else {
+ // For `dataset.limit(m).offset(n)`, only push down `limit(m)`.
+ // Keep the OFFSET operator if we failed to push down OFFSET to the data source.
+ offset.withNewChildren(Seq(newChild))
+ }
+ } else {
+ // Keep the OFFSET operator if we can't remove LIMIT operator.
+ offset
+ }
+ case globalLimit @ OffsetAndLimit(offset, limit, child) =>
+ val (newChild, canRemoveLimit) = pushDownLimit(child, limit + offset)
+ if (canRemoveLimit) {
+ // If we can remove limit, it indicates data source only have one partition.
+ // For `dataset.offset(n).limit(m)`, try to push down `limit(m + n).offset(n)`.
+ // For example, `dataset.offset(3).limit(5)`, we can push down `limit(8).offset(3)`.
+ val isPushed = pushDownOffset(newChild, offset)
+ if (isPushed) {
+ newChild
+ } else {
+ // For `dataset.offset(n).limit(m)`, try to push down `limit(m + n)`.
Review Comment:
This comment seems in a wrong place.
--
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
[GitHub] [spark] cloud-fan commented on a diff in pull request #36295: [SPARK-38978][SQL] DS V2 supports push down OFFSET operator
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #36295:
URL: https://github.com/apache/spark/pull/36295#discussion_r902319260
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala:
##########
@@ -407,8 +407,69 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper wit
case other => (other, false)
}
- def pushDownLimits(plan: LogicalPlan): LogicalPlan = plan.transform {
+ private def pushDownOffset(
+ plan: LogicalPlan,
+ offset: Int): Boolean = plan match {
+ case sHolder: ScanBuilderHolder =>
+ val isPushed = PushDownUtils.pushOffset(sHolder.builder, offset)
+ if (isPushed) {
+ sHolder.pushedOffset = Some(offset)
+ }
+ isPushed
+ case p: Project =>
+ pushDownOffset(p.child, offset)
+ case _ => false
+ }
+
+ def pushDownLimitAndOffset(plan: LogicalPlan): LogicalPlan = plan.transform {
+ case offset @ LimitAndOffset(limit, offsetValue, child) =>
+ val (newChild, canRemoveLimit) = pushDownLimit(child, limit)
+ if (canRemoveLimit) {
+ // If we can remove limit, it indicates data source only have one partition.
+ // For `dataset.limit(m).offset(n)`, try to push down `offset(n).limit(m - n)`.
+ // For example, `dataset.limit(5).offset(3)`, we can push down `offset(3).limit(2)`.
Review Comment:
The comment doesn't match the code. What it says are handled by `LimitAndOffset`. How about
```
// Only push down OFFSET if LIMIT has been fully pushed.
```
--
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