diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala
old mode 100644
new mode 100755
index c4edadba278f..83c66d0c27b6
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala
@@ -1396,6 +1396,16 @@ class SparkSqlAstBuilder(conf: SQLConf) extends AstBuilder(conf) {
       compressed = false,
       properties = rowStorage.properties ++ fileStorage.properties)
 
-    (ctx.LOCAL != null, storage, Some(DDLUtils.HIVE_PROVIDER))
+    val fileFormat = extractFileFormat(fileStorage.serde)
+    (ctx.LOCAL != null, storage, Some(fileFormat))
+  }
+
+  private def extractFileFormat(serde: Option[String]): String = {
+    serde.map({ x =>
+      val lowerCaseSerde = x.toLowerCase(Locale.ROOT)
+      if (lowerCaseSerde.contains("parquet")) "parquet"
+      else if (lowerCaseSerde.contains("orc")) "orc"
+      else DDLUtils.HIVE_PROVIDER
+    }).getOrElse(DDLUtils.HIVE_PROVIDER)
   }
 }
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/sources/InsertSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/sources/InsertSuite.scala
old mode 100644
new mode 100755
index 4f1ae069d4b8..0b755dcb5f78
--- a/sql/core/src/test/scala/org/apache/spark/sql/sources/InsertSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/sources/InsertSuite.scala
@@ -434,14 +434,13 @@ class InsertSuite extends DataSourceTest with SharedSQLContext {
     }
   }
 
-  test("Insert overwrite directory using Hive serde without turning on Hive support") {
+  test("Insert overwrite directory in hive table without turning on Hive support") {
     withTempDir { dir =>
       val path = dir.toURI.getPath
       val e = intercept[AnalysisException] {
         sql(
           s"""
              |INSERT OVERWRITE LOCAL DIRECTORY '$path'
-             |STORED AS orc
              |SELECT 1, 2
            """.stripMargin)
       }.getMessage
diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveDDLSuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveDDLSuite.scala
old mode 100644
new mode 100755
index a907fcae526c..c85d1827e763
--- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveDDLSuite.scala
+++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveDDLSuite.scala
@@ -2516,4 +2516,25 @@ class HiveDDLSuite
       }
     }
   }
+
+  test("SPARK-28659: hive insert overwrite directory should use data source " +
+    "if it is convertible") {
+    Seq("orc", "parquet").foreach { format =>
+      withTable("t1") {
+        sql(s"CREATE TABLE t1 (id int)")
+        sql(s"INSERT INTO t1 SELECT * FROM range(1, 100, 1, 10)")
+        withTempDir { dir =>
+          val pathUri = dir.toURI
+          sql(
+            s"""
+               |INSERT OVERWRITE DIRECTORY '${pathUri}'
+               |STORED AS $format
+               |SELECT * FROM t1
+           """.stripMargin)
+          val snappyFile = dir.listFiles().find(_.getName.startsWith("part"))
+          assertCompression(snappyFile, format, "SNAPPY")
+        }
+      }
+    }
+  }
 }