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

Given that now we have support for writing into DV tables and table utility operations as as part of the delta-io#1485 and delta-io#1591, we should remove the check.

Closes delta-io#1736

Author: vkorukanti

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.
    *
@@ -563,7 +562,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()