This is an automated email from the ASF dual-hosted git repository. dongjoon pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push: new db0f6b4 [SPARK-27961][SQL] DataSourceV2Relation should not have refresh method db0f6b4 is described below commit db0f6b46749dc5cddd54c45e0dc5f95574c9e4f1 Author: Gengliang Wang <gengliang.w...@databricks.com> AuthorDate: Sat Jun 8 10:59:10 2019 -0700 [SPARK-27961][SQL] DataSourceV2Relation should not have refresh method ## What changes were proposed in this pull request? The newly added `Refresh` method in PR #24401 prevented the work of moving DataSourceV2Relation into catalyst. It calls `case table: FileTable => table.fileIndex.refresh()` while `FileTable` belongs to sql/core. More importantly, Ryan Blue pointed out DataSourceV2Relation is immutable by design, it should not have refresh method. ## How was this patch tested? Unit test Closes #24815 from gengliangwang/removeRefreshTable. Authored-by: Gengliang Wang <gengliang.w...@databricks.com> Signed-off-by: Dongjoon Hyun <dh...@apple.com> --- .../datasources/v2/DataSourceV2Relation.scala | 5 ---- .../datasources/v2/FilePartitionReader.scala | 3 +-- .../org/apache/spark/sql/MetadataCacheSuite.scala | 27 +++++++++++----------- 3 files changed, 14 insertions(+), 21 deletions(-) diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Relation.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Relation.scala index 27875b3..fc91943 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Relation.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Relation.scala @@ -68,11 +68,6 @@ case class DataSourceV2Relation( override def newInstance(): DataSourceV2Relation = { copy(output = output.map(_.newInstance())) } - - override def refresh(): Unit = table match { - case table: FileTable => table.fileIndex.refresh() - case _ => // Do nothing. - } } /** diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/FilePartitionReader.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/FilePartitionReader.scala index d4bad29..e88c751 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/FilePartitionReader.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/FilePartitionReader.scala @@ -47,8 +47,7 @@ class FilePartitionReader[T](readers: Iterator[PartitionedFileReader[T]]) e.getMessage + "\n" + "It is possible the underlying files have been updated. " + "You can explicitly invalidate the cache in Spark by " + - "running 'REFRESH TABLE tableName' command in SQL or " + - "by recreating the Dataset/DataFrame involved.") + "recreating the Dataset/DataFrame involved.") case e @ (_: RuntimeException | _: IOException) if ignoreCorruptFiles => logWarning( s"Skipped the rest of the content in the corrupted file: $currentReader", e) diff --git a/sql/core/src/test/scala/org/apache/spark/sql/MetadataCacheSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/MetadataCacheSuite.scala index 664d59c..602951b 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/MetadataCacheSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/MetadataCacheSuite.scala @@ -29,7 +29,7 @@ import org.apache.spark.sql.test.SharedSQLContext abstract class MetadataCacheSuite extends QueryTest with SharedSQLContext { /** Removes one data file in the given directory. */ - private def deleteOneFileInDirectory(dir: File): Unit = { + protected def deleteOneFileInDirectory(dir: File): Unit = { assert(dir.isDirectory) val oneFile = dir.listFiles().find { file => !file.getName.startsWith("_") && !file.getName.startsWith(".") @@ -38,10 +38,9 @@ abstract class MetadataCacheSuite extends QueryTest with SharedSQLContext { oneFile.foreach(_.delete()) } - test("SPARK-16336,SPARK-27504 Suggest doing table refresh " + - "when encountering FileNotFoundException") { + test("SPARK-16336,SPARK-27961 Suggest fixing FileNotFoundException") { withTempPath { (location: File) => - // Create a Parquet directory + // Create an ORC directory spark.range(start = 0, end = 100, step = 1, numPartitions = 3) .write.orc(location.getAbsolutePath) @@ -57,13 +56,20 @@ abstract class MetadataCacheSuite extends QueryTest with SharedSQLContext { df.count() } assert(e.getMessage.contains("FileNotFoundException")) - assert(e.getMessage.contains("REFRESH")) + assert(e.getMessage.contains("recreating the Dataset/DataFrame involved")) } } +} + +class MetadataCacheV1Suite extends MetadataCacheSuite { + override protected def sparkConf: SparkConf = + super + .sparkConf + .set(SQLConf.USE_V1_SOURCE_READER_LIST, "orc") - test("SPARK-16337,SPARK-27504 temporary view refresh") { + test("SPARK-16337 temporary view refresh") { withTempView("view_refresh") { withTempPath { (location: File) => - // Create a Parquet directory + // Create an ORC directory spark.range(start = 0, end = 100, step = 1, numPartitions = 3) .write.orc(location.getAbsolutePath) @@ -113,13 +119,6 @@ abstract class MetadataCacheSuite extends QueryTest with SharedSQLContext { } } -class MetadataCacheV1Suite extends MetadataCacheSuite { - override protected def sparkConf: SparkConf = - super - .sparkConf - .set(SQLConf.USE_V1_SOURCE_READER_LIST, "orc") -} - class MetadataCacheV2Suite extends MetadataCacheSuite { override protected def sparkConf: SparkConf = super --------------------------------------------------------------------- To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org For additional commands, e-mail: commits-h...@spark.apache.org