[ https://issues.apache.org/jira/browse/SPARK-22481?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16248837#comment-16248837 ]
Ran Haim commented on SPARK-22481: ---------------------------------- Hi, I create this simple test that will show how slow refresh table can be. I create a parquet table, with 10K files just to make it obvious how slow it can get - but a table with a lot of partitions will have the same effect. {code:java} import java.util.UUID import org.apache.hadoop.conf.Configuration import org.apache.hadoop.fs.{FileSystem, Path} import org.apache.spark.SparkConf import org.apache.spark.sql.SparkSession import org.junit.runner.RunWith import org.scalatest.junit.JUnitRunner import org.scalatest.{BeforeAndAfter, FunSuite} @RunWith(classOf[JUnitRunner]) class TestRefreshTable extends FunSuite with BeforeAndAfter{ { if (System.getProperty("os.name").toLowerCase.startsWith("windows")) { val hadoopDirURI = ClassLoader.getSystemResource("hadoop").toURI.getPath System.setProperty("hadoop.home.dir",hadoopDirURI ) } } var fs: FileSystem = _ before{ fs = FileSystem.get(new Configuration()) fs.setWriteChecksum(false) } private val path = "/tmp/test" test("big table refresh test"){ val config = new SparkConf(). setMaster("local[*]"). setAppName("test"). set("spark.ui.enabled", "false") val sparkSession = SparkSession .builder() .config(config) .enableHiveSupport() .getOrCreate() try { for (i <- 1 to 10000) { val id = UUID.randomUUID().toString fs.create(new Path(path, id)) } sparkSession.sql("drop TABLE if exists test") sparkSession.sql( s""" CREATE EXTERNAL TABLE test( `rowId` bigint) ROW FORMAT SERDE 'org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe' STORED AS INPUTFORMAT 'org.apache.hadoop.hive.ql.io.parquet.MapredParquetInputFormat' OUTPUTFORMAT 'org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat' LOCATION '$path' """) val start = System.currentTimeMillis() sparkSession.catalog.refreshTable("test") val end = System.currentTimeMillis() println(s"refresh took ${end - start}") }finally { sparkSession.close() } } } {code} > 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* 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 (v6.4.14#64029) --------------------------------------------------------------------- To unsubscribe, e-mail: issues-unsubscr...@spark.apache.org For additional commands, e-mail: issues-h...@spark.apache.org