Wenchen's comments

Author: jzhuge

sql/core/src/main/scala/org/apache/spark/sql/DataFrameWriter.scala

@@ -361,6 +361,16 @@ final class DataFrameWriter[T] private[sql](ds: Dataset[T]) {
   def insertInto(tableName: String): Unit = {
     import df.sparkSession.sessionState.analyzer.{AsTableIdentifier, CatalogObjectIdentifier}
 
+    assertNotBucketed("insertInto")
+
+    if (partitioningColumns.isDefined) {
+      throw new AnalysisException(
+        "insertInto() can't be used together with partitionBy(). " +
+          "Partition columns have already been defined for the table. " +
+          "It is not necessary to use partitionBy()."
+      )
+    }
+
     df.sparkSession.sessionState.sqlParser.parseMultipartIdentifier(tableName) match {
       case CatalogObjectIdentifier(Some(catalog), ident) =>
         insertInto(catalog, ident)
@@ -372,16 +382,6 @@ final class DataFrameWriter[T] private[sql](ds: Dataset[T]) {
   private def insertInto(catalog: CatalogPlugin, ident: Identifier): Unit = {
     import org.apache.spark.sql.catalog.v2.CatalogV2Implicits._
 
-    assertNotBucketed("insertInto")
-
-    if (partitioningColumns.isDefined) {
-      throw new AnalysisException(
-        "insertInto() can't be used together with partitionBy(). " +
-          "Partition columns have already been defined for the table. " +
-          "It is not necessary to use partitionBy()."
-      )
-    }
-
     val table = DataSourceV2Relation.create(catalog.asTableCatalog.loadTable(ident))
 
     val command = modeForDSV2 match {
@@ -410,16 +410,6 @@ final class DataFrameWriter[T] private[sql](ds: Dataset[T]) {
   }
 
   private def insertInto(tableIdent: TableIdentifier): Unit = {
-    assertNotBucketed("insertInto")
-
-    if (partitioningColumns.isDefined) {
-      throw new AnalysisException(
-        "insertInto() can't be used together with partitionBy(). " +
-          "Partition columns have already been defined for the table. " +
-          "It is not necessary to use partitionBy()."
-      )
-    }
-
     runCommand(df.sparkSession, "insertInto") {
       InsertIntoTable(
         table = UnresolvedRelation(tableIdent),

sql/core/src/test/scala/org/apache/spark/sql/sources/v2/DataSourceV2DataFrameSuite.scala

@@ -62,7 +62,7 @@ class DataSourceV2DataFrameSuite extends QueryTest with SharedSQLContext with Be
     }
   }
 
-  test("insertInto: append partitioned table - dynamic clause") {
+  test("insertInto: append partitioned table") {
     val t1 = "testcat.ns1.ns2.tbl"
     withTable(t1) {
       sql(s"CREATE TABLE $t1 (id bigint, data string) USING foo PARTITIONED BY (id)")
@@ -80,7 +80,7 @@ class DataSourceV2DataFrameSuite extends QueryTest with SharedSQLContext with Be
     }
   }
 
-  test("insertInto: overwrite - dynamic clause - static mode") {
+  test("insertInto: overwrite - static mode") {
     withSQLConf(PARTITION_OVERWRITE_MODE.key -> PartitionOverwriteMode.STATIC.toString) {
       val t1 = "testcat.ns1.ns2.tbl"
       withTable(t1) {
@@ -92,7 +92,7 @@ class DataSourceV2DataFrameSuite extends QueryTest with SharedSQLContext with Be
     }
   }
 
-  test("insertInto: overwrite - dynamic clause - dynamic mode") {
+  test("insertInto: overwrite - dynamic mode") {
     withSQLConf(PARTITION_OVERWRITE_MODE.key -> PartitionOverwriteMode.DYNAMIC.toString) {
       val t1 = "testcat.ns1.ns2.tbl"
       withTable(t1) {