Skip to content

[SPARK-18703] [SQL] Drop Staging Directories and Data Files After each Insertion/CTAS of Hive serde Tables #16134

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
gatorsmile wants to merge 8 commits into from
Closed

[SPARK-18703] [SQL] Drop Staging Directories and Data Files After each Insertion/CTAS of Hive serde Tables #16134

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
8773c03
fix
gatorsmile Dec 4, 2016
34340ee
address comments.
gatorsmile Dec 4, 2016
365ea26
Merge remote-tracking branch 'upstream/master' into deleteFiles
gatorsmile Dec 10, 2016
1ed228f
address comments.
gatorsmile Dec 10, 2016
74a676c
Merge remote-tracking branch 'upstream/master' into deleteFiles
gatorsmile Dec 14, 2016
16c9da3
address comments.
gatorsmile Dec 14, 2016
5aca883
style
gatorsmile Dec 14, 2016
9da3951
address comments.
gatorsmile Dec 14, 2016
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 13 additions & 0 deletions sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/InsertIntoHiveTable.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,8 @@ import java.net.URI
import java.text.SimpleDateFormat
import java.util.{Date, Locale, Random}

import scala.util.control.NonFatal

import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.hadoop.hive.common.FileUtils
Expand Down Expand Up @@ -85,6 +87,7 @@ case class InsertIntoHiveTable(
def output: Seq[Attribute] = Seq.empty

val hadoopConf = sessionState.newHadoopConf()
val createdTempDir = new scala.collection.mutable.ArrayBuffer[Path]
Copy link
Contributor

@cloud-fan cloud-fan Dec 10, 2016

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

use Option? It looks to me that we will only create one temp dir during one insertion.

Copy link
Member Author

@gatorsmile gatorsmile Dec 10, 2016

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Sure.

val stagingDir = hadoopConf.get("hive.exec.stagingdir", ".hive-staging")

private def executionId: String = {
Expand All @@ -111,6 +114,7 @@ case class InsertIntoHiveTable(
if (!FileUtils.mkdir(fs, dir, true, hadoopConf)) {
throw new IllegalStateException("Cannot create staging directory '" + dir.toString + "'")
}
createdTempDir += dir
fs.deleteOnExit(dir)
} catch {
case e: IOException =>
Expand Down Expand Up @@ -328,6 +332,15 @@ case class InsertIntoHiveTable(
holdDDLTime)
}

// Attempt to delete the staging directory and the inclusive files. If failed, the files are
// expected to be dropped at the normal termination of VM since deleteOnExit is used.
try {
createdTempDir.foreach { path => path.getFileSystem(hadoopConf).delete(path, true) }
Copy link
Contributor

@cloud-fan cloud-fan Dec 4, 2016

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

shall we just delete the tempLocation defined in https://github.com/apache/spark/pull/16134/files#diff-d579db9a8f27e0bbef37720ab14ec3f6R179?

Copy link
Member Author

@gatorsmile gatorsmile Dec 4, 2016

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Through the call of getExternalTmpPath, tmpLocation is pointing to the child directory of the staging directory. For example,
/x/y/z/.hive-staging_hive_2016-12-03_20-56-29_149_4298858301766472202-1/-ext-10000

Here, we want to drop the whole staging directory. For example, /x/y/z/.hive-staging_hive_2016-12-03_20-56-29_149_4298858301766472202-1

At the initial fix, I used tmpLocation and then call getParent to get the parent directory. Later, I am afriad it might look tricky. Thus, I used the solution here.

} catch {
case NonFatal(e) =>
logWarning(s"Unable to delete staging directory: $stagingDir.\n" + e)
}

// Invalidate the cache.
sqlContext.sharedState.cacheManager.invalidateCache(table)
Copy link

@merlintang merlintang Dec 13, 2016

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

should we delete the staging files before or after the invalidateCache? does it matter? logically, we should invalid cache first, then remove the intermediate dataset s.t the cache can be recovered from the file from disks. am i right? please clarify?

Copy link
Member Author

@gatorsmile gatorsmile Dec 13, 2016

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think this does not matter. We do not reuse the temporary files in our implementation.

sqlContext.sessionState.catalog.refreshTable(table.catalogTable.identifier)
Expand Down
23 changes: 23 additions & 0 deletions sql/hive/src/test/scala/org/apache/spark/sql/hive/InsertIntoHiveTableSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -166,6 +166,29 @@ class InsertIntoHiveTableSuite extends QueryTest with TestHiveSingleton with Bef
sql("DROP TABLE tmp_table")
}

test("Delete the temporary staging directory and files after each insert") {
Copy link
Contributor

@cloud-fan cloud-fan Dec 11, 2016

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

let's wait for #16104 so that we can move this test to VersionSuite to test all supported hive versions.

Copy link
Member Author

@gatorsmile gatorsmile Dec 11, 2016

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Sure.

withTable("tab") {
val tmpDir = Utils.createTempDir()
Copy link
Member

@dongjoon-hyun dongjoon-hyun Dec 4, 2016

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

withTempDir instead of this?

Copy link
Member Author

@gatorsmile gatorsmile Dec 4, 2016

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Sure. Thanks!

sql(
s"""
|CREATE TABLE tab(c1 string)
|location '${tmpDir.toURI.toString}'
""".stripMargin)

(1 to 3).map { i =>
sql(s"INSERT OVERWRITE TABLE tab SELECT '$i'")
}
def listFiles(path: File): List[String] = {
val dir = path.listFiles()
val folders = dir.filter(_.isDirectory).toList
val filePaths = dir.map(_.getName).toList
filePaths ::: folders.flatMap(listFiles)
}
val expectedFiles = ".part-00000.crc" :: "part-00000" :: Nil
assert(listFiles(tmpDir).sortBy(_.toString) == expectedFiles)
}
}

test("INSERT OVERWRITE - partition IF NOT EXISTS") {
withTempDir { tmpDir =>
val table = "table_with_partition"
Expand Down