[SPARK-23195] [SQL] Keep the Hint of Cached Data #20368
Conversation
| val df1 = spark.createDataFrame(Seq((1, "4"), (2, "2"))).toDF("key", "value") | ||
| val df2 = spark.createDataFrame(Seq((1, "1"), (2, "2"))).toDF("key", "value") | ||
| broadcast(df2).cache() | ||
| if (materialized) df2.collect() |
There was a problem hiding this comment.
This PR #19864 accidentally fixes the issue when the plan is not materialized. However, it does not resolve the issue when the cached plan is materialized.
| @@ -77,7 +77,7 @@ case class InMemoryRelation( | |||
| // Underlying columnar RDD hasn't been materialized, use the stats from the plan to cache | |||
| statsOfPlanToCache | |||
| } else { | |||
| Statistics(sizeInBytes = batchStats.value.longValue) | |||
| Statistics(sizeInBytes = batchStats.value.longValue, hints = statsOfPlanToCache.hints) | |||
There was a problem hiding this comment.
Why don't you simply statsOfPlanToCache.copy(sizeInBytes = batchStats.value.longValue)?
There was a problem hiding this comment.
That is misleading. Conceptually, that is wrong. The values should be filled by the actual values from the materialized results.
| test("broadcast hint is retained in a cached plan") { | ||
| Seq(true, false).foreach { materialized => | ||
| withSQLConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1") { | ||
| val df1 = spark.createDataFrame(Seq((1, "4"), (2, "2"))).toDF("key", "value") |
There was a problem hiding this comment.
Is spark.createDataFrame(...) wrapper really required? I thought Seq((1, "4"), (2, "2")).toDF("key", "value") would just work fine.
There was a problem hiding this comment.
That should not matter.
| val df2 = spark.createDataFrame(Seq((1, "1"), (2, "2"))).toDF("key", "value") | ||
| broadcast(df2).cache() | ||
| if (materialized) df2.collect() | ||
| val df3 = df1.join(df2, Seq("key"), "inner") |
There was a problem hiding this comment.
val df3 = df1.join(df2, "key")? inner is implied, isn't it? (I'm proposing the change as this and other tests could be easily used as a learning tool to master Spark SQL's API)
There was a problem hiding this comment.
That should not matter.
There was a problem hiding this comment.
I tend to agree that tests are also examples for Spark users, we should pick the recommended usages.
There was a problem hiding this comment.
All the other cases are creating Dataframes like this. Anyway, I changed all of them in the new PR.
|
LGTM |
| @@ -63,7 +63,7 @@ case class InMemoryRelation( | |||
| tableName: Option[String])( | |||
| @transient var _cachedColumnBuffers: RDD[CachedBatch] = null, | |||
| val batchStats: LongAccumulator = child.sqlContext.sparkContext.longAccumulator, | |||
| statsOfPlanToCache: Statistics = null) | |||
| statsOfPlanToCache: Statistics) | |||
There was a problem hiding this comment.
leave no default value is fine, we do not any default value actually
There was a problem hiding this comment.
Setting null by default is risky, because we might hit NullPointerException .
|
Test build #86545 has finished for PR 20368 at commit
|
|
Thanks! Merged to master/2.3 |
## What changes were proposed in this pull request?
The broadcast hint of the cached plan is lost if we cache the plan. This PR is to correct it.
```Scala
val df1 = spark.createDataFrame(Seq((1, "4"), (2, "2"))).toDF("key", "value")
val df2 = spark.createDataFrame(Seq((1, "1"), (2, "2"))).toDF("key", "value")
broadcast(df2).cache()
df2.collect()
val df3 = df1.join(df2, Seq("key"), "inner")
```
## How was this patch tested?
Added a test.
Author: gatorsmile <gatorsmile@gmail.com>
Closes #20368 from gatorsmile/cachedBroadcastHint.
(cherry picked from commit 44cc4da)
Signed-off-by: gatorsmile <gatorsmile@gmail.com>
|
Hi, All. This seems to break both
|
|
As a result, this seems to block SparkPullRequestBuilder, too. |
|
I am reverting this PR. |
|
Oh, thank you so much for fast recovery, @gatorsmile . |
|
Thanks! revert it from master/2.3 |
What changes were proposed in this pull request?
The broadcast hint of the cached plan is lost if we cache the plan. This PR is to correct it.
How was this patch tested?
Added a test.