Skip to content

[SPARK-14191][SQL] Remove invalid Expand operator constraints #11995

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 4 commits into from
Closed

[SPARK-14191][SQL] Remove invalid Expand operator constraints #11995

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
84821b4
Fix Expand constraints.
viirya Mar 28, 2016
5045c5d
Merge remote-tracking branch 'upstream/master' into fix-expand-constr…
viirya Mar 30, 2016
23d6b37
Move constraint pruning logic into Expand.
viirya Mar 30, 2016
ab89e62
Address comment. Remove constraints for Expand operator.
viirya Mar 31, 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
3 changes: 2 additions & 1 deletion sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -327,7 +327,8 @@ object ColumnPruning extends Rule[LogicalPlan] {
case p @ Project(_, a: Aggregate) if (a.outputSet -- p.references).nonEmpty =>
p.copy(
child = a.copy(aggregateExpressions = a.aggregateExpressions.filter(p.references.contains)))
case a @ Project(_, e @ Expand(_, _, grandChild)) if (e.outputSet -- a.references).nonEmpty =>
case a @ Project(_, e @ Expand(_, _, grandChild))
if (e.outputSet -- a.references).nonEmpty =>
Copy link
Contributor

@marmbrus marmbrus Apr 1, 2016

Choose a reason for hiding this comment

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

It is best to avoid spurious changes because it pollutes git blame. I can revert this while merging this time.

Copy link
Member Author

@viirya viirya Apr 2, 2016

Choose a reason for hiding this comment

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

Ok. Got it.

val newOutput = e.output.filter(a.references.contains(_))
val newProjects = e.projections.map { proj =>
proj.zip(e.output).filter { case (e, a) =>
Expand Down
3 changes: 2 additions & 1 deletion 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 @@ -519,14 +519,15 @@ case class Expand(
projections: Seq[Seq[Expression]],
output: Seq[Attribute],
child: LogicalPlan) extends UnaryNode {

override def references: AttributeSet =
AttributeSet(projections.flatten.flatMap(_.references))

override def statistics: Statistics = {
val sizeInBytes = super.statistics.sizeInBytes * projections.length
Statistics(sizeInBytes = sizeInBytes)
}

override protected def validConstraints: Set[Expression] = Set.empty[Expression]
Copy link
Contributor

@marmbrus marmbrus Apr 1, 2016

Choose a reason for hiding this comment

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

I'm also going to add a comment here to explain why this is empty.

Copy link
Member Author

@viirya viirya Apr 2, 2016

Choose a reason for hiding this comment

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

Thanks.

}

/**
Expand Down
1 change: 0 additions & 1 deletion sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/ColumnPruningSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -116,7 +116,6 @@ class ColumnPruningSuite extends PlanTest {
Seq('a, 'b, 'c, 'aa.int, 'gid.int),
input)).analyze
val optimized = Optimize.execute(query)

val expected =
Copy link
Contributor

@marmbrus marmbrus Apr 1, 2016

Choose a reason for hiding this comment

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

same

Aggregate(
Seq('aa, 'gid),
Expand Down
27 changes: 27 additions & 0 deletions ...alyst/src/test/scala/org/apache/spark/sql/catalyst/plans/ConstraintPropagationSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -88,6 +88,33 @@ class ConstraintPropagationSuite extends SparkFunSuite {
IsNotNull(resolveColumn(aliasedRelation.analyze, "a")))))
}

test("propagating constraints in expand") {
val tr = LocalRelation('a.int, 'b.int, 'c.int)

assert(tr.analyze.constraints.isEmpty)

// We add IsNotNull constraints for 'a, 'b and 'c into LocalRelation
// by creating notNullRelation.
val notNullRelation = tr.where('c.attr > 10 && 'a.attr < 5 && 'b.attr > 2)
verifyConstraints(notNullRelation.analyze.constraints,
ExpressionSet(Seq(resolveColumn(notNullRelation.analyze, "c") > 10,
IsNotNull(resolveColumn(notNullRelation.analyze, "c")),
resolveColumn(notNullRelation.analyze, "a") < 5,
IsNotNull(resolveColumn(notNullRelation.analyze, "a")),
resolveColumn(notNullRelation.analyze, "b") > 2,
IsNotNull(resolveColumn(notNullRelation.analyze, "b")))))

val expand = Expand(
Seq(
Seq('c, Literal.create(null, StringType), 1),
Seq('c, 'a, 2)),
Seq('c, 'a, 'gid.int),
Project(Seq('a, 'c),
notNullRelation))
verifyConstraints(expand.analyze.constraints,
ExpressionSet(Seq.empty[Expression]))
}

test("propagating constraints in aliases") {
val tr = LocalRelation('a.int, 'b.string, 'c.int)

Expand Down