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/30 07:59:12 UTC
[GitHub] [spark] beliefer opened a new pull request, #36417: [SPARK-28330][SQL][FOLLOWUP] Remove logical GlobalLimitAndOffset
beliefer opened a new pull request, #36417:
URL: https://github.com/apache/spark/pull/36417
### What changes were proposed in this pull request?
https://github.com/apache/spark/pull/35975 supports offset clause, it create a logical node named
`GlobalLimitAndOffset`. In fact, we can avoid use this node and use `Offset` instead and the latter is good with unify name.
### Why are the changes needed?
Improve the implement of offset clause.
### Does this PR introduce _any_ user-facing change?
'No'.
New feature.
### How was this patch tested?
Exists test cases.
--
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 #36417: [SPARK-39057][SQL] Offset could work without Limit
Posted by GitBox <gi...@apache.org>.
beliefer commented on PR #36417:
URL: https://github.com/apache/spark/pull/36417#issuecomment-1122159431
@cloud-fan Thank you very much !
--
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 #36417: [SPARK-39057][SQL] Offset could work without Limit
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #36417:
URL: https://github.com/apache/spark/pull/36417#discussion_r866824230
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala:
##########
@@ -1871,21 +1874,18 @@ object EliminateLimits extends Rule[LogicalPlan] {
}
/**
- * Rewrite [[Offset]] as [[GlobalLimitAndOffset]] or [[LocalLimit]],
- * merging the expressions into one single expression. See [[Limit]] for more information
- * about the difference between [[LocalLimit]] and [[GlobalLimit]].
+ * This rule optimizes Offset operators by:
+ * 1. Eliminate [[Offset]] operators if offset == 0.
+ * 2. Replace [[Offset]] operators to empty [[LocalRelation]]
+ * if [[Offset]]'s child max row <= offset.
*/
-object RewriteOffsets extends Rule[LogicalPlan] {
+object EliminateOffsets extends Rule[LogicalPlan] {
def apply(plan: LogicalPlan): LogicalPlan = plan transform {
- case GlobalLimit(le, Offset(oe, grandChild)) =>
- GlobalLimitAndOffset(le, oe, grandChild)
- case localLimit @ LocalLimit(le, Offset(oe, grandChild)) =>
- val offset = oe.eval().asInstanceOf[Int]
- if (offset == 0) {
- localLimit.withNewChildren(Seq(grandChild))
- } else {
- Offset(oe, LocalLimit(Add(le, oe), grandChild))
- }
+ case Offset(oe, child) if oe.foldable && oe.eval().asInstanceOf[Int] == 0 =>
+ child
+ case Offset(oe, child)
+ if oe.foldable && child.maxRows.exists(_ <= oe.eval().asInstanceOf[Int]) =>
+ LocalRelation(child.output, data = Seq.empty, isStreaming = plan.isStreaming)
Review Comment:
```suggestion
LocalRelation(child.output, data = Seq.empty, isStreaming = child.isStreaming)
```
--
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 #36417: [SPARK-39057][SQL] Offset could work without Limit
Posted by GitBox <gi...@apache.org>.
beliefer commented on code in PR #36417:
URL: https://github.com/apache/spark/pull/36417#discussion_r866603469
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala:
##########
@@ -92,15 +92,17 @@ abstract class SparkStrategies extends QueryPlanner[SparkPlan] {
CollectLimitExec(limit, planLater(child)) :: Nil
case LimitAndOffset(IntegerLiteral(limit), IntegerLiteral(offset),
Sort(order, true, child)) if limit + offset < conf.topKSortFallbackThreshold =>
- TakeOrderedAndProjectExec(
- limit, order, child.output, planLater(child), Some(offset)) :: Nil
+ TakeOrderedAndProjectExec(
Review Comment:
I'm dazzled
--
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 #36417: [SPARK-39057][SQL] Offset could work without Limit
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #36417:
URL: https://github.com/apache/spark/pull/36417#discussion_r866553021
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala:
##########
@@ -818,8 +820,6 @@ abstract class SparkStrategies extends QueryPlanner[SparkPlan] {
execution.LocalLimitExec(limit, planLater(child)) :: Nil
case logical.GlobalLimit(IntegerLiteral(limit), child) =>
execution.GlobalLimitExec(limit, planLater(child)) :: Nil
- case logical.GlobalLimitAndOffset(IntegerLiteral(limit), IntegerLiteral(offset), child) =>
Review Comment:
we should put `case logical.Offset(IntegerLiteral(offset), child) =>` here, as it's not a special 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 #36417: [SPARK-39057][SQL] Offset could work without Limit
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #36417:
URL: https://github.com/apache/spark/pull/36417#discussion_r866554687
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/limit.scala:
##########
@@ -70,9 +78,11 @@ case class CollectLimitExec(
val singlePartitionRDD = if (childRDD.getNumPartitions == 1) {
childRDD
} else {
- val locallyLimited =
- offsetOpt.map(offset => childRDD.mapPartitionsInternal(_.take(limit + offset)))
- .getOrElse(childRDD.mapPartitionsInternal(_.take(limit)))
+ val locallyLimited = if (offset > 0) {
Review Comment:
We also need to take care of the case if `limit = -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 #36417: [SPARK-39057][SQL] Offset could work without Limit
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #36417:
URL: https://github.com/apache/spark/pull/36417#discussion_r866827428
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/limit.scala:
##########
@@ -207,32 +230,47 @@ case class GlobalLimitExec(limit: Int, child: SparkPlan) extends BaseLimitExec {
* the child's single output partition.
*/
case class GlobalLimitAndOffsetExec(
- limit: Int,
+ limit: Int = -1,
offset: Int,
child: SparkPlan) extends BaseLimitExec {
+ assert(offset > 0)
override def requiredChildDistribution: List[Distribution] = AllTuples :: Nil
- override def doExecute(): RDD[InternalRow] = child.execute().mapPartitions { iter =>
- iter.take(limit + offset).drop(offset)
+ override def doExecute(): RDD[InternalRow] = if (limit >= 0) {
+ child.execute().mapPartitions(iter => iter.take(limit + offset).drop(offset))
Review Comment:
not related to this PR, but since we are touch here, let's use `mapPartitionsInternal` which is more efficient.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/limit.scala:
##########
@@ -207,32 +230,47 @@ case class GlobalLimitExec(limit: Int, child: SparkPlan) extends BaseLimitExec {
* the child's single output partition.
*/
case class GlobalLimitAndOffsetExec(
- limit: Int,
+ limit: Int = -1,
offset: Int,
child: SparkPlan) extends BaseLimitExec {
+ assert(offset > 0)
override def requiredChildDistribution: List[Distribution] = AllTuples :: Nil
- override def doExecute(): RDD[InternalRow] = child.execute().mapPartitions { iter =>
- iter.take(limit + offset).drop(offset)
+ override def doExecute(): RDD[InternalRow] = if (limit >= 0) {
+ child.execute().mapPartitions(iter => iter.take(limit + offset).drop(offset))
Review Comment:
not related to this PR, but since we are touching here, let's use `mapPartitionsInternal` which is more efficient.
--
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 #36417: [SPARK-39057][SQL] Offset could work without Limit
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #36417:
URL: https://github.com/apache/spark/pull/36417#discussion_r866827813
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/limit.scala:
##########
@@ -250,17 +288,20 @@ case class TakeOrderedAndProjectExec(
limit: Int,
sortOrder: Seq[SortOrder],
projectList: Seq[NamedExpression],
- child: SparkPlan, offsetOpt: Option[Int] = None) extends UnaryExecNode {
+ child: SparkPlan,
+ offset: Int = 0) extends UnaryExecNode {
override def output: Seq[Attribute] = {
projectList.map(_.toAttribute)
}
override def executeCollect(): Array[InternalRow] = {
val ord = new LazilyGeneratedOrdering(sortOrder, child.output)
- val data = offsetOpt
- .map(offset => child.execute().map(_.copy()).takeOrdered(limit + offset)(ord).drop(offset))
- .getOrElse(child.execute().map(_.copy()).takeOrdered(limit)(ord))
+ val data = if (offset > 0) {
+ child.execute().map(_.copy()).takeOrdered(limit + offset)(ord).drop(offset)
Review Comment:
ditto, use `mapPartitionsInternal`
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/limit.scala:
##########
@@ -286,11 +327,15 @@ case class TakeOrderedAndProjectExec(
val singlePartitionRDD = if (childRDD.getNumPartitions == 1) {
childRDD
} else {
- val localTopK = offsetOpt.map(offset => childRDD.mapPartitions { iter =>
- Utils.takeOrdered(iter.map(_.copy()), limit + offset)(ord)
- }).getOrElse(childRDD.mapPartitions { iter =>
- Utils.takeOrdered(iter.map(_.copy()), limit)(ord)
- })
+ val localTopK = if (offset > 0) {
+ childRDD.mapPartitions { iter =>
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 pull request #36417: [SPARK-39057][SQL] Offset could work without Limit
Posted by GitBox <gi...@apache.org>.
beliefer commented on PR #36417:
URL: https://github.com/apache/spark/pull/36417#issuecomment-1115547178
ping @dtenedor 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] cloud-fan commented on a diff in pull request #36417: [SPARK-39057][SQL] Offset could work without Limit
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #36417:
URL: https://github.com/apache/spark/pull/36417#discussion_r866557269
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/limit.scala:
##########
@@ -207,32 +222,62 @@ case class GlobalLimitExec(limit: Int, child: SparkPlan) extends BaseLimitExec {
* the child's single output partition.
*/
case class GlobalLimitAndOffsetExec(
- limit: Int,
- offset: Int,
+ limit: Int = -1,
+ offset: Int = 0,
Review Comment:
We don't need a default value for `offset`. If there is no offset, we should use `GlobalLimitExec`.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/limit.scala:
##########
@@ -207,32 +222,62 @@ case class GlobalLimitExec(limit: Int, child: SparkPlan) extends BaseLimitExec {
* the child's single output partition.
*/
case class GlobalLimitAndOffsetExec(
- limit: Int,
- offset: Int,
+ limit: Int = -1,
+ offset: Int = 0,
Review Comment:
and we can put a `assert(offset > 0)`
--
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 #36417: [SPARK-39057][SQL] Offset could work without Limit
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #36417:
URL: https://github.com/apache/spark/pull/36417#discussion_r866557980
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/limit.scala:
##########
@@ -250,17 +295,19 @@ case class TakeOrderedAndProjectExec(
limit: Int,
sortOrder: Seq[SortOrder],
projectList: Seq[NamedExpression],
- child: SparkPlan, offsetOpt: Option[Int] = None) extends UnaryExecNode {
+ child: SparkPlan, offset: Int = 0) extends UnaryExecNode {
Review Comment:
style nit: put the parameter in a new line
--
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 #36417: [SPARK-39057][SQL] Offset could work without Limit
Posted by GitBox <gi...@apache.org>.
beliefer commented on code in PR #36417:
URL: https://github.com/apache/spark/pull/36417#discussion_r866602901
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala:
##########
@@ -1871,14 +1871,12 @@ object EliminateLimits extends Rule[LogicalPlan] {
}
/**
- * Rewrite [[Offset]] as [[GlobalLimitAndOffset]] or [[LocalLimit]],
- * merging the expressions into one single expression. See [[Limit]] for more information
- * about the difference between [[LocalLimit]] and [[GlobalLimit]].
+ * Rewrite [[Offset]] by eliminate [[Offset]] or merge offset value and limit value into
+ * [[LocalLimit]]. See [[Limit]] for more information about the difference between
+ * [[LocalLimit]] and [[GlobalLimit]].
*/
object RewriteOffsets extends Rule[LogicalPlan] {
def apply(plan: LogicalPlan): LogicalPlan = plan transform {
- case GlobalLimit(le, Offset(oe, grandChild)) =>
- GlobalLimitAndOffset(le, oe, grandChild)
case localLimit @ LocalLimit(le, Offset(oe, grandChild)) =>
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 #36417: [SPARK-39057][SQL] Offset could work without Limit
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #36417:
URL: https://github.com/apache/spark/pull/36417#discussion_r866548301
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala:
##########
@@ -1871,14 +1871,12 @@ object EliminateLimits extends Rule[LogicalPlan] {
}
/**
- * Rewrite [[Offset]] as [[GlobalLimitAndOffset]] or [[LocalLimit]],
- * merging the expressions into one single expression. See [[Limit]] for more information
- * about the difference between [[LocalLimit]] and [[GlobalLimit]].
+ * Rewrite [[Offset]] by eliminate [[Offset]] or merge offset value and limit value into
+ * [[LocalLimit]]. See [[Limit]] for more information about the difference between
+ * [[LocalLimit]] and [[GlobalLimit]].
*/
object RewriteOffsets extends Rule[LogicalPlan] {
def apply(plan: LogicalPlan): LogicalPlan = plan transform {
- case GlobalLimit(le, Offset(oe, grandChild)) =>
- GlobalLimitAndOffset(le, oe, grandChild)
case localLimit @ LocalLimit(le, Offset(oe, grandChild)) =>
Review Comment:
can we move this logic to `LimitPushDown`? then we don't need this rule.
--
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 #36417: [SPARK-39057][SQL] Offset could work without Limit
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #36417:
URL: https://github.com/apache/spark/pull/36417#discussion_r866549645
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala:
##########
@@ -92,15 +92,17 @@ abstract class SparkStrategies extends QueryPlanner[SparkPlan] {
CollectLimitExec(limit, planLater(child)) :: Nil
case LimitAndOffset(IntegerLiteral(limit), IntegerLiteral(offset),
Sort(order, true, child)) if limit + offset < conf.topKSortFallbackThreshold =>
- TakeOrderedAndProjectExec(
- limit, order, child.output, planLater(child), Some(offset)) :: Nil
+ TakeOrderedAndProjectExec(
Review Comment:
the new indentation is incorrect.
--
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 #36417: [SPARK-39057][SQL] Offset could work without Limit
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #36417:
URL: https://github.com/apache/spark/pull/36417#discussion_r866559243
##########
sql/core/src/test/resources/sql-tests/inputs/subquery/in-subquery/in-limit.sql:
##########
@@ -159,3 +159,54 @@ GROUP BY t1b
ORDER BY t1b NULLS last
LIMIT 1
OFFSET 1;
+
+-- OFFSET in parent side
+-- TC 03.01
+SELECT *
+FROM t1
+WHERE t1a IN (SELECT t2a
+ FROM t2
+ WHERE t1d = t2d)
+OFFSET 2;
+
+-- TC 03.02
+SELECT *
+FROM t1
+WHERE t1c IN (SELECT t2c
+ FROM t2
+ WHERE t2b >= 8
+ OFFSET 2)
+OFFSET 4;
+
+-- TC 03.03
+SELECT Count(DISTINCT( t1a )),
+ t1b
+FROM t1
+WHERE t1d IN (SELECT t2d
+ FROM t2
+ ORDER BY t2c, t2d
+ OFFSET 2)
+GROUP BY t1b
+ORDER BY t1b DESC NULLS FIRST
+OFFSET 1;
+
+-- OFFSET with NOT IN
+-- TC 03.04
+SELECT *
+FROM t1
+WHERE t1b NOT IN (SELECT t2b
+ FROM t2
+ WHERE t2b > 6
+ OFFSET 2);
+
+-- TC 03.05
+SELECT Count(DISTINCT( t1a )),
+ t1b
+FROM t1
+WHERE t1d NOT IN (SELECT t2d
+ FROM t2
+ ORDER BY t2b DESC nulls first, t2d
+ OFFSET 1)
+GROUP BY t1b
+ORDER BY t1b NULLS last
+OFFSET 1;
Review Comment:
do we have tests for non-subquery offset-only queries?
--
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 #36417: [SPARK-39057][SQL] Offset could work without Limit
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #36417:
URL: https://github.com/apache/spark/pull/36417#discussion_r866652724
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala:
##########
@@ -750,6 +749,11 @@ object LimitPushDown extends Rule[LogicalPlan] {
Limit(le, Project(a.aggregateExpressions, LocalLimit(le, a.child)))
case Limit(le @ IntegerLiteral(1), p @ Project(_, a: Aggregate)) if a.groupOnly =>
Limit(le, p.copy(child = Project(a.aggregateExpressions, LocalLimit(le, a.child))))
+ // Merge offset value and limit value into LocalLimit and pushes down LocalLimit through Offset.
+ case LocalLimit(le, Offset(oe @ IntegerLiteral(offset), grandChild)) if offset > 0 =>
+ Offset(oe, LocalLimit(Add(le, oe), grandChild))
+ // Eliminate Offset if offset value equals 0.
+ case Offset(IntegerLiteral(0), child) => child
Review Comment:
We can add a new rule `EliminateOffsets` for this. And there is one more idea: if `child.maxRows < offset`, then we can simply turn the query into a empty local relation.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] beliefer commented on a diff in pull request #36417: [SPARK-39057][SQL] Offset could work without Limit
Posted by GitBox <gi...@apache.org>.
beliefer commented on code in PR #36417:
URL: https://github.com/apache/spark/pull/36417#discussion_r867287201
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/limit.scala:
##########
@@ -207,32 +230,47 @@ case class GlobalLimitExec(limit: Int, child: SparkPlan) extends BaseLimitExec {
* the child's single output partition.
*/
case class GlobalLimitAndOffsetExec(
- limit: Int,
+ limit: Int = -1,
offset: Int,
child: SparkPlan) extends BaseLimitExec {
+ assert(offset > 0)
override def requiredChildDistribution: List[Distribution] = AllTuples :: Nil
- override def doExecute(): RDD[InternalRow] = child.execute().mapPartitions { iter =>
- iter.take(limit + offset).drop(offset)
+ override def doExecute(): RDD[InternalRow] = if (limit >= 0) {
+ child.execute().mapPartitions(iter => iter.take(limit + offset).drop(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 a diff in pull request #36417: [SPARK-39057][SQL] Offset could work without Limit
Posted by GitBox <gi...@apache.org>.
beliefer commented on code in PR #36417:
URL: https://github.com/apache/spark/pull/36417#discussion_r866627978
##########
sql/core/src/test/resources/sql-tests/inputs/subquery/in-subquery/in-limit.sql:
##########
@@ -159,3 +159,54 @@ GROUP BY t1b
ORDER BY t1b NULLS last
LIMIT 1
OFFSET 1;
+
+-- OFFSET in parent side
+-- TC 03.01
+SELECT *
+FROM t1
+WHERE t1a IN (SELECT t2a
+ FROM t2
+ WHERE t1d = t2d)
+OFFSET 2;
+
+-- TC 03.02
+SELECT *
+FROM t1
+WHERE t1c IN (SELECT t2c
+ FROM t2
+ WHERE t2b >= 8
+ OFFSET 2)
+OFFSET 4;
+
+-- TC 03.03
+SELECT Count(DISTINCT( t1a )),
+ t1b
+FROM t1
+WHERE t1d IN (SELECT t2d
+ FROM t2
+ ORDER BY t2c, t2d
+ OFFSET 2)
+GROUP BY t1b
+ORDER BY t1b DESC NULLS FIRST
+OFFSET 1;
+
+-- OFFSET with NOT IN
+-- TC 03.04
+SELECT *
+FROM t1
+WHERE t1b NOT IN (SELECT t2b
+ FROM t2
+ WHERE t2b > 6
+ OFFSET 2);
+
+-- TC 03.05
+SELECT Count(DISTINCT( t1a )),
+ t1b
+FROM t1
+WHERE t1d NOT IN (SELECT t2d
+ FROM t2
+ ORDER BY t2b DESC nulls first, t2d
+ OFFSET 1)
+GROUP BY t1b
+ORDER BY t1b NULLS last
+OFFSET 1;
Review Comment:
Uncommented in postgreSQL/limit.sql
--
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 #36417: [SPARK-39057][SQL] Offset could work without Limit
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #36417:
URL: https://github.com/apache/spark/pull/36417#discussion_r866649081
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala:
##########
@@ -750,6 +749,11 @@ object LimitPushDown extends Rule[LogicalPlan] {
Limit(le, Project(a.aggregateExpressions, LocalLimit(le, a.child)))
case Limit(le @ IntegerLiteral(1), p @ Project(_, a: Aggregate)) if a.groupOnly =>
Limit(le, p.copy(child = Project(a.aggregateExpressions, LocalLimit(le, a.child))))
+ // Merge offset value and limit value into LocalLimit and pushes down LocalLimit through Offset.
+ case LocalLimit(le, Offset(oe @ IntegerLiteral(offset), grandChild)) if offset > 0 =>
Review Comment:
I think the pushdown is correct even if `offset == 0`. We don't need `if offset > 0`
--
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 #36417: [SPARK-39057][SQL] Offset could work without Limit
Posted by GitBox <gi...@apache.org>.
cloud-fan closed pull request #36417: [SPARK-39057][SQL] Offset could work without Limit
URL: https://github.com/apache/spark/pull/36417
--
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 #36417: [SPARK-39057][SQL] Offset could work without Limit
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on PR #36417:
URL: https://github.com/apache/spark/pull/36417#issuecomment-1122154997
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