[SPARK-27739] df.persist should save stats from optimized plan

• Type: Improvement
• Status: Resolved
• Priority: Minor
• Resolution: Fixed
• Affects Version/s: 3.0.0
• Fix Version/s: 3.0.0
• Component/s: SQL
• Assignee: John Zhuge
• Reporter: John Zhuge
• Created: 16/May/19 00:55
• Updated: 15/Aug/19 03:23
• Resolved: 14/Aug/19 15:16

Description

CacheManager.cacheQuery passes the stats for `planToCache` to InMemoryRelation. Since the plan has not been optimized, the stats is inaccurate because project and filter have not been applied. I'd suggest passing the stats from the optimized plan.

class CacheManager extends Logging {
...
  def cacheQuery(
      query: Dataset[_],
      tableName: Option[String] = None,
      storageLevel: StorageLevel = MEMORY_AND_DISK): Unit = {
    val planToCache = query.logicalPlan
    if (lookupCachedData(planToCache).nonEmpty) {
      logWarning("Asked to cache already cached data.")
    } else {
      val sparkSession = query.sparkSession
      val inMemoryRelation = InMemoryRelation(
        sparkSession.sessionState.conf.useCompression,
        sparkSession.sessionState.conf.columnBatchSize, storageLevel,
        sparkSession.sessionState.executePlan(planToCache).executedPlan,
        tableName,
        planToCache)                  <<<<<==
...
}

object InMemoryRelation {

  def apply(
      useCompression: Boolean,
      batchSize: Int,
      storageLevel: StorageLevel,
      child: SparkPlan,
      tableName: Option[String],
      logicalPlan: LogicalPlan): InMemoryRelation = {
    val cacheBuilder = CachedRDDBuilder(useCompression, batchSize, storageLevel, child, tableName)
    val relation = new InMemoryRelation(child.output, cacheBuilder, logicalPlan.outputOrdering)
    relation.statsOfPlanToCache = logicalPlan.stats           <<<<<==
    relation
  }

Issue Links

• links to: GitHub Pull Request #24623