changes

spark/src/main/resources/error/delta-error-classes.json

@@ -897,6 +897,13 @@
     ],
     "sqlState" : "0AKDE"
   },
+  "DELTA_FEATURE_DROP_DEPENDENT_FEATURE" : {
+    "message" : [
+      "Cannot drop table feature `<feature>` because some other features (<dependentFeatures>) in this table depends on `<feature>`.",
+      "Consider dropping them first before dropping this feature."
+    ],
+    "sqlState" : "0AKDE"
+  },
   "DELTA_FEATURE_DROP_FEATURE_NOT_PRESENT" : {
     "message" : [
       "Cannot drop <feature> from this table because it is not currently present in the table's protocol."

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

@@ -2334,6 +2334,14 @@ trait DeltaErrorsBase
       messageParameters = Array(feature))
   }
 
+  def dropTableFeatureFailedBecauseOfDependentFeatures(
+      feature: String,
+      dependentFeatures: Seq[String]): DeltaTableFeatureException = {
+    new DeltaTableFeatureException(
+      errorClass = "DELTA_FEATURE_DROP_DEPENDENT_FEATURE",
+      messageParameters = Array(feature, dependentFeatures.mkString(", "), feature))
+  }
+
   def dropTableFeatureConflictRevalidationFailed(
       conflictingCommit: Option[CommitInfo] = None): DeltaTableFeatureException = {
     val concurrentCommit = DeltaErrors.concurrentModificationExceptionMsg(

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

@@ -15,8 +15,11 @@
  */
 
 package org.apache.spark.sql.delta
+
 import java.util.concurrent.TimeUnit
 
+import scala.util.control.NonFatal
+
 import org.apache.spark.sql.delta.catalog.DeltaTableV2
 import org.apache.spark.sql.delta.commands.{AlterTableSetPropertiesDeltaCommand, AlterTableUnsetPropertiesDeltaCommand, DeltaReorgTableCommand, DeltaReorgTableMode, DeltaReorgTableSpec}
 import org.apache.spark.sql.delta.metering.DeltaLogging
@@ -187,6 +190,29 @@ case class InCommitTimestampsPreDowngradeCommand(table: DeltaTableV2)
   }
 }
 
+case class VacuumProtocolCheckPreDowngradeCommand(table: DeltaTableV2)
+  extends PreDowngradeTableFeatureCommand
+  with DeltaLogging {
+
+  /**
+   * Returns true when it performs a cleaning action. When no action was required
+   * it returns false.
+   * For downgrading the [[VacuumProtocolCheckTableFeature]], we don't need remove any traces, we
+   * just need to remove the feature from the [[Protocol]].
+   */
+  override def removeFeatureTracesIfNeeded(): Boolean = {
+    val dependentFeatures = VacuumProtocolCheckTableFeature.otherFeaturesRequiringThisFeature
+    val dependentFeaturesInProtocol =
+      dependentFeatures.filter(table.initialSnapshot.protocol.isFeatureSupported(_))
+    if (dependentFeaturesInProtocol.nonEmpty) {
+      val dependentFeatureNames = dependentFeaturesInProtocol.map(_.name)
+      throw DeltaErrors.dropTableFeatureFailedBecauseOfDependentFeatures(
+        VacuumProtocolCheckTableFeature.name, dependentFeatureNames.toSeq)
+    }
+    false
+  }
+}
+
 case class TypeWideningPreDowngradeCommand(table: DeltaTableV2)
   extends PreDowngradeTableFeatureCommand
   with DeltaLogging {

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

@@ -729,6 +729,22 @@ object InCommitTimestampTableFeature
  */
 object VacuumProtocolCheckTableFeature
   extends ReaderWriterFeature(name = "vacuumProtocolCheck")
+  with RemovableFeature {
+
+  val otherFeaturesRequiringThisFeature = Set(ManagedCommitTableFeature)
+
+  override def preDowngradeCommand(table: DeltaTableV2): PreDowngradeTableFeatureCommand = {
+    VacuumProtocolCheckPreDowngradeCommand(table)
+  }
+
+  // The delta snapshot doesn't have any trace of the [[VacuumProtocolCheckTableFeature]] feature.
+  // Other than it being present in PROTOCOL, which will be handled by the table feature downgrade
+  // command once this method returns true.
+  override def validateRemoval(snapshot: Snapshot): Boolean = true
+
+  // None of the actions uses [[VacuumProtocolCheckTableFeature]]
+  override def actionUsesFeature(action: Action): Boolean = false
+}
 
 /**
  * Features below are for testing only, and are being registered to the system only in the testing

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

@@ -3534,6 +3534,78 @@ trait DeltaProtocolVersionSuiteBase extends QueryTest
     testV2CheckpointTableFeatureDrop(V2Checkpoint.Format.PARQUET, true, true)
   }
 
+  private def testRemoveVacuumProtocolCheckTableFeature(
+      enableFeatureInitially: Boolean,
+      additionalTableProperties: Seq[(String, String)] = Seq.empty,
+      downgradeFailsWithException: Option[String] = None,
+      featureExpectedAtTheEnd: Boolean = false): Unit = {
+    val featureName = VacuumProtocolCheckTableFeature.name
+    withTempDir { dir =>
+      val deltaLog = DeltaLog.forTable(spark, dir)
+      val finalAdditionalTableProperty = if (enableFeatureInitially) {
+        additionalTableProperties ++
+          Seq((s"$FEATURE_PROP_PREFIX${featureName}", "supported"))
+      } else {
+        additionalTableProperties
+      }
+      var additionalTablePropertyString =
+        finalAdditionalTableProperty.map { case (k, v) => s"'$k' = '$v'" }.mkString(", ")
+      if (additionalTablePropertyString.nonEmpty) {
+        additionalTablePropertyString = s", $additionalTablePropertyString"
+      }
+      sql(
+        s"""CREATE TABLE delta.`${deltaLog.dataPath}` (id bigint) USING delta
+           |TBLPROPERTIES (
+           |  delta.minReaderVersion = $TABLE_FEATURES_MIN_READER_VERSION,
+           |  delta.minWriterVersion = $TABLE_FEATURES_MIN_WRITER_VERSION
+           |  $additionalTablePropertyString
+           |)""".stripMargin)
+
+      val protocol = deltaLog.update().protocol
+      assert(protocol.minReaderVersion == TABLE_FEATURES_MIN_READER_VERSION)
+      assert(protocol.minWriterVersion == TABLE_FEATURES_MIN_WRITER_VERSION)
+      assert(protocol.readerFeatures.get.contains(featureName)
+        === enableFeatureInitially)
+      downgradeFailsWithException match {
+        case Some(exceptionClass) =>
+          val e = intercept[DeltaTableFeatureException] {
+            AlterTableDropFeatureDeltaCommand(DeltaTableV2(spark, deltaLog.dataPath), featureName)
+              .run(spark)
+          }
+          assert(e.getErrorClass == exceptionClass)
+        case None =>
+          AlterTableDropFeatureDeltaCommand(DeltaTableV2(spark, deltaLog.dataPath), featureName)
+            .run(spark)
+      }
+      val latestProtocolReaderFeatures = deltaLog.update().protocol.readerFeatures.getOrElse(Set())
+      assert(
+        latestProtocolReaderFeatures.contains(VacuumProtocolCheckTableFeature.name) ===
+          featureExpectedAtTheEnd)
+      assertPropertiesAndShowTblProperties(deltaLog, tableHasFeatures = featureExpectedAtTheEnd)
+    }
+  }
+
+  test("Remove VacuumProtocolCheckTableFeature when it was enabled") {
+    testRemoveVacuumProtocolCheckTableFeature(enableFeatureInitially = true)
+  }
+
+  test("Removing VacuumProtocolCheckTableFeature should fail when dependent feature " +
+      "Managed Commit is enabled") {
+    testRemoveVacuumProtocolCheckTableFeature(
+      enableFeatureInitially = true,
+      additionalTableProperties = Seq(
+        (s"$FEATURE_PROP_PREFIX${ManagedCommitTableFeature.name}", "supported")),
+      downgradeFailsWithException = Some("DELTA_FEATURE_DROP_DEPENDENT_FEATURE"),
+      featureExpectedAtTheEnd = true)
+  }
+
+  test("Removing VacuumProtocolCheckTableFeature should fail when it is not enabled") {
+    testRemoveVacuumProtocolCheckTableFeature(
+      enableFeatureInitially = false,
+      downgradeFailsWithException = Some("DELTA_FEATURE_DROP_FEATURE_NOT_PRESENT")
+    )
+  }
+
   private def validateICTRemovalMetrics(
       usageLogs: Seq[UsageRecord],
       expectEnablementProperty: Boolean,