Downgrade to the minimum supported protocol when a table is using onl…

…y legacy features
delta-io · Mar 13, 2024 · 34caa80 · 34caa80
1 parent 60914cd
commit 34caa80
Show file tree

Hide file tree

Showing 2 changed files with 15 additions and 5 deletions.
diff --git a/spark/src/main/scala/org/apache/spark/sql/delta/actions/TableFeatureSupport.scala b/spark/src/main/scala/org/apache/spark/sql/delta/actions/TableFeatureSupport.scala
@@ -264,11 +264,12 @@ trait TableFeatureSupport { this: Protocol =>
     // remove a legacy feature in a table that only contains legacy features.
     if (to.readerAndWriterFeatureNames.isEmpty) {
       val featureNames = readerAndWriterFeatureNames - droppedFeatureName
-      val sameLegacyFeaturesSupported = featureNames == to.implicitlySupportedFeatures.map(_.name)
+      val toFeatureNames = to.implicitlySupportedFeatures.map(_.name)
+      val legacyFeaturesSupportedByDowngradedPotocol = featureNames.forall(toFeatureNames.contains)
       val minRequiredVersions = TableFeatureProtocolUtils.minimumRequiredVersions(
         featureNames.flatMap(TableFeature.featureNameToFeature).toSeq)
 
-      return sameLegacyFeaturesSupported &&
+      return legacyFeaturesSupportedByDowngradedPotocol &&
         (to.minReaderVersion, to.minWriterVersion) == minRequiredVersions &&
         readerAndWriterFeatures.filterNot(_.isLegacyFeature).size <= 1
     }
@@ -378,9 +379,9 @@ trait TableFeatureSupport { this: Protocol =>
       !newProtocol.supportsReaderFeatures && !newProtocol.supportsWriterFeatures,
       s"Downgraded protocol should not support table features, but got $newProtocol.")
 
-    // Ensure the legacy protocol supports features exactly as the current protocol.
-    if (this.implicitlyAndExplicitlySupportedFeatures ==
-      newProtocol.implicitlyAndExplicitlySupportedFeatures) {
+    // Ensure the legacy protocol supports a superset of features of the current protocol.
+    if (this.implicitlyAndExplicitlySupportedFeatures.forall(
+      newProtocol.implicitlyAndExplicitlySupportedFeatures.contains)) {
       newProtocol
     } else {
       this

diff --git a/spark/src/test/scala/org/apache/spark/sql/delta/DeltaProtocolVersionSuite.scala b/spark/src/test/scala/org/apache/spark/sql/delta/DeltaProtocolVersionSuite.scala
@@ -3438,6 +3438,15 @@ trait DeltaProtocolVersionSuiteBase extends QueryTest
       expectedDowngradedProtocol = protocolWithReaderFeature(TestRemovableReaderWriterFeature))
   }
 
+  test("Downgrade protocol version (1, 4) for CDF tables") {
+    testProtocolVersionDowngrade(
+      initialMinReaderVersion = 3,
+      initialMinWriterVersion = 7,
+      featuresToAdd = Seq(TestRemovableWriterFeature, ChangeDataFeedTableFeature),
+      featuresToRemove = Seq(TestRemovableWriterFeature),
+      expectedDowngradedProtocol = Protocol(1, 4))
+  }
+
   private def dropV2CheckpointsTableFeature(spark: SparkSession, log: DeltaLog): Unit = {
     spark.sql(s"ALTER TABLE delta.`${log.dataPath}` DROP FEATURE " +
       s"`${V2CheckpointTableFeature.name}`")