feat: Enable destinations to completely skip migration that are not supported (#1560)

Some destinations don't support all migrations types. This allows them to opt out

Author: bbernays

plugin/testing_write.go

@@ -48,6 +48,12 @@ type SafeMigrations struct {
 	MovePKToCQOnly         bool
 }
 
+// Migrations defines which migrations should be skipped completely
+type Migrations struct {
+	RemoveUniqueConstraint bool
+	MovePKToCQOnly         bool
+}
+
 type WriterTestSuiteTests struct {
 	// SkipUpsert skips testing with message.Insert and Upsert=true.
 	// Usually when a destination is not supporting primary keys
@@ -68,6 +74,8 @@ type WriterTestSuiteTests struct {
 	// SafeMigrations defines which tests should work with force migration
 	// and which should pass with safe migration
 	SafeMigrations SafeMigrations
+
+	SkipSpecificMigrations Migrations
 }
 
 type NewPluginFunc func() *Plugin

plugin/testing_write_migrate.go

@@ -250,6 +250,9 @@ func (s *WriterTestSuite) testMigrate(
 	})
 
 	t.Run("remove_unique_constraint_only"+suffix, func(t *testing.T) {
+		if s.tests.SkipSpecificMigrations.RemoveUniqueConstraint {
+			t.Skip("skipping test completely: remove_unique_constraint_only")
+		}
 		if !forceMigrate && !s.tests.SafeMigrations.RemoveUniqueConstraint {
 			t.Skip("skipping test: remove_unique_constraint_only")
 		}
@@ -277,6 +280,9 @@ func (s *WriterTestSuite) testMigrate(
 	})
 
 	t.Run("move_to_cq_id_only"+suffix, func(t *testing.T) {
+		if s.tests.SkipSpecificMigrations.MovePKToCQOnly {
+			t.Skip("skipping test completely: move_to_cq_id_only")
+		}
 		if !forceMigrate && !s.tests.SafeMigrations.MovePKToCQOnly {
 			t.Skip("skipping test: move_to_cq_id_only")
 		}
@@ -303,8 +309,11 @@ func (s *WriterTestSuite) testMigrate(
 		}
 	})
 	t.Run("move_to_cq_id_only_adding_pkc"+suffix, func(t *testing.T) {
+		if s.tests.SkipSpecificMigrations.MovePKToCQOnly {
+			t.Skip("skipping test completely: move_to_cq_id_only_adding_pkc")
+		}
 		if !forceMigrate && !s.tests.SafeMigrations.MovePKToCQOnly {
-			t.Skip("skipping test: move_to_cq_id_only_adding_pk")
+			t.Skip("skipping test: move_to_cq_id_only_adding_pkc")
 		}
 		tableName := "cq_move_to_cq_id_only_adding_pkc" + suffix + "_" + tableUUIDSuffix()
 		source := &schema.Table{