[
https://issues.apache.org/jira/browse/SPARK-22481?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Ran Haim updated SPARK-22481:
-----------------------------
Description:
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* create a dataset, and this is
redundent most of the time, we only need the dataset if the table is cached.
code before the change:
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 the change:
/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))
}
}
was:
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* create a dataset, and this is
redundent most of the time, we only need the dataset if the table is cached.
code before the change:
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.
* val df = Dataset.ofRows(sparkSession, logicalPlan)*
// Uncache the logicalPlan.
sparkSession.sharedState.cacheManager.uncacheQuery(df, blocking = true)
// Cache it again.
sparkSession.sharedState.cacheManager.cacheQuery(df,
Some(tableIdent.table))
}
}
after the change:
/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 table = sparkSession.table(tableIdent)*
if (isCached(table)) {
// Uncache the logicalPlan.
sparkSession.sharedState.cacheManager.uncacheQuery(table, blocking = true)
// Cache it again.
sparkSession.sharedState.cacheManager.cacheQuery(table,
Some(tableIdent.table))
}
}
> CatalogImpl.refreshTable is slow
> --------------------------------
>
> Key: SPARK-22481
> URL: https://issues.apache.org/jira/browse/SPARK-22481
> Project: Spark
> Issue Type: Bug
> Components: SQL
> Affects Versions: 2.1.1, 2.1.2, 2.2.0
> Reporter: Ran Haim
> Priority: Critical
>
> 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* create a dataset, and this
> is redundent most of the time, we only need the dataset if the table is
> cached.
> code before the change:
> 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 the change:
> /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
(v6.4.14#64029)
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscr...@spark.apache.org
For additional commands, e-mail: issues-h...@spark.apache.org
