[SPARK-56316][SQL] Support static partition overwrite for V2 file tables

Implement SupportsOverwriteV2 for V2 file tables to support static
partition overwrite (INSERT OVERWRITE TABLE t PARTITION(p=1) SELECT ...).

Key changes:
- FileTable: replace SupportsTruncate with SupportsOverwriteV2 on
  WriteBuilder, implement overwrite(predicates)
- FileWrite: extend toBatch() to delete only the matching partition
  directory, ordered by partitionSchema
- FileTable.CAPABILITIES: add OVERWRITE_BY_FILTER
- All 6 format Write/Table classes: plumb overwritePredicates parameter

This is a prerequisite for SPARK-56304 (ifPartitionNotExists).

Author: LuciferYang

connector/avro/src/main/scala/org/apache/spark/sql/v2/avro/AvroTable.scala

@@ -44,9 +44,9 @@ case class AvroTable(
 
   override def newWriteBuilder(info: LogicalWriteInfo): WriteBuilder = {
     createFileWriteBuilder(info) {
-      (mergedInfo, partSchema, bSpec, customLocs, dynamicOverwrite, truncate) =>
+      (mergedInfo, partSchema, bSpec, customLocs, dynamicOverwrite, truncate, overPreds) =>
       AvroWrite(paths, formatName, supportsDataType, mergedInfo, partSchema, bSpec,
-        customLocs, dynamicOverwrite, truncate)
+        overPreds, customLocs, dynamicOverwrite, truncate)
     }
   }
 

connector/avro/src/main/scala/org/apache/spark/sql/v2/avro/AvroWrite.scala

@@ -20,6 +20,7 @@ import org.apache.hadoop.mapreduce.Job
 
 import org.apache.spark.sql.avro.AvroUtils
 import org.apache.spark.sql.catalyst.catalog.BucketSpec
+import org.apache.spark.sql.connector.expressions.filter.Predicate
 import org.apache.spark.sql.connector.write.LogicalWriteInfo
 import org.apache.spark.sql.execution.datasources.OutputWriterFactory
 import org.apache.spark.sql.execution.datasources.v2.FileWrite
@@ -33,6 +34,7 @@ case class AvroWrite(
     info: LogicalWriteInfo,
     partitionSchema: StructType,
     override val bucketSpec: Option[BucketSpec] = None,
+    override val overwritePredicates: Option[Array[Predicate]] = None,
     override val customPartitionLocations: Map[Map[String, String], String] = Map.empty,
     override val dynamicPartitionOverwrite: Boolean,
     override val isTruncate: Boolean) extends FileWrite {

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

@@ -31,9 +31,10 @@ import org.apache.spark.sql.connector.catalog.{CatalogV2Util, Column,
   Table, TableCapability}
 import org.apache.spark.sql.connector.catalog.TableCapability._
 import org.apache.spark.sql.connector.expressions.Transform
+import org.apache.spark.sql.connector.expressions.filter.{AlwaysTrue, Predicate}
 import org.apache.spark.sql.connector.write.{LogicalWriteInfo,
   LogicalWriteInfoImpl, SupportsDynamicOverwrite,
-  SupportsTruncate, Write, WriteBuilder}
+  SupportsOverwriteV2, Write, WriteBuilder}
 import org.apache.spark.sql.errors.QueryCompilationErrors
 import org.apache.spark.sql.execution.datasources._
 import org.apache.spark.sql.execution.streaming.runtime.MetadataLogFileIndex
@@ -308,19 +309,23 @@ abstract class FileTable(
       buildWrite: (LogicalWriteInfo, StructType,
         Option[BucketSpec],
         Map[Map[String, String], String],
-        Boolean, Boolean) => Write
+        Boolean, Boolean,
+        Option[Array[Predicate]]) => Write
   ): WriteBuilder = {
-    new WriteBuilder with SupportsDynamicOverwrite with SupportsTruncate {
+    new WriteBuilder with SupportsDynamicOverwrite
+        with SupportsOverwriteV2 {
       private var isDynamicOverwrite = false
-      private var isTruncate = false
+      private var overwritePredicates
+        : Option[Array[Predicate]] = None
 
       override def overwriteDynamicPartitions(): WriteBuilder = {
         isDynamicOverwrite = true
         this
       }
 
-      override def truncate(): WriteBuilder = {
-        isTruncate = true
+      override def overwrite(
+          predicates: Array[Predicate]): WriteBuilder = {
+        overwritePredicates = Some(predicates)
         this
       }
 
@@ -362,10 +367,13 @@ abstract class FileTable(
               .getOrElse(fromIndex)
           }
         val bSpec = catalogTable.flatMap(_.bucketSpec)
+        val isTruncate = overwritePredicates.exists(
+          _.exists(_.isInstanceOf[AlwaysTrue]))
         val customLocs = getCustomPartitionLocations(
           partSchema)
         buildWrite(merged, partSchema, bSpec,
-          customLocs, isDynamicOverwrite, isTruncate)
+          customLocs, isDynamicOverwrite, isTruncate,
+          overwritePredicates)
       }
     }
   }
@@ -577,7 +585,8 @@ abstract class FileTable(
 
 object FileTable {
   private val CAPABILITIES = util.EnumSet.of(
-    BATCH_READ, BATCH_WRITE, TRUNCATE, OVERWRITE_DYNAMIC)
+    BATCH_READ, BATCH_WRITE, TRUNCATE,
+    OVERWRITE_BY_FILTER, OVERWRITE_DYNAMIC)
 
   /** Option key for injecting stored row count from ANALYZE TABLE into FileScan. */
   val NUM_ROWS_KEY: String = "__numRows"

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

@@ -34,6 +34,7 @@ import org.apache.spark.sql.catalyst.util.{CaseInsensitiveMap, DateTimeUtils}
 import org.apache.spark.sql.connector.distributions.{Distribution, Distributions}
 import org.apache.spark.sql.connector.expressions.{Expressions, SortDirection}
 import org.apache.spark.sql.connector.expressions.{SortOrder => V2SortOrder}
+import org.apache.spark.sql.connector.expressions.filter.Predicate
 import org.apache.spark.sql.connector.write.{BatchWrite, LogicalWriteInfo, RequiresDistributionAndOrdering, Write}
 import org.apache.spark.sql.errors.QueryCompilationErrors
 import org.apache.spark.sql.execution.datasources.{BasicWriteJobStatsTracker, DataSource, OutputWriterFactory, V1WritesUtils, WriteJobDescription}
@@ -52,6 +53,7 @@ trait FileWrite extends Write
   def info: LogicalWriteInfo
   def partitionSchema: StructType
   def bucketSpec: Option[BucketSpec] = None
+  def overwritePredicates: Option[Array[Predicate]] = None
   def customPartitionLocations: Map[Map[String, String], String] = Map.empty
   def dynamicPartitionOverwrite: Boolean = false
   def isTruncate: Boolean = false
@@ -93,15 +95,31 @@ trait FileWrite extends Write
       fs.mkdirs(qualifiedPath)
     }
 
-    // For truncate (full overwrite), delete existing data before writing.
     if (isTruncate && fs.exists(qualifiedPath)) {
+      // Full overwrite: delete all non-hidden data
       fs.listStatus(qualifiedPath).foreach { status =>
-        // Preserve hidden files/dirs (e.g., _SUCCESS, .spark-staging-*)
         if (!status.getPath.getName.startsWith("_") &&
             !status.getPath.getName.startsWith(".")) {
           fs.delete(status.getPath, true)
         }
       }
+    } else if (overwritePredicates.exists(_.nonEmpty) &&
+        fs.exists(qualifiedPath)) {
+      // Static partition overwrite: delete only matching partition dir.
+      // Extract partition spec from predicates and order by
+      // partitionSchema to match the directory structure.
+      val specMap = overwritePredicates.get
+        .flatMap(FileWrite.predicateToPartitionSpec)
+        .toMap
+      if (specMap.nonEmpty) {
+        val partPath = partitionSchema.fieldNames
+          .flatMap(col => specMap.get(col).map(v => s"$col=$v"))
+          .mkString("/")
+        val targetPath = new Path(qualifiedPath, partPath)
+        if (fs.exists(targetPath)) {
+          fs.delete(targetPath, true)
+        }
+      }
     }
 
     val job = getJobInstance(hadoopConf, path)
@@ -219,3 +237,31 @@ trait FileWrite extends Write
   }
 }
 
+private[v2] object FileWrite {
+  /**
+   * Extract a (column, value) pair from a V2 equality
+   * predicate (e.g., `p <=> 1` => `("p", "1")`).
+   */
+  def predicateToPartitionSpec(
+      predicate: Predicate): Option[(String, String)] = {
+    if (predicate.name() == "=" || predicate.name() == "<=>") {
+      val children = predicate.children()
+      if (children.length == 2) {
+        val name = children(0) match {
+          case ref: org.apache.spark.sql.connector
+              .expressions.NamedReference =>
+            Some(ref.fieldNames().head)
+          case _ => None
+        }
+        val value = children(1) match {
+          case lit: org.apache.spark.sql.connector
+              .expressions.Literal[_] =>
+            Some(lit.value().toString)
+          case _ => None
+        }
+        for (n <- name; v <- value) yield (n, v)
+      } else None
+    } else None
+  }
+}
+

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

@@ -52,9 +52,9 @@ case class CSVTable(
 
   override def newWriteBuilder(info: LogicalWriteInfo): WriteBuilder = {
     createFileWriteBuilder(info) {
-      (mergedInfo, partSchema, bSpec, customLocs, dynamicOverwrite, truncate) =>
+      (mergedInfo, partSchema, bSpec, customLocs, dynamicOverwrite, truncate, overPreds) =>
       CSVWrite(paths, formatName, supportsWriteDataType, mergedInfo, partSchema, bSpec,
-        customLocs, dynamicOverwrite, truncate)
+        overPreds, customLocs, dynamicOverwrite, truncate)
     }
   }
 

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

@@ -21,6 +21,7 @@ import org.apache.hadoop.mapreduce.{Job, TaskAttemptContext}
 import org.apache.spark.sql.catalyst.catalog.BucketSpec
 import org.apache.spark.sql.catalyst.csv.CSVOptions
 import org.apache.spark.sql.catalyst.util.CompressionCodecs
+import org.apache.spark.sql.connector.expressions.filter.Predicate
 import org.apache.spark.sql.connector.write.LogicalWriteInfo
 import org.apache.spark.sql.execution.datasources.{CodecStreams, OutputWriter, OutputWriterFactory}
 import org.apache.spark.sql.execution.datasources.csv.CsvOutputWriter
@@ -35,6 +36,7 @@ case class CSVWrite(
     info: LogicalWriteInfo,
     partitionSchema: StructType,
     override val bucketSpec: Option[BucketSpec] = None,
+    override val overwritePredicates: Option[Array[Predicate]] = None,
     override val customPartitionLocations: Map[Map[String, String], String] = Map.empty,
     override val dynamicPartitionOverwrite: Boolean,
     override val isTruncate: Boolean) extends FileWrite {

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

@@ -51,9 +51,9 @@ case class JsonTable(
 
   override def newWriteBuilder(info: LogicalWriteInfo): WriteBuilder = {
     createFileWriteBuilder(info) {
-      (mergedInfo, partSchema, bSpec, customLocs, dynamicOverwrite, truncate) =>
+      (mergedInfo, partSchema, bSpec, customLocs, dynamicOverwrite, truncate, overPreds) =>
       JsonWrite(paths, formatName, supportsDataType, mergedInfo, partSchema, bSpec,
-        customLocs, dynamicOverwrite, truncate)
+        overPreds, customLocs, dynamicOverwrite, truncate)
     }
   }
 

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

@@ -21,6 +21,7 @@ import org.apache.hadoop.mapreduce.{Job, TaskAttemptContext}
 import org.apache.spark.sql.catalyst.catalog.BucketSpec
 import org.apache.spark.sql.catalyst.json.JSONOptions
 import org.apache.spark.sql.catalyst.util.CompressionCodecs
+import org.apache.spark.sql.connector.expressions.filter.Predicate
 import org.apache.spark.sql.connector.write.LogicalWriteInfo
 import org.apache.spark.sql.execution.datasources.{CodecStreams, OutputWriter, OutputWriterFactory}
 import org.apache.spark.sql.execution.datasources.json.JsonOutputWriter
@@ -35,6 +36,7 @@ case class JsonWrite(
     info: LogicalWriteInfo,
     partitionSchema: StructType,
     override val bucketSpec: Option[BucketSpec] = None,
+    override val overwritePredicates: Option[Array[Predicate]] = None,
     override val customPartitionLocations: Map[Map[String, String], String] = Map.empty,
     override val dynamicPartitionOverwrite: Boolean,
     override val isTruncate: Boolean) extends FileWrite {

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

@@ -45,9 +45,9 @@ case class OrcTable(
 
   override def newWriteBuilder(info: LogicalWriteInfo): WriteBuilder = {
     createFileWriteBuilder(info) {
-      (mergedInfo, partSchema, bSpec, customLocs, dynamicOverwrite, truncate) =>
+      (mergedInfo, partSchema, bSpec, customLocs, dynamicOverwrite, truncate, overPreds) =>
       OrcWrite(paths, formatName, supportsDataType, mergedInfo, partSchema, bSpec,
-        customLocs, dynamicOverwrite, truncate)
+        overPreds, customLocs, dynamicOverwrite, truncate)
     }
   }
 

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

@@ -22,6 +22,7 @@ import org.apache.orc.OrcConf.{COMPRESS, MAPRED_OUTPUT_SCHEMA}
 import org.apache.orc.mapred.OrcStruct
 
 import org.apache.spark.sql.catalyst.catalog.BucketSpec
+import org.apache.spark.sql.connector.expressions.filter.Predicate
 import org.apache.spark.sql.connector.write.LogicalWriteInfo
 import org.apache.spark.sql.execution.datasources.{OutputWriter, OutputWriterFactory}
 import org.apache.spark.sql.execution.datasources.orc.{OrcOptions, OrcOutputWriter, OrcUtils}
@@ -36,6 +37,7 @@ case class OrcWrite(
     info: LogicalWriteInfo,
     partitionSchema: StructType,
     override val bucketSpec: Option[BucketSpec] = None,
+    override val overwritePredicates: Option[Array[Predicate]] = None,
     override val customPartitionLocations: Map[Map[String, String], String] = Map.empty,
     override val dynamicPartitionOverwrite: Boolean,
     override val isTruncate: Boolean) extends FileWrite {