Skip to content

[SPARK-35231][SQL] logical.Range override maxRowsPerPartition #32350

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
zhengruifeng wants to merge 4 commits into from
Closed

[SPARK-35231][SQL] logical.Range override maxRowsPerPartition #32350

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
f82b0ab
init
zhengruifeng Apr 26, 2021
0a69629
add test
zhengruifeng Apr 29, 2021
30db1c2
fix test
zhengruifeng Apr 29, 2021
a3e26c7
use a simple test
zhengruifeng May 6, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 12 additions & 0 deletions ...st/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -69,6 +69,7 @@ case class Project(projectList: Seq[NamedExpression], child: LogicalPlan)
extends OrderPreservingUnaryNode {
override def output: Seq[Attribute] = projectList.map(_.toAttribute)
override def maxRows: Option[Long] = child.maxRows
override def maxRowsPerPartition: Option[Long] = child.maxRowsPerPartition
Copy link
Contributor Author

@zhengruifeng zhengruifeng Apr 29, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

this override is needed for the added test


override lazy val resolved: Boolean = {
val hasSpecialExpressions = projectList.exists ( _.collect {
Expand Down Expand Up @@ -161,6 +162,7 @@ case class Filter(condition: Expression, child: LogicalPlan)
override def output: Seq[Attribute] = child.output

override def maxRows: Option[Long] = child.maxRows
override def maxRowsPerPartition: Option[Long] = child.maxRowsPerPartition

final override val nodePatterns: Seq[TreePattern] = Seq(FILTER)

Expand Down Expand Up @@ -742,6 +744,16 @@ case class Range(
}
}

override def maxRowsPerPartition: Option[Long] = {
if (numSlices.isDefined) {
var m = numElements / numSlices.get
if (numElements % numSlices.get != 0) m += 1
if (m.isValidLong) Some(m.toLong) else maxRows
} else {
maxRows
}
}

override def computeStats(): Statistics = {
if (numElements == 0) {
Statistics(sizeInBytes = 0, rowCount = Some(0))
Expand Down
11 changes: 10 additions & 1 deletion sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/plans/LogicalPlanSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,9 @@
package org.apache.spark.sql.catalyst.plans

import org.apache.spark.SparkFunSuite
import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute, AttributeReference, Literal, NamedExpression}
import org.apache.spark.sql.catalyst.dsl.expressions._
import org.apache.spark.sql.catalyst.dsl.plans._
import org.apache.spark.sql.catalyst.expressions._
import org.apache.spark.sql.catalyst.plans.logical._
import org.apache.spark.sql.types.IntegerType

Expand Down Expand Up @@ -96,4 +98,11 @@ class LogicalPlanSuite extends SparkFunSuite {
OneRowRelation())
assert(result.sameResult(expected))
}

test("SPARK-35231: logical.Range override maxRowsPerPartition") {
assert(Range(0, 100, 1, 3).maxRowsPerPartition === Some(34))
assert(Range(0, 100, 1, 4).maxRowsPerPartition === Some(25))
assert(Range(0, 100, 1, 3).select('id).maxRowsPerPartition === Some(34))
assert(Range(0, 100, 1, 3).where('id % 2 === 1).maxRowsPerPartition === Some(34))
}
}