Skip to content

[SQL] SPARK-6489: Optimize lateral view with explode to not unnecessary columns. #5358

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

[SQL] SPARK-6489: Optimize lateral view with explode to not unnecessary columns. #5358

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
1b29835
[SQL] SPARK-6489: Optimize lateral view with explode to not read unne…
dreamquster Apr 4, 2015
376d332
[SQL] SPARK-6489: adding test files
dreamquster Apr 4, 2015
644f688
SPARK-6489: explode optimize add project child plan in unittest
dreamquster Apr 8, 2015
9e7aaec
SPARK-6489: explode optimize - push project under generate clause
dreamquster Apr 15, 2015
8909a5d
SPARK-6489: explode optimize - only deal with generate, other
dreamquster Apr 15, 2015
6014acc
SPARK-6489: explode optimize - fixed pruning test
dreamquster Apr 15, 2015
54abc3a
SPARK-6489: explode optimize - restore comparison working directory
dreamquster Apr 15, 2015
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
15 changes: 15 additions & 0 deletions 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 @@ -171,8 +171,23 @@ object ColumnPruning extends Rule[LogicalPlan] {

Project(substitutedProjection, child)

case gen@Generate(generator: Explode, isJoin, isOuter, alias, child) =>
val allReferences = gen.references ++ gen.parentReferences
val pruneProject = prunedChild(child, allReferences)
Generate(generator, isJoin, isOuter, alias, pruneProject)
Copy link
Contributor

@chenghao-intel chenghao-intel Apr 6, 2015

Choose a reason for hiding this comment

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

Is that OK if we just do like

case gen @ Generate(_, _, _, _, child) 
  if (child.outputSet -- gen.references).nonEmpty =>
  gen.copy(child = Project(a.references.toSeq, child))

Copy link
Contributor

@chenghao-intel chenghao-intel Apr 6, 2015

Choose a reason for hiding this comment

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

Then we don't need other code change except the unit test.

Copy link
Author

@dreamquster dreamquster Apr 7, 2015

Choose a reason for hiding this comment

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

Ok,thanks:)


// Eliminate no-op Projects
case Project(projectList, child) if child.output == projectList => child

case plan =>
plan.children.foreach { c =>
c match {
case gen@Generate(generator: Explode, isJoin, isOuter, alias, child) =>
gen.parentReferences = plan.references;
case _ => // nothing
}
}
plan
Copy link
Contributor

@marmbrus marmbrus Apr 8, 2015

Choose a reason for hiding this comment

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

This change is no longer needed.

}

/** Applies a projection only when the child is producing unnecessary attributes */
Expand Down
2 changes: 2 additions & 0 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 @@ -55,6 +55,8 @@ case class Generate(
child: LogicalPlan)
extends UnaryNode {

var parentReferences:AttributeSet = AttributeSet(Seq.empty)

Copy link
Contributor

@marmbrus marmbrus Apr 8, 2015

Choose a reason for hiding this comment

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

This mutable state is no longer needed.

protected def generatorOutput: Seq[Attribute] = {
val output = alias
.map(a => generator.output.map(_.withQualifiers(a :: Nil)))
Expand Down
8 changes: 7 additions & 1 deletion sql/hive/src/main/scala/org/apache/spark/sql/hive/test/TestHive.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -367,7 +367,13 @@ class TestHiveContext(sc: SparkContext) extends HiveContext(sc) {
INSERT OVERWRITE TABLE episodes_part PARTITION (doctor_pt=1)
SELECT title, air_date, doctor FROM episodes
""".cmd
)
),
TestTable("person",
("CREATE TABLE person(name string, age int, data array<INT>) " +
"ROW FORMAT DELIMITED FIELDS TERMINATED BY ',' " +
"COLLECTION ITEMS TERMINATED BY ':'").cmd,
s"LOAD DATA LOCAL INPATH '${getHiveFile("data/files/person.txt")}' INTO TABLE person".cmd
)
)

hiveQTestUtilTables.foreach(registerTestTable)
Expand Down
5 changes: 5 additions & 0 deletions sql/hive/src/test/resources/data/files/person.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,5 @@
A, 20, 10:12:19
B, 25, 7:8:4
C, 19, 12:4:232
D, 73, 243:53:7835
E, 88, 1345:23:532532:353
14 changes: 14 additions & 0 deletions sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/PruningSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -87,6 +87,20 @@ class PruningSuite extends HiveComparisonTest with BeforeAndAfter {
Seq("key"),
Seq.empty)

createPruningTest("Column pruning - explode with aggregate",
"SELECT name, sum(d) AS sumd FROM person LATERAL VIEW explode(data) d AS d GROUP BY name",
Seq("name", "sumd"),
Seq("data","name"),
Seq.empty)
Copy link
Contributor

@marmbrus marmbrus Jun 14, 2015

Choose a reason for hiding this comment

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

I don't think we need to go all the way to hive to create an integration test. Instead create some unit tests in ...catalyst.optimizer.


createPruningTest("Column pruning - outer explode with limit",
"SELECT name FROM person LATERAL VIEW OUTER explode(data) outd AS d limit 3",
Seq("name"),
Seq("data","name"),
Seq.empty)



// Partition pruning tests

createPruningTest("Partition pruning - non-partitioned, non-trivial project",
Expand Down