Skip to content

[SPARK-13636][SQL] Directly consume UnsafeRow in wholestage codegen plans #11484

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
viirya wants to merge 7 commits into from
Closed

[SPARK-13636][SQL] Directly consume UnsafeRow in wholestage codegen plans #11484

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
9c77d73
init import.
viirya Mar 2, 2016
6941eb1
Add some comments.
viirya Mar 3, 2016
00e5090
Merge remote-tracking branch 'upstream/master' into direct-consume-un…
viirya Mar 9, 2016
7428fd4
Address comments.
viirya Mar 9, 2016
6400eb2
Simplify the solution.
viirya Mar 9, 2016
dea644a
Address comments.
viirya Mar 9, 2016
6f0ae35
Address comments.
viirya Mar 10, 2016
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
Prev Previous commit
Next Next commit
Address comments.
  • Loading branch information
@viirya
viirya committed Mar 9, 2016
commit 7428fd4ce7008631194a29c9c02a3e2cd0aa9a7c
2 changes: 1 addition & 1 deletion sql/core/src/main/scala/org/apache/spark/sql/execution/Sort.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -155,7 +155,7 @@ case class Sort(
""".stripMargin.trim
}

override def doConsume(ctx: CodegenContext, input: Seq[ExprCode], row: String = null): String = {
override def doConsume(ctx: CodegenContext, input: Seq[ExprCode], row: String): String = {
if (row != null) {
s"$sorterVariable.insertRow((UnsafeRow)$row.copy());"
Copy link
Contributor

@davies davies Mar 9, 2016

Choose a reason for hiding this comment

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

Do we need the copy here? I think the sorter will copy it by itself.

Copy link
Member Author

@viirya viirya Mar 10, 2016

Choose a reason for hiding this comment

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

Yes. I checked it. Will remove this copy call.

} else {
Expand Down
4 changes: 2 additions & 2 deletions sql/core/src/main/scala/org/apache/spark/sql/execution/WholeStageCodegen.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -68,7 +68,7 @@ trait CodegenSupport extends SparkPlan {
protected var parent: CodegenSupport = null

/**
* Whether this SparkPlan accepts UnsafeRow as input in consumeChild.
* Whether this SparkPlan accepts UnsafeRow as input in doConsume.
*/
def consumeUnsafeRow: Boolean = false
Copy link
Contributor

@davies davies Mar 9, 2016

Choose a reason for hiding this comment

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

preferUnsafeRow ?


Expand Down Expand Up @@ -211,7 +211,7 @@ trait CodegenSupport extends SparkPlan {
* if (isNull1 || !value2) continue;
* # call consume(), which will call parent.doConsume()
*/
protected def doConsume(ctx: CodegenContext, input: Seq[ExprCode], row: String = null): String = {
protected def doConsume(ctx: CodegenContext, input: Seq[ExprCode], row: String): String = {
throw new UnsupportedOperationException
}
}
Expand Down
4 changes: 2 additions & 2 deletions sql/core/src/main/scala/org/apache/spark/sql/execution/basicOperators.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -49,7 +49,7 @@ case class Project(projectList: Seq[NamedExpression], child: SparkPlan)
references.filter(a => usedMoreThanOnce.contains(a.exprId))
}

override def doConsume(ctx: CodegenContext, input: Seq[ExprCode], row: String = null): String = {
override def doConsume(ctx: CodegenContext, input: Seq[ExprCode], row: String): String = {
val exprs = projectList.map(x =>
ExpressionCanonicalizer.execute(BindReferences.bindReference(x, child.output)))
ctx.currentVars = input
Expand Down Expand Up @@ -88,7 +88,7 @@ case class Filter(condition: Expression, child: SparkPlan) extends UnaryNode wit
child.asInstanceOf[CodegenSupport].produce(ctx, this)
}

override def doConsume(ctx: CodegenContext, input: Seq[ExprCode], row: String = null): String = {
override def doConsume(ctx: CodegenContext, input: Seq[ExprCode], row: String): String = {
val numOutput = metricTerm(ctx, "numOutputRows")
val expr = ExpressionCanonicalizer.execute(
BindReferences.bindReference(condition, child.output))
Expand Down
2 changes: 1 addition & 1 deletion sql/core/src/main/scala/org/apache/spark/sql/execution/debug/package.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -136,7 +136,7 @@ package object debug {
child.asInstanceOf[CodegenSupport].produce(ctx, this)
}

override def doConsume(ctx: CodegenContext, input: Seq[ExprCode], row: String = null): String = {
override def doConsume(ctx: CodegenContext, input: Seq[ExprCode], row: String): String = {
consume(ctx, input)
}
}
Expand Down
2 changes: 1 addition & 1 deletion sql/core/src/main/scala/org/apache/spark/sql/execution/limit.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -65,7 +65,7 @@ trait BaseLimit extends UnaryNode with CodegenSupport {
child.asInstanceOf[CodegenSupport].produce(ctx, this)
}

override def doConsume(ctx: CodegenContext, input: Seq[ExprCode], row: String = null): String = {
override def doConsume(ctx: CodegenContext, input: Seq[ExprCode], row: String): String = {
val stopEarly = ctx.freshName("stopEarly")
ctx.addMutableState("boolean", stopEarly, s"$stopEarly = false;")

Expand Down