Remove the check that disables writes to Delta tables with deletion vectors

core/src/main/scala/org/apache/spark/sql/delta/OptimisticTransaction.scala

@@ -139,7 +139,6 @@ class OptimisticTransaction
     (implicit override val clock: Clock)
   extends OptimisticTransactionImpl
   with DeltaLogging {
-  DeletionVectorUtils.assertDeletionVectorsNotReadable(spark, snapshot.metadata, snapshot.protocol)
 
   /** Creates a new OptimisticTransaction.
    *
@@ -552,7 +551,6 @@ trait OptimisticTransactionImpl extends TransactionalWrite
     newMetadataTmp = RowId.verifyAndUpdateMetadata(
       spark, protocol, snapshot.metadata, newMetadataTmp, isCreatingNewTable)
 
-    DeletionVectorUtils.assertDeletionVectorsNotEnabled(spark, newMetadataTmp, protocol)
     assertMetadata(newMetadataTmp)
     logInfo(s"Updated metadata from ${newMetadata.getOrElse("-")} to $newMetadataTmp")
     newMetadata = Some(newMetadataTmp)

core/src/main/scala/org/apache/spark/sql/delta/commands/DeleteCommand.scala

@@ -22,7 +22,6 @@ import org.apache.spark.sql.delta.commands.DeleteCommand.{rewritingFilesMsg, FIN
 import org.apache.spark.sql.delta.commands.MergeIntoCommand.totalBytesAndDistinctPartitionValues
 import org.apache.spark.sql.delta.files.TahoeBatchFileIndex
 import org.apache.spark.sql.delta.sources.DeltaSQLConf
-import org.apache.spark.sql.delta.util.Utils
 import com.fasterxml.jackson.databind.annotation.JsonDeserialize
 
 import org.apache.spark.SparkContext
@@ -440,10 +439,8 @@ case class DeleteCommand(
 
   def shouldWritePersistentDeletionVectors(
       spark: SparkSession, txn: OptimisticTransaction): Boolean = {
-    // DELETE with DVs only enabled for tests.
-    Utils.isTesting &&
-      spark.conf.get(DeltaSQLConf.DELETE_USE_PERSISTENT_DELETION_VECTORS) &&
-        DeletionVectorUtils.deletionVectorsWritable(txn.snapshot)
+    spark.conf.get(DeltaSQLConf.DELETE_USE_PERSISTENT_DELETION_VECTORS) &&
+      DeletionVectorUtils.deletionVectorsWritable(txn.snapshot)
   }
 }
 

core/src/main/scala/org/apache/spark/sql/delta/commands/DeletionVectorUtils.scala

@@ -85,43 +85,6 @@ trait DeletionVectorUtils {
     protocol.isFeatureSupported(DeletionVectorsTableFeature) &&
       metadata.format.provider == "parquet" // DVs are only supported on parquet tables.
   }
-
-  /**
-   * Utility method that checks the table has no Deletion Vectors enabled. Deletion vectors
-   * are supported in read-only mode for now. Any updates to tables with deletion vectors
-   * feature are disabled until we add support.
-   */
-  def assertDeletionVectorsNotReadable(
-      spark: SparkSession, metadata: Metadata, protocol: Protocol): Unit = {
-    val disable =
-      Utils.isTesting && // We are in testing and enabled blocking updates on DV tables
-          spark.conf.get(DeltaSQLConf.DELTA_ENABLE_BLOCKING_UPDATES_ON_DV_TABLES)
-    if (!disable && deletionVectorsReadable(protocol, metadata)) {
-      throw new UnsupportedOperationException(
-        "Updates to tables with Deletion Vectors feature enabled are not supported in " +
-          "this version of Delta Lake.")
-    }
-  }
-
-  /**
-   * Utility method that checks the table metadata has no deletion vectors enabled. Deletion vectors
-   * are supported in read-only mode for now. Any updates to metadata to enable deletion vectors are
-   * blocked until we add support.
-   */
-  def assertDeletionVectorsNotEnabled(
-    spark: SparkSession, metadata: Metadata, protocol: Protocol): Unit = {
-    val disable =
-      Utils.isTesting && // We are in testing and enabled blocking updates on DV tables
-        spark.conf.get(DeltaSQLConf.DELTA_ENABLE_BLOCKING_UPDATES_ON_DV_TABLES)
-    if (!disable &&
-      (protocol.isFeatureSupported(DeletionVectorsTableFeature) ||
-        DeltaConfigs.ENABLE_DELETION_VECTORS_CREATION.fromMetaData(metadata)
-      )
-    ) {
-      throw new UnsupportedOperationException(
-        "Enabling Deletion Vectors on the table is not supported in this version of Delta Lake.")
-    }
-  }
 }
 
 // To access utilities from places where mixing in a trait is inconvenient.

core/src/main/scala/org/apache/spark/sql/delta/commands/VacuumCommand.scala

@@ -114,9 +114,6 @@ object VacuumCommand extends VacuumCommandImpl with Serializable {
       require(snapshot.version >= 0, "No state defined for this table. Is this really " +
         "a Delta table? Refusing to garbage collect.")
 
-      DeletionVectorUtils.assertDeletionVectorsNotReadable(
-        spark, snapshot.metadata, snapshot.protocol)
-
       val snapshotTombstoneRetentionMillis = DeltaLog.tombstoneRetentionMillis(snapshot.metadata)
       val retentionMillis = retentionHours.map(h => TimeUnit.HOURS.toMillis(math.round(h)))
       checkRetentionPeriodSafety(spark, retentionMillis, snapshotTombstoneRetentionMillis)

core/src/main/scala/org/apache/spark/sql/delta/sources/DeltaSQLConf.scala

@@ -1157,15 +1157,6 @@ trait DeltaSQLConfBase {
       .booleanConf
       .createWithDefault(true)
 
-  val DELTA_ENABLE_BLOCKING_UPDATES_ON_DV_TABLES =
-    buildConf("deletionVectors.updates.blocking.enabled")
-        .internal()
-        .doc(
-          """Enable blocking updates on tables with Deletion Vectors
-            |Only change this for testing!""".stripMargin)
-        .booleanConf
-        .createWithDefault(true)
-
   val DELTA_DUPLICATE_ACTION_CHECK_ENABLED =
     buildConf("duplicateActionCheck.enabled")
       .internal()

core/src/test/scala/org/apache/spark/sql/delta/deletionvectors/DeletionVectorsSuite.scala

@@ -603,8 +603,4 @@ object DeletionVectorsSuite {
   // Table at version 4: inserts rows with id = 1353, 1567
   val table3V4Added = Set(1353, 1567)
   val expectedTable3DataV4 = expectedTable3DataV3 ++ table3V4Added
-
-  // Table with DV table feature as supported but no DVs
-  val table4Path = "src/test/resources/delta/table-with-dv-feature-enabled"
-  val expectedTable4DataV0 = Seq(1L)
 }