Skip to content

[SPARK-2327] [SQL] Fix nullabilities of Join/Generate/Aggregate. #1266

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

[SPARK-2327] [SQL] Fix nullabilities of Join/Generate/Aggregate. #1266

Show file tree
Hide file tree
Changes from 6 commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
f20f196
Fix Join output nullabilities.
ueshin Jun 30, 2014
09532ec
Fix Generate output nullabilities.
ueshin Jun 30, 2014
0e31e37
Fix Aggregate resultAttribute nullabilities.
ueshin Jun 30, 2014
a0fc9bc
Fix scalastyle errors.
ueshin Jun 30, 2014
799ce56
Add nullabilization to Generate of SparkPlan.
ueshin Jul 1, 2014
df1ae53
Modify nullabilize to leave attribute if not resolved.
ueshin Jul 1, 2014
3ace83a
Add withNullability to Attribute and use it to change nullabilities.
ueshin Jul 4, 2014
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
46 changes: 38 additions & 8 deletions sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicOperators.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,7 @@
package org.apache.spark.sql.catalyst.plans.logical

import org.apache.spark.sql.catalyst.expressions._
import org.apache.spark.sql.catalyst.plans.{LeftSemi, JoinType}
import org.apache.spark.sql.catalyst.plans._
import org.apache.spark.sql.catalyst.types._

case class Project(projectList: Seq[NamedExpression], child: LogicalPlan) extends UnaryNode {
Expand Down Expand Up @@ -46,10 +46,22 @@ case class Generate(
child: LogicalPlan)
extends UnaryNode {

protected def generatorOutput: Seq[Attribute] =
alias
protected def generatorOutput: Seq[Attribute] = {
val output = alias
.map(a => generator.output.map(_.withQualifiers(a :: Nil)))
.getOrElse(generator.output)
if (join && outer) {
output.map {
case attr if !attr.resolved => attr
case attr if !attr.nullable =>
AttributeReference(
attr.name, attr.dataType, nullable = true)(attr.exprId, attr.qualifiers)
case attr => attr
Copy link
Contributor

@marmbrus marmbrus Jul 3, 2014

Choose a reason for hiding this comment

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

This is a pretty common pattern. We already have withNullability, which you could use in these cases if we move it up as an abstract method in Attribute and add a no-op version in UnresolvedAttribute.

Copy link
Member Author

@ueshin ueshin Jul 4, 2014

Choose a reason for hiding this comment

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

Ah, I see.
It is the same as withQualifiers, right?

Copy link
Contributor

@marmbrus marmbrus Jul 4, 2014

Choose a reason for hiding this comment

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

Yeah, exactly.

}
} else {
output
}
}

override def output =
if (join) child.output ++ generatorOutput else generatorOutput
Expand Down Expand Up @@ -81,11 +93,29 @@ case class Join(
condition: Option[Expression]) extends BinaryNode {

override def references = condition.map(_.references).getOrElse(Set.empty)
override def output = joinType match {
case LeftSemi =>
left.output
case _ =>
left.output ++ right.output
override def output = {
def nullabilize(output: Seq[Attribute]) = {
output.map {
case attr if !attr.resolved => attr
case attr if !attr.nullable =>
AttributeReference(
attr.name, attr.dataType, nullable = true)(attr.exprId, attr.qualifiers)
case attr => attr
}
}

joinType match {
case LeftSemi =>
left.output
case LeftOuter =>
left.output ++ nullabilize(right.output)
case RightOuter =>
nullabilize(left.output) ++ right.output
case FullOuter =>
nullabilize(left.output) ++ nullabilize(right.output)
case _ =>
left.output ++ right.output
}
}
}

Expand Down
4 changes: 2 additions & 2 deletions sql/core/src/main/scala/org/apache/spark/sql/execution/Aggregate.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -83,8 +83,8 @@ case class Aggregate(
case a: AggregateExpression =>
ComputedAggregate(
a,
BindReferences.bindReference(a, childOutput).asInstanceOf[AggregateExpression],
AttributeReference(s"aggResult:$a", a.dataType, nullable = true)())
BindReferences.bindReference(a, childOutput),
AttributeReference(s"aggResult:$a", a.dataType, a.nullable)())
}
}.toArray

Expand Down
17 changes: 15 additions & 2 deletions sql/core/src/main/scala/org/apache/spark/sql/execution/Generate.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,7 @@
package org.apache.spark.sql.execution

import org.apache.spark.annotation.DeveloperApi
import org.apache.spark.sql.catalyst.expressions.{Generator, JoinedRow, Literal, Projection}
import org.apache.spark.sql.catalyst.expressions._

/**
* :: DeveloperApi ::
Expand All @@ -39,8 +39,21 @@ case class Generate(
child: SparkPlan)
extends UnaryNode {

protected def generatorOutput: Seq[Attribute] = {
if (join && outer) {
generator.output.map {
case attr if !attr.nullable =>
AttributeReference(
attr.name, attr.dataType, nullable = true)(attr.exprId, attr.qualifiers)
case attr => attr
}
} else {
generator.output
}
}

override def output =
if (join) child.output ++ generator.output else generator.output
if (join) child.output ++ generatorOutput else generatorOutput

override def execute() = {
if (join) {
Expand Down
22 changes: 21 additions & 1 deletion sql/core/src/main/scala/org/apache/spark/sql/execution/joins.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -319,7 +319,27 @@ case class BroadcastNestedLoopJoin(

override def otherCopyArgs = sqlContext :: Nil

def output = left.output ++ right.output
def output = {
Copy link
Contributor

@marmbrus marmbrus Jul 3, 2014

Choose a reason for hiding this comment

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

Existing: add override when implementing abstract methods.

Copy link
Member Author

@ueshin ueshin Jul 4, 2014

Choose a reason for hiding this comment

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

Ok, I'll add it.

def nullabilize(output: Seq[Attribute]) = {
output.map {
case attr if !attr.nullable =>
AttributeReference(
attr.name, attr.dataType, nullable = true)(attr.exprId, attr.qualifiers)
case attr => attr
}
}

joinType match {
case LeftOuter =>
left.output ++ nullabilize(right.output)
case RightOuter =>
nullabilize(left.output) ++ right.output
case FullOuter =>
nullabilize(left.output) ++ nullabilize(right.output)
case _ =>
left.output ++ right.output
}
}

/** The Streamed Relation */
def left = streamed
Expand Down