[Spark] Add support for dropping Deletion Vectors (#4009)

<!--
Thanks for sending a pull request!  Here are some tips for you:
1. If this is your first time, please read our contributor guidelines:
https://github.com/delta-io/delta/blob/master/CONTRIBUTING.md
2. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP]
Your PR title ...'.
  3. Be sure to keep the PR description updated to reflect all changes.
  4. Please write your PR title to summarize what this PR proposes.
5. If possible, provide a concise example to reproduce the issue for a
faster review.
6. If applicable, include the corresponding issue number in the PR title
and link it in the body.
-->

#### Which Delta project/connector is this regarding?
<!--
Please add the component selected below to the beginning of the pull
request title
For example: [Spark] Title of my pull request
-->

- [x] Spark
- [ ] Standalone
- [ ] Flink
- [ ] Kernel
- [ ] Other (fill in here)

## Description

<!--
- Describe what this PR changes.
- Describe why we need the change.
 
If this PR resolves an issue be sure to include "Resolves #XXX" to
correctly link and close the issue upon merge.
-->
This PR adds support for dropping deletion vectors.

## How was this patch tested?

<!--
If tests were added, say they were added here. Please make sure to test
the changes thoroughly including negative and positive cases if
possible.
If the changes were tested in any way other than unit tests, please
clarify how you tested step by step (ideally copy and paste-able, so
that other reviewers can test and check, and descendants can verify in
the future).
If the changes were not tested, please explain why.
-->
Added tests in `DeltaProtocolVersionSuite`.

## Does this PR introduce _any_ user-facing changes?

<!--
If yes, please clarify the previous behavior and the change this PR
proposes - provide the console output, description and/or an example to
show the behavior difference if possible.
If possible, please also clarify if this is a user-facing change
compared to the released Delta Lake versions or within the unreleased
branches such as master.
If no, write 'No'.
-->
Yes. See description.

Author: andreaschat-db

spark/src/main/scala/org/apache/spark/sql/delta/PreDowngradeTableFeatureCommand.scala

@@ -129,6 +129,59 @@ case class TestLegacyReaderWriterFeaturePreDowngradeCommand(table: DeltaTableV2)
   }
 }
 
+class DeletionVectorsRemovalMetrics(
+    val numDeletionVectorsToRemove: Long,
+    val numDeletionVectorRowsToRemove: Long,
+    var downgradeTimeMs: Long = 0L)
+
+case class DeletionVectorsPreDowngradeCommand(table: DeltaTableV2)
+  extends PreDowngradeTableFeatureCommand
+  with DeltaLogging {
+
+  /**
+   * We first remove the table feature property to prevent any transactions from committing
+   * new DVs. This will cause any concurrent transactions tox fail. Then, we run PURGE
+   * to remove existing DVs from the latest snapshot.
+   * Note, during the protocol downgrade phase we validate whether all invariants still hold.
+   * This should detect if any concurrent txns enabled the feature and/or added DVs again.
+   *
+   * @return Returns true if it removed DV metadata property and/or DVs. False otherwise.
+   */
+  override def removeFeatureTracesIfNeeded(): Boolean = {
+    // Latest snapshot looks clean. No action is required. We may proceed
+    // to the protocol downgrade phase.
+    if (DeletionVectorsTableFeature.validateRemoval(table.initialSnapshot)) return false
+
+    val startTimeNs = System.nanoTime()
+    val properties = Seq(DeltaConfigs.ENABLE_DELETION_VECTORS_CREATION.key)
+    AlterTableUnsetPropertiesDeltaCommand(
+      table, properties, ifExists = true, fromDropFeatureCommand = true).run(table.spark)
+
+    val snapshot = table.update()
+    val metrics = new DeletionVectorsRemovalMetrics(
+      numDeletionVectorsToRemove = snapshot.numDeletionVectorsOpt.getOrElse(0L),
+      numDeletionVectorRowsToRemove = snapshot.numDeletedRecordsOpt.getOrElse(0L))
+
+    // Wrap `table` in a ResolvedTable that can be passed to DeltaReorgTableCommand. The catalog &
+    // table ID won't be used by DeltaReorgTableCommand.
+    import org.apache.spark.sql.connector.catalog.CatalogV2Implicits._
+    val catalog = table.spark.sessionState.catalogManager.currentCatalog.asTableCatalog
+    val tableId = Seq(table.name()).asIdentifier
+
+    DeltaReorgTableCommand(target = ResolvedTable.create(catalog, tableId, table))(Nil)
+      .run(table.spark)
+
+    metrics.downgradeTimeMs =
+      TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startTimeNs)
+
+    recordDeltaEvent(
+      table.deltaLog,
+      opType = "delta.deletionVectorsFeatureRemovalMetrics",
+      data = metrics)
+    true
+  }
+}
+
 case class V2CheckpointPreDowngradeCommand(table: DeltaTableV2)
   extends PreDowngradeTableFeatureCommand
   with DeltaLogging {

spark/src/main/scala/org/apache/spark/sql/delta/TableFeature.scala

@@ -20,6 +20,7 @@ import java.util.Locale
 
 import org.apache.spark.sql.delta.actions._
 import org.apache.spark.sql.delta.catalog.DeltaTableV2
+import org.apache.spark.sql.delta.commands.DeletionVectorUtils
 import org.apache.spark.sql.delta.constraints.{Constraints, Invariants}
 import org.apache.spark.sql.delta.coordinatedcommits.CoordinatedCommitsUtils
 import org.apache.spark.sql.delta.redirect.{RedirectReaderWriter, RedirectWriterOnly}
@@ -641,6 +642,7 @@ object VariantTypeTableFeature extends ReaderWriterFeature(name = "variantType")
 
 object DeletionVectorsTableFeature
   extends ReaderWriterFeature(name = "deletionVectors")
+  with RemovableFeature
   with FeatureAutomaticallyEnabledByMetadata {
   override def automaticallyUpdateProtocolOfExistingTables: Boolean = true
 
@@ -650,6 +652,27 @@ object DeletionVectorsTableFeature
       spark: SparkSession): Boolean = {
     DeltaConfigs.ENABLE_DELETION_VECTORS_CREATION.fromMetaData(metadata)
   }
+
+  override def validateRemoval(snapshot: Snapshot): Boolean = {
+    val dvsWritable = DeletionVectorUtils.deletionVectorsWritable(snapshot)
+    val dvsExist = snapshot.numDeletionVectorsOpt.getOrElse(0L) > 0
+
+    !(dvsWritable || dvsExist)
+  }
+
+  override def actionUsesFeature(action: Action): Boolean = {
+    action match {
+      case m: Metadata => DeltaConfigs.ENABLE_DELETION_VECTORS_CREATION.fromMetaData(m)
+      case a: AddFile => a.deletionVector != null
+      case r: RemoveFile => r.deletionVector != null
+      // In general, CDC actions do not contain DVs. We added this for safety.
+      case cdc: AddCDCFile => cdc.deletionVector != null
+      case _ => false
+    }
+  }
+
+  override def preDowngradeCommand(table: DeltaTableV2): PreDowngradeTableFeatureCommand =
+    DeletionVectorsPreDowngradeCommand(table)
 }
 
 object RowTrackingFeature extends WriterFeature(name = "rowTracking")

spark/src/test/scala/org/apache/spark/sql/delta/DeletionVectorsTestUtils.scala

@@ -21,6 +21,8 @@ import java.util.UUID
 
 import org.apache.spark.sql.delta.DeltaOperations.Truncate
 import org.apache.spark.sql.delta.actions.{Action, AddFile, DeletionVectorDescriptor, RemoveFile}
+import org.apache.spark.sql.delta.catalog.DeltaTableV2
+import org.apache.spark.sql.delta.commands.AlterTableDropFeatureDeltaCommand
 import org.apache.spark.sql.delta.deletionvectors.{RoaringBitmapArray, RoaringBitmapArrayFormat}
 import org.apache.spark.sql.delta.sources.DeltaSQLConf
 import org.apache.spark.sql.delta.storage.dv.DeletionVectorStore
@@ -72,6 +74,15 @@ trait DeletionVectorsTestUtils extends QueryTest with SharedSparkSession with De
     }
   }
 
+  def dropDVTableFeature(
+      spark: SparkSession,
+      log: DeltaLog,
+      truncateHistory: Boolean): Unit =
+    AlterTableDropFeatureDeltaCommand(
+      DeltaTableV2(spark, log.dataPath),
+      DeletionVectorsTableFeature.name,
+      truncateHistory = truncateHistory).run(spark)
+
   /** Helper to run 'fn' with a temporary Delta table. */
   def withTempDeltaTable(
       dataDF: DataFrame,

spark/src/test/scala/org/apache/spark/sql/delta/DeltaProtocolVersionSuite.scala

@@ -30,6 +30,7 @@ import org.apache.spark.sql.delta.actions._
 import org.apache.spark.sql.delta.actions.TableFeatureProtocolUtils._
 import org.apache.spark.sql.delta.catalog.DeltaTableV2
 import org.apache.spark.sql.delta.commands.{AlterTableDropFeatureDeltaCommand, AlterTableSetPropertiesDeltaCommand, AlterTableUnsetPropertiesDeltaCommand}
+import org.apache.spark.sql.delta.commands.DeletionVectorUtils
 import org.apache.spark.sql.delta.coordinatedcommits._
 import org.apache.spark.sql.delta.sources.DeltaSQLConf
 import org.apache.spark.sql.delta.test.DeltaSQLCommandTest
@@ -54,7 +55,8 @@ import org.apache.spark.util.ManualClock
 
 trait DeltaProtocolVersionSuiteBase extends QueryTest
   with SharedSparkSession
-  with DeltaSQLCommandTest {
+  with DeltaSQLCommandTest
+  with DeletionVectorsTestUtils {
 
   // `.schema` generates NOT NULL columns which requires writer protocol 2. We convert all to
   // NULLable to avoid silent writer protocol version bump.
@@ -3408,6 +3410,96 @@ trait DeltaProtocolVersionSuiteBase extends QueryTest
       expectedDowngradedProtocol = protocolWithReaderFeature(TestRemovableReaderWriterFeature))
   }
 
+  for {
+    truncateHistory <- BOOLEAN_DOMAIN
+    enableCDF <- if (truncateHistory) Seq(false) else BOOLEAN_DOMAIN
+  } test(s"Remove Deletion Vectors feature " +
+      s"truncateHistory: $truncateHistory, enableCDF: $enableCDF") {
+    val targetDF = spark.range(start = 0, end = 100, step = 1, numPartitions = 2)
+    withSQLConf(
+        DeltaConfigs.ENABLE_DELETION_VECTORS_CREATION.defaultTablePropertyKey -> "true",
+        DeltaConfigs.CHANGE_DATA_FEED.defaultTablePropertyKey -> enableCDF.toString) {
+      withTempPath { dir =>
+        val clock = new ManualClock(System.currentTimeMillis())
+        val targetLog = DeltaLog.forTable(spark, dir, clock)
+        val defaultRetentionPeriod =
+          DeltaConfigs.LOG_RETENTION.fromMetaData(targetLog.update().metadata).toString
+
+        targetDF.write.format("delta").save(dir.toString)
+
+        val targetTable = io.delta.tables.DeltaTable.forPath(dir.toString)
+
+        // Add some DVs.
+        targetTable.delete("id >= 90")
+
+        // Assert that DVs exist.
+        val preDowngradeSnapshot = targetLog.update()
+        assert(DeletionVectorUtils.deletionVectorsWritable(preDowngradeSnapshot))
+        assert(preDowngradeSnapshot.numDeletionVectorsOpt === Some(1L))
+
+        // Attempting to drop Deletion Vectors feature will prohibit adding new DVs and remove
+        // all DVs from the latest snapshot, but ultimately fail, because history will still
+        // contain traces of the feature. For this reason, we have to wait for the retention period
+        // to be over before we can downgrade the protocol.
+        val e1 = intercept[DeltaTableFeatureException] {
+          dropDVTableFeature(spark, targetLog, truncateHistory = false)
+        }
+        checkError(
+          e1,
+          "DELTA_FEATURE_DROP_WAIT_FOR_RETENTION_PERIOD",
+          parameters = Map(
+            "feature" -> DeletionVectorsTableFeature.name,
+            "logRetentionPeriodKey" -> "delta.logRetentionDuration",
+            "logRetentionPeriod" -> defaultRetentionPeriod,
+            "truncateHistoryLogRetentionPeriod" -> truncateHistoryDefaultLogRetention.toString))
+
+        val postCleanupSnapshot = targetLog.update()
+        assert(!DeletionVectorUtils.deletionVectorsWritable(postCleanupSnapshot))
+        assert(postCleanupSnapshot.numDeletionVectorsOpt.getOrElse(0L) === 0)
+        assert(postCleanupSnapshot.numDeletedRecordsOpt.getOrElse(0L) === 0)
+
+        spark.range(100, 120).write.format("delta").mode("append").save(dir.getCanonicalPath)
+        spark.range(120, 140).write.format("delta").mode("append").save(dir.getCanonicalPath)
+
+        // Table still contains historical data with DVs. Attempt should fail.
+        val e2 = intercept[DeltaTableFeatureException] {
+          dropDVTableFeature(spark, targetLog, truncateHistory = false)
+        }
+        checkError(
+          e2,
+          "DELTA_FEATURE_DROP_HISTORICAL_VERSIONS_EXIST",
+          parameters = Map(
+            "feature" -> DeletionVectorsTableFeature.name,
+            "logRetentionPeriodKey" -> "delta.logRetentionDuration",
+            "logRetentionPeriod" -> defaultRetentionPeriod,
+            "truncateHistoryLogRetentionPeriod" -> truncateHistoryDefaultLogRetention.toString))
+
+        // Pretend retention period has passed.
+        val clockAdvanceMillis = if (truncateHistory) {
+          DeltaConfigs.getMilliSeconds(truncateHistoryDefaultLogRetention) +
+            TimeUnit.HOURS.toMillis(24)
+        } else {
+          targetLog.deltaRetentionMillis(targetLog.update().metadata) + TimeUnit.DAYS.toMillis(3)
+        }
+        clock.advance(clockAdvanceMillis)
+
+        // Cleanup logs.
+        targetLog.cleanUpExpiredLogs(targetLog.update())
+
+        // History is now clean. We should be able to remove the feature.
+        dropDVTableFeature(spark, targetLog, truncateHistory)
+
+        val postDowngradeSnapshot = targetLog.update()
+        val protocol = postDowngradeSnapshot.protocol
+        assert(!DeletionVectorUtils.deletionVectorsWritable(postDowngradeSnapshot))
+        assert(postDowngradeSnapshot.numDeletionVectorsOpt.getOrElse(0L) === 0)
+        assert(postDowngradeSnapshot.numDeletedRecordsOpt.getOrElse(0L) === 0)
+        assert(!protocol.readerFeatureNames.contains(DeletionVectorsTableFeature.name))
+      }
+    }
+  }
+
+
   test("Can drop reader+writer feature when there is nothing to clean") {
     withTempPath { dir =>
       val clock = new ManualClock(System.currentTimeMillis())