Skip to content

[SPARK-27674][SQL] the hint should not be dropped after cache lookup #24580

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
cloud-fan wants to merge 2 commits into from
Closed

[SPARK-27674][SQL] the hint should not be dropped after cache lookup #24580

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
8934bf4
the hint should not be dropped after cache lookup
cloud-fan May 10, 2019
48e55fa
address comments
cloud-fan May 14, 2019
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
2 changes: 1 addition & 1 deletion ...talyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/EliminateResolvedHint.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -56,7 +56,7 @@ object EliminateResolvedHint extends Rule[LogicalPlan] {
* in this method will be cleaned up later by this rule, and may emit warnings depending on the
* configurations.
*/
private def extractHintsFromPlan(plan: LogicalPlan): (LogicalPlan, Seq[HintInfo]) = {
private[sql] def extractHintsFromPlan(plan: LogicalPlan): (LogicalPlan, Seq[HintInfo]) = {
plan match {
case h: ResolvedHint =>
val (plan, hints) = extractHintsFromPlan(h.child)
Expand Down
22 changes: 12 additions & 10 deletions sql/core/src/main/scala/org/apache/spark/sql/execution/CacheManager.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,8 @@ import org.apache.hadoop.fs.{FileSystem, Path}

import org.apache.spark.internal.Logging
import org.apache.spark.sql.{Dataset, SparkSession}
import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeMap, SubqueryExpression}
import org.apache.spark.sql.catalyst.expressions.{Attribute, SubqueryExpression}
import org.apache.spark.sql.catalyst.optimizer.EliminateResolvedHint
import org.apache.spark.sql.catalyst.plans.logical.{IgnoreCachedData, LogicalPlan, ResolvedHint}
import org.apache.spark.sql.execution.columnar.InMemoryRelation
import org.apache.spark.sql.execution.command.CommandUtils
Expand Down Expand Up @@ -212,17 +213,18 @@ class CacheManager extends Logging {
def useCachedData(plan: LogicalPlan): LogicalPlan = {
val newPlan = plan transformDown {
case command: IgnoreCachedData => command
// Do not lookup the cache by hint node. Hint node is special, we should ignore it when
// canonicalizing plans, so that plans which are same except hint can hit the same cache.
// However, we also want to keep the hint info after cache lookup. Here we skip the hint
// node, so that the returned caching plan won't replace the hint node and drop the hint info
// from the original plan.
case hint: ResolvedHint => hint

case currentFragment =>
lookupCachedData(currentFragment)
.map(_.cachedRepresentation.withOutput(currentFragment.output))
.getOrElse(currentFragment)
lookupCachedData(currentFragment).map { cached =>
// After cache lookup, we should still keep the hints from the input plan.

@gatorsmile gatorsmile May 13, 2019 •
edited
Loading

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

If the original cached plan has a hint, should we keep/respect them? We need to define a clear behavior in our cache manager.

@cloud-fan cloud-fan May 13, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

It doesn't matter, because

  1. as a cache key, the lookup relies on semanticEquals, so having the hint node in the plan has no effect.
  2. the cache lookup returns InMemoryRelation, which has no hint.

I think the behavior is pretty clear: for any query, the hint behavior should be the same no matter some sub-plans are cached or not.

@gatorsmile gatorsmile May 13, 2019

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

Basically, we ignore the hints that are specified in the original cached plans. If users want to use hints, they should specify them in the queries.

val hints = EliminateResolvedHint.extractHintsFromPlan(currentFragment)._2

@gatorsmile gatorsmile May 13, 2019

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

extractHintsFromPlan(currentFragment)._2 was originally a private function. Asking the caller to call reverse is weird. We can add a new function in EliminateResolvedHint or even add a new object for Hint processing.

@cloud-fan cloud-fan May 14, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

It's natural to return the hints in a top-down fashion. And the caller side is free to process the returned hints, including reverse it.

val cachedPlan = cached.cachedRepresentation.withOutput(currentFragment.output)
// The returned hint list is in top-down order. We should reverse it so that the top hint
// is still in the top node.
hints.reverse.foldLeft[LogicalPlan](cachedPlan) { case (p, hint) =>

@maryannxue maryannxue May 11, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

I suppose hints can be lost further down in the tree by matching "canonicalized". Do we need to take care of that as well?

@cloud-fan cloud-fan May 13, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

for hints that don't take effect in the original query, we can drop it.

@cloud-fan cloud-fan May 13, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

Actually we have to drop these un-accessible hints. The cache lookup returns a leaf node InMemoryRelation, and we should only add back the accessible hints.

@maryannxue maryannxue May 13, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

We don't need to drop them, right? Hints are transparent in canonicalization. But I agree the inner hints don't matter, coz they will be replaced with a leaf node anyway.

I'm wondering though, can we change the lookupCachedData instead? like:

def lookupCachedData(plan: LogicalPlan): Option[CachedData] = plan match {
  case ResolvedHint(child, hints) => lookupCachedData(child).map(p => ResolvedHint(p, hints))
  case _ => cachedData.find(cd => plan.sameResult(cd.plan))
}

@cloud-fan cloud-fan May 14, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

What if the plan is Filter(ResolvedHint(...))? This PR is trying to fix the problem that when the hint node is not the root node.

BTW lookupCachedData needs to return a CachedData, so we can't add hint node there.

@gatorsmile gatorsmile May 13, 2019

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

Do we have a test case for covering the logic of reverse?

ResolvedHint(p, hint)

@viirya viirya May 14, 2019

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

Is this the same (semantically) as original cached plan?

We can take one example in added test: broadcast(spark.range(1000)).filter($"id" > 100). Originally, the plan broadcasted is spark.range(1000). After using cached data, seems cached spark.range(1000).filter($"id" > 100) is broadcasted by the hint, actually. It is slightly difference, but maybe in significant effect it might cause?

@cloud-fan cloud-fan May 15, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

The semantic of a hint node is special. By design only join node has hints, so Hint(Filter(Relation)) is the same as Filter(Hint(Relation)), as they both indicate that the left/right sub-tree of a join node has a hint.

@viirya viirya May 15, 2019

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

Ok, I see. Makes sense and it's fine.

}
}.getOrElse(currentFragment)
}

newPlan transformAllExpressions {
Expand Down
49 changes: 33 additions & 16 deletions sql/core/src/test/scala/org/apache/spark/sql/CachedTableSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -939,22 +939,39 @@ class CachedTableSuite extends QueryTest with SQLTestUtils with SharedSQLContext
}

test("Cache should respect the broadcast hint") {
val df = broadcast(spark.range(1000)).cache()
val df2 = spark.range(1000).cache()
df.count()
df2.count()

// Test the broadcast hint.
val joinPlan = df.join(df2, "id").queryExecution.optimizedPlan
val hint = joinPlan.collect {
case Join(_, _, _, _, hint) => hint
}
assert(hint.size == 1)
assert(hint(0).leftHint.get.strategy.contains(BROADCAST))
assert(hint(0).rightHint.isEmpty)

// Clean-up
df.unpersist()
def testHint(df: Dataset[_]): Unit = {
val df2 = spark.range(2000).cache()
df2.count()

def checkHintExists(): Unit = {
// Test the broadcast hint.
val joinPlan = df.join(df2, "id").queryExecution.optimizedPlan
val hint = joinPlan.collect {
case Join(_, _, _, _, hint) => hint
}
assert(hint.size == 1)
assert(hint(0).leftHint.get.strategy.contains(BROADCAST))
assert(hint(0).rightHint.isEmpty)
}

// Make sure the hint does exist when `df` is not cached.
checkHintExists()

df.cache()
df.count()
// Make sure the hint still exists when `df` is cached.
checkHintExists()

// Clean-up
df.unpersist()

@gatorsmile gatorsmile May 13, 2019 •
edited
Loading

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

Use try finally?

finally {
  df.unpersist()
}

}

// The hint is the root node
testHint(broadcast(spark.range(1000)))
// The hint is under subquery alias
testHint(broadcast(spark.range(1000)).as("df"))
// The hint is under filter
testHint(broadcast(spark.range(1000)).filter($"id" > 100))
}

test("analyzes column statistics in cached query") {
Expand Down