SPARK-26666: Update for review comments.

Author: rdblue

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

@@ -978,7 +978,7 @@ class Analyzer(
       case a @ Aggregate(groupingExprs, aggExprs, appendColumns: AppendColumns) =>
         a.mapExpressions(resolveExpressionTopDown(_, appendColumns))
 
-      case o: OverwriteByExpression if !o.writeResolved =>
+      case o: OverwriteByExpression if !o.outputResolved =>
         // do not resolve expression attributes until the query attributes are resolved against the
         // table by ResolveOutputRelation. that rule will alias the attributes to the table's names.
         o
@@ -2251,7 +2251,7 @@ class Analyzer(
   object ResolveOutputRelation extends Rule[LogicalPlan] {
     override def apply(plan: LogicalPlan): LogicalPlan = plan.resolveOperators {
       case append @ AppendData(table, query, isByName)
-          if table.resolved && query.resolved && !append.writeResolved =>
+          if table.resolved && query.resolved && !append.outputResolved =>
         val projection = resolveOutputColumns(table.name, table.output, query, isByName)
 
         if (projection != query) {
@@ -2261,7 +2261,7 @@ class Analyzer(
         }
 
       case overwrite @ OverwriteByExpression(table, _, query, isByName)
-        if table.resolved && query.resolved && !overwrite.writeResolved =>
+          if table.resolved && query.resolved && !overwrite.outputResolved =>
         val projection = resolveOutputColumns(table.name, table.output, query, isByName)
 
         if (projection != query) {
@@ -2271,7 +2271,7 @@ class Analyzer(
         }
 
       case overwrite @ OverwritePartitionsDynamic(table, query, isByName)
-        if table.resolved && query.resolved && !overwrite.writeResolved =>
+          if table.resolved && query.resolved && !overwrite.outputResolved =>
         val projection = resolveOutputColumns(table.name, table.output, query, isByName)
 
         if (projection != query) {

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

@@ -373,9 +373,9 @@ trait V2WriteCommand extends Command {
 
   override def children: Seq[LogicalPlan] = Seq(query)
 
-  override lazy val resolved: Boolean = writeResolved
+  override lazy val resolved: Boolean = outputResolved
 
-  def writeResolved: Boolean = {
+  def outputResolved: Boolean = {
     table.resolved && query.resolved && query.output.size == table.output.size &&
         query.output.zip(table.output).forall {
           case (inAttr, outAttr) =>
@@ -413,7 +413,7 @@ case class OverwriteByExpression(
     deleteExpr: Expression,
     query: LogicalPlan,
     isByName: Boolean) extends V2WriteCommand {
-  override lazy val resolved: Boolean = writeResolved && deleteExpr.resolved
+  override lazy val resolved: Boolean = outputResolved && deleteExpr.resolved
 }
 
 object OverwriteByExpression {

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

@@ -29,6 +29,9 @@ public interface SupportsPushDownFilters extends ScanBuilder {
 
   /**
    * Pushes down filters, and returns filters that need to be evaluated after scanning.
+   * <p>
+   * Rows should be returned from the data source if and only if all of the filters match. That is,
+   * filters must be interpreted as ANDed together.
    */
   Filter[] pushFilters(Filter[] filters);
 

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

@@ -29,6 +29,9 @@
 public interface SupportsOverwrite extends WriteBuilder, SupportsTruncate {
   /**
    * Configures a write to replace data matching the filters with data committed in the write.
+   * <p>
+   * Rows must be deleted from the data source if and only if all of the filters match. That is,
+   * filters must be interpreted as ANDed together.
    *
    * @param filters filters used to match data to overwrite
    * @return this write builder for method chaining

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

@@ -46,6 +46,11 @@ case class WriteToDataSourceV2(batchWrite: BatchWrite, query: LogicalPlan)
   override def output: Seq[Attribute] = Nil
 }
 
+/**
+ * Physical plan node for append into a v2 table.
+ *
+ * Rows in the output data set are appended.
+ */
 case class AppendDataExec(
     table: SupportsBatchWrite,
     writeOptions: DataSourceOptions,
@@ -63,9 +68,19 @@ case class AppendDataExec(
   }
 }
 
+/**
+ * Physical plan node for overwrite into a v2 table.
+ *
+ * Overwrites data in a table matched by a set of filters. Rows matching all of the filters will be
+ * deleted and rows in the output data set are appended.
+ *
+ * This plan is used to implement SaveMode.Overwrite. The behavior of SaveMode.Overwrite is to
+ * truncate the table -- delete all rows -- and append the output data set. This uses the filter
+ * AlwaysTrue to delete all rows.
+ */
 case class OverwriteByExpressionExec(
     table: SupportsBatchWrite,
-    filters: Array[Filter],
+    deleteWhere: Array[Filter],
     writeOptions: DataSourceOptions,
     query: SparkPlan) extends V2TableWriteExec with BatchWriteHelper {
 
@@ -75,15 +90,15 @@ case class OverwriteByExpressionExec(
 
   override protected def doExecute(): RDD[InternalRow] = {
     val batchWrite = newWriteBuilder() match {
-      case builder: SupportsTruncate if isTruncate(filters) =>
+      case builder: SupportsTruncate if isTruncate(deleteWhere) =>
         builder.truncate().buildForBatch()
 
-      case builder: SupportsOverwrite =>
-        builder.overwrite(filters).buildForBatch()
-
-      case builder: SupportsSaveMode =>
+      case builder: SupportsSaveMode if isTruncate(deleteWhere) =>
         builder.mode(SaveMode.Overwrite).buildForBatch()
 
+      case builder: SupportsOverwrite =>
+        builder.overwrite(deleteWhere).buildForBatch()
+
       case _ =>
         throw new SparkException(s"Table does not support dynamic partition overwrite: $table")
     }
@@ -92,6 +107,15 @@ case class OverwriteByExpressionExec(
   }
 }
 
+/**
+ * Physical plan node for dynamic partition overwrite into a v2 table.
+ *
+ * Dynamic partition overwrite is the behavior of Hive INSERT OVERWRITE ... PARTITION queries, and
+ * Spark INSERT OVERWRITE queries when spark.sql.sources.partitionOverwriteMode=dynamic. Each
+ * partition in the output data set replaces the corresponding existing partition in the table or
+ * creates a new partition. Existing partitions for which there is no data in the output data set
+ * are not modified.
+ */
 case class OverwritePartitionsDynamicExec(
     table: SupportsBatchWrite,
     writeOptions: DataSourceOptions,

sql/core/src/main/scala/org/apache/spark/sql/sources/filters.scala

@@ -17,7 +17,7 @@
 
 package org.apache.spark.sql.sources
 
-import org.apache.spark.annotation.Stable
+import org.apache.spark.annotation.{Evolving, Stable}
 
 ////////////////////////////////////////////////////////////////////////////////////////////////////
 // This file defines all the filters that we can push down to the data sources.
@@ -222,19 +222,23 @@ case class StringContains(attribute: String, value: String) extends Filter {
 /**
  * A filter that always evaluates to `true`.
  */
+@Evolving
 case class AlwaysTrue() extends Filter {
   override def references: Array[String] = Array.empty
 }
 
+@Evolving
 object AlwaysTrue extends AlwaysTrue {
 }
 
 /**
  * A filter that always evaluates to `false`.
  */
+@Evolving
 case class AlwaysFalse() extends Filter {
   override def references: Array[String] = Array.empty
 }
 
+@Evolving
 object AlwaysFalse extends AlwaysFalse {
 }