v1

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

@@ -3493,6 +3493,99 @@ trait DeltaProtocolVersionSuiteBase extends QueryTest
     }
   }
 
+  def validVersions = Seq((1, 1), (1, 2), (1, 3), (1, 4), (2, 5), (2, 6), (1, 7), (3, 7))
+  def invalidVersions = Seq((2, 2), (2, 3))
+  for ((readerVersion, writerVersion) <- validVersions ++ invalidVersions)
+  test(s"Legacy features are added when setting legacy versions: " +
+      s"readerVersionToSet = $readerVersion, writerVersionToSet = $writerVersion") {
+    withTempDir { dir =>
+      val deltaLog = DeltaLog.forTable(spark, dir)
+
+      // Creates a table with (3, 7) versions with the given table feature.
+      createTableWithFeature(
+        deltaLog,
+        TestRemovableWriterFeature,
+        TestRemovableWriterFeature.TABLE_PROP_KEY)
+
+      sql(
+        s"""
+           |ALTER TABLE delta.`${deltaLog.dataPath}` SET TBLPROPERTIES (
+           |  'delta.minReaderVersion' = $readerVersion,
+           |  'delta.minWriterVersion' = $writerVersion
+           |)""".stripMargin)
+
+      val expected = Protocol(readerVersion, writerVersion)
+        .implicitlySupportedFeatures + TestRemovableWriterFeature
+      assert(deltaLog.update().protocol.readerAndWriterFeatureNames === expected.map(_.name))
+    }
+  }
+
+  test(s"Setting protocol versions in a legacy table should not add any features") {
+    withTempDir { dir =>
+      val deltaLog = DeltaLog.forTable(spark, dir)
+
+      sql(
+        s"""CREATE TABLE delta.`${deltaLog.dataPath}` (id bigint) USING delta
+           |TBLPROPERTIES (
+           |delta.minReaderVersion = '1',
+           |delta.minWriterVersion = '2'
+           |)""".stripMargin)
+
+      sql(
+        s"""
+           |ALTER TABLE delta.`${deltaLog.dataPath}` SET TBLPROPERTIES (
+           |  'delta.minReaderVersion' = '1',
+           |  'delta.minWriterVersion' = '4'
+           |)""".stripMargin)
+
+      // There should be no explicitly added legacy features.
+      assert(deltaLog.update().protocol == Protocol(1, 4))
+    }
+  }
+
+  for {
+    setLegacyVersions <- BOOLEAN_DOMAIN
+    tableFeatureToAdd <- Seq(TestRemovableWriterFeature, TestRemovableReaderWriterFeature)
+  } test(s"SOP for downgrading to legacy protocol versions for tables created " +
+      s"with table features. " +
+      s"setLegacyVersions: $setLegacyVersions, tableFeatureToAdd: ${tableFeatureToAdd.name}") {
+    withTempDir { dir =>
+      val deltaLog = DeltaLog.forTable(spark, dir)
+
+      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,
+           |delta.feature.${tableFeatureToAdd.name} = 'supported',
+           |delta.feature.${ChangeDataFeedTableFeature.name} = 'supported'
+           |)""".stripMargin)
+
+      if (setLegacyVersions) {
+        sql(
+          s"""
+             |ALTER TABLE delta.`${deltaLog.dataPath}` SET TBLPROPERTIES (
+             |  'delta.minReaderVersion' = ${ChangeDataFeedTableFeature.minReaderVersion},
+             |  'delta.minWriterVersion' = ${ChangeDataFeedTableFeature.minWriterVersion}
+             |)""".stripMargin)
+      }
+
+      AlterTableDropFeatureDeltaCommand(
+        DeltaTableV2(spark, deltaLog.dataPath),
+        tableFeatureToAdd.name,
+        truncateHistory = tableFeatureToAdd.isReaderWriterFeature).run(spark)
+
+      val expectedProtocol = if (setLegacyVersions) {
+        Protocol(1, 4)
+      } else {
+        Protocol(TABLE_FEATURES_MIN_READER_VERSION, TABLE_FEATURES_MIN_WRITER_VERSION)
+          .withFeature(ChangeDataFeedTableFeature)
+
+      }
+      assert(deltaLog.update().protocol === expectedProtocol)
+    }
+  }
+
   private def dropV2CheckpointsTableFeature(spark: SparkSession, log: DeltaLog): Unit = {
     spark.sql(s"ALTER TABLE delta.`${log.dataPath}` DROP FEATURE " +
       s"`${V2CheckpointTableFeature.name}`")