diff options
author | Steve Loughran <stevel@apache.org> | 2016-08-17 11:42:57 -0700 |
---|---|---|
committer | Marcelo Vanzin <vanzin@cloudera.com> | 2016-08-17 11:43:01 -0700 |
commit | cc97ea188e1d5b8e851d1a8438b8af092783ec04 (patch) | |
tree | 6128ca95597fd4765608e986f501dafcbbccf464 /sql | |
parent | 4d92af310ad29ade039e4130f91f2a3d9180deef (diff) | |
download | spark-cc97ea188e1d5b8e851d1a8438b8af092783ec04.tar.gz spark-cc97ea188e1d5b8e851d1a8438b8af092783ec04.tar.bz2 spark-cc97ea188e1d5b8e851d1a8438b8af092783ec04.zip |
[SPARK-16736][CORE][SQL] purge superfluous fs calls
A review of the code, working back from Hadoop's `FileSystem.exists()` and `FileSystem.isDirectory()` code, then removing uses of the calls when superfluous.
1. delete is harmless if called on a nonexistent path, so don't do any checks before deletes
1. any `FileSystem.exists()` check before `getFileStatus()` or `open()` is superfluous as the operation itself does the check. Instead the `FileNotFoundException` is caught and triggers the downgraded path. When a `FileNotFoundException` was thrown before, the code still creates a new FNFE with the error messages. Though now the inner exceptions are nested, for easier diagnostics.
Initially, relying on Jenkins test runs.
One troublespot here is that some of the codepaths are clearly error situations; it's not clear that they have coverage anyway. Trying to create the failure conditions in tests would be ideal, but it will also be hard.
Author: Steve Loughran <stevel@apache.org>
Closes #14371 from steveloughran/cloud/SPARK-16736-superfluous-fs-calls.
Diffstat (limited to 'sql')
3 files changed, 19 insertions, 19 deletions
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/HDFSBackedStateStoreProvider.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/HDFSBackedStateStoreProvider.scala index 3335755fd3..bec966b15e 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/HDFSBackedStateStoreProvider.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/HDFSBackedStateStoreProvider.scala @@ -17,7 +17,7 @@ package org.apache.spark.sql.execution.streaming.state -import java.io.{DataInputStream, DataOutputStream, IOException} +import java.io.{DataInputStream, DataOutputStream, FileNotFoundException, IOException} import scala.collection.JavaConverters._ import scala.collection.mutable @@ -171,7 +171,7 @@ private[state] class HDFSBackedStateStoreProvider( if (tempDeltaFileStream != null) { tempDeltaFileStream.close() } - if (tempDeltaFile != null && fs.exists(tempDeltaFile)) { + if (tempDeltaFile != null) { fs.delete(tempDeltaFile, true) } logInfo("Aborted") @@ -278,14 +278,12 @@ private[state] class HDFSBackedStateStoreProvider( /** Initialize the store provider */ private def initialize(): Unit = { - if (!fs.exists(baseDir)) { + try { fs.mkdirs(baseDir) - } else { - if (!fs.isDirectory(baseDir)) { + } catch { + case e: IOException => throw new IllegalStateException( - s"Cannot use ${id.checkpointLocation} for storing state data for $this as " + - s"$baseDir already exists and is not a directory") - } + s"Cannot use ${id.checkpointLocation} for storing state data for $this: $e ", e) } } @@ -340,13 +338,16 @@ private[state] class HDFSBackedStateStoreProvider( private def updateFromDeltaFile(version: Long, map: MapType): Unit = { val fileToRead = deltaFile(version) - if (!fs.exists(fileToRead)) { - throw new IllegalStateException( - s"Error reading delta file $fileToRead of $this: $fileToRead does not exist") - } var input: DataInputStream = null + val sourceStream = try { + fs.open(fileToRead) + } catch { + case f: FileNotFoundException => + throw new IllegalStateException( + s"Error reading delta file $fileToRead of $this: $fileToRead does not exist", f) + } try { - input = decompressStream(fs.open(fileToRead)) + input = decompressStream(sourceStream) var eof = false while(!eof) { @@ -405,8 +406,6 @@ private[state] class HDFSBackedStateStoreProvider( private def readSnapshotFile(version: Long): Option[MapType] = { val fileToRead = snapshotFile(version) - if (!fs.exists(fileToRead)) return None - val map = new MapType() var input: DataInputStream = null @@ -443,6 +442,9 @@ private[state] class HDFSBackedStateStoreProvider( } logInfo(s"Read snapshot file for version $version of $this from $fileToRead") Some(map) + } catch { + case _: FileNotFoundException => + None } finally { if (input != null) input.close() } diff --git a/sql/hive/src/test/java/org/apache/spark/sql/hive/JavaMetastoreDataSourcesSuite.java b/sql/hive/src/test/java/org/apache/spark/sql/hive/JavaMetastoreDataSourcesSuite.java index e73117c814..061c7431a6 100644 --- a/sql/hive/src/test/java/org/apache/spark/sql/hive/JavaMetastoreDataSourcesSuite.java +++ b/sql/hive/src/test/java/org/apache/spark/sql/hive/JavaMetastoreDataSourcesSuite.java @@ -75,9 +75,7 @@ public class JavaMetastoreDataSourcesSuite { hiveManagedPath = new Path( catalog.hiveDefaultTableFilePath(new TableIdentifier("javaSavedTable"))); fs = hiveManagedPath.getFileSystem(sc.hadoopConfiguration()); - if (fs.exists(hiveManagedPath)){ - fs.delete(hiveManagedPath, true); - } + fs.delete(hiveManagedPath, true); List<String> jsonObjects = new ArrayList<>(10); for (int i = 0; i < 10; i++) { diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/MetastoreDataSourcesSuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/MetastoreDataSourcesSuite.scala index c36b0275f4..3892fe87e2 100644 --- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/MetastoreDataSourcesSuite.scala +++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/MetastoreDataSourcesSuite.scala @@ -375,7 +375,7 @@ class MetastoreDataSourcesSuite extends QueryTest with SQLTestUtils with TestHiv sessionState.catalog.hiveDefaultTableFilePath(TableIdentifier("ctasJsonTable")) val filesystemPath = new Path(expectedPath) val fs = filesystemPath.getFileSystem(spark.sessionState.newHadoopConf()) - if (fs.exists(filesystemPath)) fs.delete(filesystemPath, true) + fs.delete(filesystemPath, true) // It is a managed table when we do not specify the location. sql( |