Address gatorsmile's comments

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/UnsupportedOperationChecker.scala

@@ -142,7 +142,6 @@ object UnsupportedOperationChecker {
             "Distinct aggregations are not supported on streaming DataFrames/Datasets. Consider " +
               "using approx_count_distinct() instead.")
 
-
         case _: Command =>
           throwError("Commands like CreateTable*, AlterTable*, Show* are not supported with " +
             "streaming DataFrames/Datasets")

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala

@@ -210,7 +210,7 @@ class AstBuilder(conf: SQLConf) extends SqlBaseBaseVisitor[AnyRef] with Logging
       case hiveDir: InsertOverwriteHiveDirContext => visitInsertOverwriteHiveDir(hiveDir)
     }
 
-    InsertIntoDir(isLocal, storage, provider, query)
+    InsertIntoDir(isLocal, storage, provider, query, overwrite = true)
   }
 
   /**

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scala

@@ -363,7 +363,8 @@ case class InsertIntoDir(
     isLocal: Boolean,
     storage: CatalogStorageFormat,
     provider: Option[String],
-    child: LogicalPlan)
+    child: LogicalPlan,
+    overwrite: Boolean = true)
   extends LogicalPlan {
 
   override def children: Seq[LogicalPlan] = child :: Nil

sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala

@@ -1556,7 +1556,7 @@ class SparkSqlAstBuilder(conf: SQLConf) extends AstBuilder(conf) {
    *
    * Expected format:
    * {{{
-   *   INSERT OVERWRITE DIRECTORY
+   *   INSERT OVERWRITE [LOCAL] DIRECTORY
    *   path
    *   [ROW FORMAT row_format]
    *   [STORED AS file_format]

sql/core/src/main/scala/org/apache/spark/sql/execution/command/InsertIntoDataSourceDirCommand.scala

@@ -35,13 +35,14 @@ import org.apache.spark.sql.execution.datasources._
 case class InsertIntoDataSourceDirCommand(
     storage: CatalogStorageFormat,
     provider: Option[String],
-    query: LogicalPlan) extends RunnableCommand {
+    query: LogicalPlan,
+    overwrite: Boolean) extends RunnableCommand {
 
   override def innerChildren: Seq[LogicalPlan] = Seq(query)
 
   override def run(sparkSession: SparkSession): Seq[Row] = {
     assert(innerChildren.length == 1)
-    assert(!storage.locationUri.isEmpty)
+    assert(storage.locationUri.nonEmpty)
     assert(provider.isDefined)
 
     // Create the relation based on the input logical plan: `data`.
@@ -52,8 +53,9 @@ case class InsertIntoDataSourceDirCommand(
       options = storage.properties ++ pathOption,
       catalogTable = None)
 
+    val saveMode = if (overwrite) SaveMode.Overwrite else SaveMode.ErrorIfExists
     try {
-      dataSource.writeAndRead(SaveMode.Overwrite, query)
+      dataSource.writeAndRead(saveMode, query)
     } catch {
       case ex: AnalysisException =>
         logError(s"Failed to write to directory " + storage.locationUri.toString, ex)

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSourceStrategy.scala

@@ -141,8 +141,8 @@ case class DataSourceAnalysis(conf: SQLConf) extends Rule[LogicalPlan] with Cast
         parts, query, overwrite, false) if parts.isEmpty =>
       InsertIntoDataSourceCommand(l, query, overwrite)
 
-    case InsertIntoDir(_, storage, provider, query) =>
-      InsertIntoDataSourceDirCommand(storage, provider, query)
+    case InsertIntoDir(_, storage, provider, query, overwrite) if provider.nonEmpty =>
+      InsertIntoDataSourceDirCommand(storage, provider, query, overwrite)
 
     case i @ InsertIntoTable(
         l @ LogicalRelation(t: HadoopFsRelation, _, table), parts, query, overwrite, _) =>

sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLCommandSuite.scala

@@ -490,7 +490,7 @@ class DDLCommandSuite extends PlanTest {
   test("insert overwrite directory") {
     val v1 = "INSERT OVERWRITE DIRECTORY '/tmp/file' USING parquet SELECT 1 as a"
     parser.parsePlan(v1) match {
-      case InsertIntoDir(_, storage, provider, query) =>
+      case InsertIntoDir(_, storage, provider, query, overwrite) =>
         assert(storage.locationUri != None && storage.locationUri.get.toString == "/tmp/file")
       case other =>
         fail(s"Expected to parse ${classOf[InsertIntoDataSourceDirCommand].getClass.getName}" +
@@ -510,7 +510,7 @@ class DDLCommandSuite extends PlanTest {
         | SELECT 1 as a
       """.stripMargin
     parser.parsePlan(v3) match {
-      case InsertIntoDir(_, storage, provider, query) =>
+      case InsertIntoDir(_, storage, provider, query, overwrite) =>
         assert(storage.locationUri != None && provider == Some("json"))
       case other =>
         fail(s"Expected to parse ${classOf[InsertIntoDataSourceDirCommand].getClass.getName}" +

sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveStrategies.scala

@@ -157,8 +157,8 @@ object HiveAnalysis extends Rule[LogicalPlan] {
     case CreateTable(tableDesc, mode, Some(query)) if DDLUtils.isHiveTable(tableDesc) =>
       CreateHiveTableAsSelectCommand(tableDesc, query, mode)
 
-    case InsertIntoDir(isLocal, storage, _, child) =>
-      InsertIntoHiveDirCommand(isLocal, storage, child)
+    case InsertIntoDir(isLocal, storage, _, child, overwrite) =>
+      InsertIntoHiveDirCommand(isLocal, storage, child, overwrite)
   }
 }
 

sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/InsertIntoHiveDirCommand.scala

@@ -38,13 +38,14 @@ import org.apache.spark.util.Utils
 case class InsertIntoHiveDirCommand(
     isLocal: Boolean,
     storage: CatalogStorageFormat,
-    query: LogicalPlan) extends SaveAsHiveFile {
+    query: LogicalPlan,
+    overwrite: Boolean) extends SaveAsHiveFile {
 
   override def children: Seq[LogicalPlan] = query :: Nil
 
   override def run(sparkSession: SparkSession, children: Seq[SparkPlan]): Seq[Row] = {
     assert(children.length == 1)
-    assert(!storage.locationUri.isEmpty)
+    assert(storage.locationUri.nonEmpty)
 
     val Array(cols, types) = children.head.output.foldLeft(Array("", "")) { case (r, a) =>
       r(0) = r(0) + a.name + ","
@@ -79,14 +80,22 @@ case class InsertIntoHiveDirCommand(
         val localFileSystem = FileSystem.getLocal(jobConf)
         val localPath = localFileSystem.makeQualified(targetPath)
         if (localFileSystem.exists(localPath)) {
-          localFileSystem.delete(localPath, true)
+          if (overwrite) {
+            localFileSystem.delete(localPath, true)
+          } else {
+            throw new RuntimeException("Directory '" + localPath.toString + "' already exists")
+          }
         }
         localPath
       } else {
         val qualifiedPath = FileUtils.makeQualified(targetPath, hadoopConf)
         val dfs = qualifiedPath.getFileSystem(jobConf)
         if (dfs.exists(qualifiedPath)) {
-          dfs.delete(qualifiedPath, true)
+          if (overwrite) {
+            dfs.delete(qualifiedPath, true)
+          } else {
+            throw new RuntimeException("Directory '" + qualifiedPath.toString + "' already exists")
+          }
         } else {
           dfs.mkdirs(qualifiedPath.getParent)
         }

sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/SaveAsHiveFile.scala

@@ -31,14 +31,15 @@ import org.apache.spark.sql.hive.HiveShim.{ShimFileSinkDesc => FileSinkDesc}
 // Base trait from which all hive insert statement physical execution extends.
 private[hive] trait SaveAsHiveFile extends DataWritingCommand {
 
-  protected def saveAsHiveFile(sparkSession: SparkSession,
-                               plan: SparkPlan,
-                               hadoopConf: Configuration,
-                               fileSinkConf: FileSinkDesc,
-                               outputLocation: String,
-                               partitionAttributes: Seq[Attribute] = Nil,
-                               bucketSpec: Option[BucketSpec] = None,
-                               options: Map[String, String] = Map.empty): Unit = {
+  protected def saveAsHiveFile(
+      sparkSession: SparkSession,
+      plan: SparkPlan,
+      hadoopConf: Configuration,
+      fileSinkConf: FileSinkDesc,
+      outputLocation: String,
+      partitionAttributes: Seq[Attribute] = Nil,
+      bucketSpec: Option[BucketSpec] = None,
+      options: Map[String, String] = Map.empty): Unit = {
 
     val sessionState = sparkSession.sessionState
 

sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/SQLQuerySuite.scala

@@ -2046,74 +2046,81 @@ class SQLQuerySuite extends QueryTest with SQLTestUtils with TestHiveSingleton {
 
     val path = Utils.createTempDir()
     path.delete()
-    checkAnswer(
-      sql(s"INSERT OVERWRITE LOCAL DIRECTORY '${path.toString}' SELECT * FROM src where key < 10"),
-      Seq.empty[Row])
+    withTempDir { dir =>
+      val path = dir.toURI.getPath
 
-    checkAnswer(
-      sql(s"""INSERT OVERWRITE LOCAL DIRECTORY '${path.toString}'
+      checkAnswer(
+        sql(s"INSERT OVERWRITE LOCAL DIRECTORY '${path}' SELECT * FROM src where key < 10"),
+        Seq.empty[Row])
+
+      checkAnswer(
+        sql(
+          s"""
+             |INSERT OVERWRITE LOCAL DIRECTORY '${path}'
              |STORED AS orc
-             |SELECT * FROM src where key < 10""".stripMargin),
-      Seq.empty[Row])
+             |SELECT * FROM src where key < 10
+          """.stripMargin),
+        Seq.empty[Row])
 
-    // use orc data source to check the data of path is right.
-    sql(
-      s"""CREATE TEMPORARY TABLE orc_source
-         |USING org.apache.spark.sql.hive.orc
-         |OPTIONS (
-         |  PATH '${path.getCanonicalPath}'
-         |)
-       """.stripMargin)
-    checkAnswer(
-      sql("select * from orc_source"),
-      sql("select * from src where key < 10").collect()
-    )
+      // use orc data source to check the data of path is right.
+      sql(
+        s"""
+           |CREATE TEMPORARY TABLE orc_source
+           |USING org.apache.spark.sql.hive.orc
+           |OPTIONS (
+           |  PATH '${dir.getCanonicalPath}'
+           |)
+        """.stripMargin)
 
-    Utils.deleteRecursively(path)
-    dropTempTable("orc_source")
+      checkAnswer(
+        sql("select * from orc_source"),
+        sql("select * from src where key < 10").collect())
+
+      dropTempTable("orc_source")
+    }
   }
 
   test("insert overwrite to dir from temp table") {
-    import org.apache.spark.util.Utils
+    withTempView("test_insert_table") {
+      spark.range(10).selectExpr("id", "id AS str").createOrReplaceTempView("test_insert_table")
 
-    sparkContext
-      .parallelize(1 to 10)
-      .map(i => TestData(i, i.toString))
-      .toDF()
-      .registerTempTable("test_insert_table")
+      withTempDir { dir =>
+        val path = dir.toURI.getPath
 
-    val path = Utils.createTempDir()
-    path.delete()
-    checkAnswer(
-      sql(
-        s"""
-           |INSERT OVERWRITE LOCAL DIRECTORY '${path.toString}'
-           |ROW FORMAT DELIMITED FIELDS TERMINATED BY ','
-           |SELECT * FROM test_insert_table
-         """.stripMargin),
-      Seq.empty[Row])
+        checkAnswer(
+          sql(
+            s"""
+               |INSERT OVERWRITE LOCAL DIRECTORY '${path}'
+               |ROW FORMAT DELIMITED FIELDS TERMINATED BY ','
+               |SELECT * FROM test_insert_table
+             """.stripMargin),
+          Seq.empty[Row])
 
-    checkAnswer(
-      sql(s"""
-        INSERT OVERWRITE LOCAL DIRECTORY '${path.toString}'
-             |STORED AS orc
-             |SELECT * FROM test_insert_table""".stripMargin),
-      Seq.empty[Row])
+        checkAnswer(
+          sql(
+            s"""
+               |INSERT OVERWRITE LOCAL DIRECTORY '${path}'
+               |STORED AS orc
+               |SELECT * FROM test_insert_table
+             """.stripMargin),
+          Seq.empty[Row])
 
-    // use orc data source to check the data of path is right.
-    sql(
-      s"""CREATE TEMPORARY TABLE orc_source
-         |USING org.apache.spark.sql.hive.orc
-         |OPTIONS (
-         |  PATH '${path.getCanonicalPath}'
-         |)
-       """.stripMargin)
-    checkAnswer(
-      sql("select * from orc_source"),
-      sql("select * from test_insert_table").collect()
-    )
-    Utils.deleteRecursively(path)
-    dropTempTable("test_insert_table")
-    dropTempTable("orc_source")
+        // use orc data source to check the data of path is right.
+        sql(
+          s"""
+             |CREATE TEMPORARY TABLE orc_source
+             |USING org.apache.spark.sql.hive.orc
+             |OPTIONS (
+             |  PATH '${dir.getCanonicalPath}'
+             |)
+           """.stripMargin)
+
+        checkAnswer(
+          sql("select * from orc_source"),
+          sql("select * from test_insert_table").collect())
+
+        dropTempTable("orc_source")
+      }
+    }
   }
 }