spark-issues mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From "Sean Owen (JIRA)" <j...@apache.org>
Subject [jira] [Updated] (SPARK-22481) CatalogImpl.refreshTable is slow
Date Fri, 16 Nov 2018 16:30:00 GMT

     [ https://issues.apache.org/jira/browse/SPARK-22481?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]

Sean Owen updated SPARK-22481:
------------------------------
      Priority: Major  (was: Critical)
    Issue Type: Improvement  (was: Bug)

> CatalogImpl.refreshTable is slow
> --------------------------------
>
>                 Key: SPARK-22481
>                 URL: https://issues.apache.org/jira/browse/SPARK-22481
>             Project: Spark
>          Issue Type: Improvement
>          Components: SQL
>    Affects Versions: 2.1.1, 2.1.2, 2.2.0
>            Reporter: Ran Haim
>            Priority: Major
>
> CatalogImpl.refreshTable was updated in 2.1.1 and since than it has become really slow.
> The cause of the issue is that it is now *always* creates a dataset, and this is redundant
most of the time, we only need the dataset if the table is cached.
> before 2.1.1:
>   override def refreshTable(tableName: String): Unit = {
>     val tableIdent = sparkSession.sessionState.sqlParser.parseTableIdentifier(tableName)
>     // Temp tables: refresh (or invalidate) any metadata/data cached in the plan recursively.
>     // Non-temp tables: refresh the metadata cache.
>     sessionCatalog.refreshTable(tableIdent)
>     // If this table is cached as an InMemoryRelation, drop the original
>     // cached version and make the new version cached lazily.
>     val logicalPlan = sparkSession.sessionState.catalog.lookupRelation(tableIdent)
>     // Use lookupCachedData directly since RefreshTable also takes databaseName.
>     val isCached = sparkSession.sharedState.cacheManager.lookupCachedData(logicalPlan).nonEmpty
>     if (isCached) {
>       // Create a data frame to represent the table.
>       // TODO: Use uncacheTable once it supports database name.
>      {color:red} val df = Dataset.ofRows(sparkSession, logicalPlan){color}
>       // Uncache the logicalPlan.
>       sparkSession.sharedState.cacheManager.uncacheQuery(df, blocking = true)
>       // Cache it again.
>       sparkSession.sharedState.cacheManager.cacheQuery(df, Some(tableIdent.table))
>     }
>   }
> after 2.1.1:
>    override def refreshTable(tableName: String): Unit = {
>     val tableIdent = sparkSession.sessionState.sqlParser.parseTableIdentifier(tableName)
>     // Temp tables: refresh (or invalidate) any metadata/data cached in the plan recursively.
>     // Non-temp tables: refresh the metadata cache.
>     sessionCatalog.refreshTable(tableIdent)
>     // If this table is cached as an InMemoryRelation, drop the original
>     // cached version and make the new version cached lazily.
> {color:red}   val table = sparkSession.table(tableIdent){color}
>     if (isCached(table)) {
>       // Uncache the logicalPlan.
>       sparkSession.sharedState.cacheManager.uncacheQuery(table, blocking = true)
>       // Cache it again.
>       sparkSession.sharedState.cacheManager.cacheQuery(table, Some(tableIdent.table))
>     }
>   }



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@spark.apache.org
For additional commands, e-mail: issues-help@spark.apache.org


Mime
View raw message