address comments

Author: cloud-fan

sql/core/src/main/java/org/apache/spark/sql/sources/v2/TableProvider.java

@@ -27,7 +27,8 @@
  * have a public, 0-arg constructor.
  * <p>
  * Note that, TableProvider can only apply data operations to existing tables, like read, append,
- * delete, and overwrite. Not operations that require metadata changes, like create/drop tables.
+ * delete, and overwrite. It does not support the operations that require metadata changes, like
+ * create/drop tables.
  * <p>
  * The major responsibility of this interface is to return a {@link Table} for read/write.
  * </p>

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

@@ -58,7 +58,7 @@ final class DataFrameWriter[T] private[sql](ds: Dataset[T]) {
    * <li>`SaveMode.ErrorIfExists`: throw an exception at runtime.</li>
    * </ul>
    * <p>
-   * When writing to data source v1, the default option is `ErrorIfExist`. When writing to data
+   * When writing to data source v1, the default option is `ErrorIfExists`. When writing to data
    * source v2, the default option is `Append`.
    *
    * @since 1.4.0
@@ -80,14 +80,15 @@ final class DataFrameWriter[T] private[sql](ds: Dataset[T]) {
    * @since 1.4.0
    */
   def mode(saveMode: String): DataFrameWriter[T] = {
-    mode(saveMode.toLowerCase(Locale.ROOT) match {
-      case "overwrite" => SaveMode.Overwrite
-      case "append" => SaveMode.Append
-      case "ignore" => SaveMode.Ignore
-      case "error" | "errorifexists" | "default" => SaveMode.ErrorIfExists
+    saveMode.toLowerCase(Locale.ROOT) match {
+      case "overwrite" => mode(SaveMode.Overwrite)
+      case "append" => mode(SaveMode.Append)
+      case "ignore" => mode(SaveMode.Ignore)
+      case "error" | "errorifexists" => mode(SaveMode.ErrorIfExists)
+      case "default" => this
       case _ => throw new IllegalArgumentException(s"Unknown save mode: $saveMode. " +
         "Accepted save modes are 'overwrite', 'append', 'ignore', 'error', 'errorifexists'.")
-    })
+    }
   }
 
   /**
@@ -269,10 +270,8 @@ final class DataFrameWriter[T] private[sql](ds: Dataset[T]) {
 
       import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Implicits._
       provider.getTable(dsOptions) match {
-        // TODO: for backward compatibility reasons, the builtin file source needs to support all
-        // the save modes, which violates the semantic of `TableProvider`. Here we special-case
-        // file source and pass the save mode to file source directly. This hack can be removed
-        // after we figure out a general interface for path-based data sources.
+        // TODO (SPARK-27815): To not break existing tests, here we treat file source as a special
+        // case, and pass the save mode to file source directly. This hack should be removed.
         case table: FileTable =>
           val write = table.newWriteBuilder(dsOptions).asInstanceOf[FileWriteBuilder]
             .mode(modeForDSV1) // should not change default mode for file source.
@@ -300,9 +299,9 @@ final class DataFrameWriter[T] private[sql](ds: Dataset[T]) {
                 OverwriteByExpression.byName(relation, df.logicalPlan, Literal(true))
               }
 
-            case _ =>
+            case other =>
               throw new AnalysisException(s"TableProvider implementation $source cannot be " +
-                "written with ErrorIfExists or Ignore modes, please use Append or Overwrite " +
+                s"written with $other mode, please use Append or Overwrite " +
                 "modes instead.")
           }
 

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

@@ -46,6 +46,7 @@ private[noop] object NoopTable extends Table with SupportsWrite {
     Set(
       TableCapability.BATCH_WRITE,
       TableCapability.STREAMING_WRITE,
+      TableCapability.TRUNCATE,
       TableCapability.ACCEPT_ANY_SCHEMA).asJava
   }
 }

sql/core/src/test/scala/org/apache/spark/sql/test/DataFrameReaderWriterSuite.scala

@@ -38,11 +38,15 @@ import org.apache.spark.internal.io.HadoopMapReduceCommitProtocol
 import org.apache.spark.scheduler.{SparkListener, SparkListenerJobStart}
 import org.apache.spark.sql._
 import org.apache.spark.sql.catalyst.TableIdentifier
+import org.apache.spark.sql.catalyst.plans.logical.{AppendData, LogicalPlan, OverwriteByExpression}
+import org.apache.spark.sql.execution.QueryExecution
 import org.apache.spark.sql.execution.datasources.DataSourceUtils
+import org.apache.spark.sql.execution.datasources.noop.NoopDataSource
 import org.apache.spark.sql.execution.datasources.parquet.SpecificParquetRecordReaderBase
 import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.sources._
 import org.apache.spark.sql.types._
+import org.apache.spark.sql.util.QueryExecutionListener
 import org.apache.spark.util.Utils
 
 
@@ -239,15 +243,75 @@ class DataFrameReaderWriterSuite extends QueryTest with SharedSQLContext with Be
   }
 
   test("save mode") {
-    val df = spark.read
+    spark.range(10).write
       .format("org.apache.spark.sql.test")
-      .load()
+      .mode(SaveMode.ErrorIfExists)
+      .save()
+    assert(LastOptions.saveMode === SaveMode.ErrorIfExists)
 
-    df.write
+    spark.range(10).write
+      .format("org.apache.spark.sql.test")
+      .mode(SaveMode.Append)
+      .save()
+    assert(LastOptions.saveMode === SaveMode.Append)
+
+    // By default the save mode is `ErrorIfExists` for data source v1.
+    spark.range(10).write
       .format("org.apache.spark.sql.test")
-      .mode(SaveMode.ErrorIfExists)
       .save()
     assert(LastOptions.saveMode === SaveMode.ErrorIfExists)
+
+    spark.range(10).write
+      .format("org.apache.spark.sql.test")
+      .mode("default")
+      .save()
+    assert(LastOptions.saveMode === SaveMode.ErrorIfExists)
+  }
+
+  test("save mode for data source v2") {
+    var plan: LogicalPlan = null
+    val listener = new QueryExecutionListener {
+      override def onSuccess(funcName: String, qe: QueryExecution, durationNs: Long): Unit = {
+        plan = qe.analyzed
+
+      }
+      override def onFailure(funcName: String, qe: QueryExecution, exception: Exception): Unit = {}
+    }
+
+    spark.listenerManager.register(listener)
+    try {
+      // append mode creates `AppendData`
+      spark.range(10).write
+        .format(classOf[NoopDataSource].getName)
+        .mode(SaveMode.Append)
+        .save()
+      sparkContext.listenerBus.waitUntilEmpty(1000)
+      assert(plan.isInstanceOf[AppendData])
+
+      // overwrite mode creates `OverwriteByExpression`
+      spark.range(10).write
+        .format(classOf[NoopDataSource].getName)
+        .mode(SaveMode.Overwrite)
+        .save()
+      sparkContext.listenerBus.waitUntilEmpty(1000)
+      assert(plan.isInstanceOf[OverwriteByExpression])
+
+      // By default the save mode is `ErrorIfExists` for data source v2.
+      spark.range(10).write
+        .format(classOf[NoopDataSource].getName)
+        .save()
+      sparkContext.listenerBus.waitUntilEmpty(1000)
+      assert(plan.isInstanceOf[AppendData])
+
+      spark.range(10).write
+        .format(classOf[NoopDataSource].getName)
+        .mode("default")
+        .save()
+      sparkContext.listenerBus.waitUntilEmpty(1000)
+      assert(plan.isInstanceOf[AppendData])
+    } finally {
+      spark.listenerManager.unregister(listener)
+    }
   }
 
   test("test path option in load") {