br/restore: use a more relax check for changefeeds when restoring

br/pkg/task/restore.go

@@ -754,6 +754,14 @@ func runRestore(c context.Context, g glue.Glue, cmdName string, cfg *RestoreConf
 	if err != nil {
 		return errors.Trace(err)
 	}
+	if cfg.CheckRequirements {
+		err := checkIncompatibleChangefeed(ctx, backupMeta.EndVersion, mgr.GetDomain().GetEtcdClient())
+		log.Info("Checking incompatible TiCDC changefeeds before restoring.",
+			logutil.ShortError(err), zap.Uint64("restore-ts", backupMeta.EndVersion))
+		if err != nil {
+			return errors.Trace(err)
+		}
+	}
 
 	backupVersion := version.NormalizeBackupVersion(backupMeta.ClusterVersion)
 	if cfg.CheckRequirements && backupVersion != nil {

br/pkg/task/stream.go

@@ -1116,15 +1116,16 @@ func checkTaskExists(ctx context.Context, cfg *RestoreConfig, etcdCLI *clientv3.
 			"create log-backup task again and create a full backup on this cluster", tasks[0].Info.Name)
 	}
 
-	// check cdc changefeed
-	if cfg.CheckRequirements {
-		nameSet, err := cdcutil.GetCDCChangefeedNameSet(ctx, etcdCLI)
-		if err != nil {
-			return err
-		}
-		if !nameSet.Empty() {
-			return errors.Errorf("%splease remove changefeed(s) before restore", nameSet.MessageToUser())
-		}
+	return nil
+}
+
+func checkIncompatibleChangefeed(ctx context.Context, backupTS uint64, etcdCLI *clientv3.Client) error {
+	nameSet, err := cdcutil.GetIncompatibleChangefeedsWithSafeTS(ctx, etcdCLI, backupTS)
+	if err != nil {
+		return err
+	}
+	if !nameSet.Empty() {
+		return errors.Errorf("%splease remove changefeed(s) before restore", nameSet.MessageToUser())
 	}
 	return nil
 }

lightning/pkg/importer/precheck_impl.go

@@ -837,7 +837,7 @@ func (ci *CDCPITRCheckItem) Check(ctx context.Context) (*precheck.CheckResult, e
 		errorMsg = append(errorMsg, fmt.Sprintf("found PiTR log streaming task(s): %v,", names))
 	}
 
-	nameSet, err := cdcutil.GetCDCChangefeedNameSet(ctx, ci.etcdCli)
+	nameSet, err := cdcutil.GetRunningChangefeeds(ctx, ci.etcdCli)
 	if err != nil {
 		return nil, errors.Trace(err)
 	}

pkg/executor/importer/precheck.go

@@ -130,7 +130,7 @@ func (*LoadDataController) checkCDCPiTRTasks(ctx context.Context) error {
 		return exeerrors.ErrLoadDataPreCheckFailed.FastGenByArgs(fmt.Sprintf("found PiTR log streaming task(s): %v,", names))
 	}
 
-	nameSet, err := cdcutil.GetCDCChangefeedNameSet(ctx, cli.GetClient())
+	nameSet, err := cdcutil.GetRunningChangefeeds(ctx, cli.GetClient())
 	if err != nil {
 		return errors.Trace(err)
 	}

pkg/util/cdcutil/BUILD.bazel

@@ -6,6 +6,7 @@ go_library(
     importpath = "github.com/pingcap/tidb/pkg/util/cdcutil",
     visibility = ["//visibility:public"],
     deps = [
+        "//pkg/util/mathutil",
         "@com_github_pingcap_errors//:errors",
         "@com_github_pingcap_log//:log",
         "@io_etcd_go_etcd_client_v3//:client",
@@ -16,10 +17,13 @@ go_library(
 go_test(
     name = "cdcutil_test",
     timeout = "short",
-    srcs = ["cdc_test.go"],
+    srcs = [
+        "cdc_test.go",
+        "export_for_test.go",
+    ],
+    embed = [":cdcutil"],
     flaky = True,
     deps = [
-        ":cdcutil",
         "@com_github_stretchr_testify//require",
         "@io_etcd_go_etcd_client_v3//:client",
         "@io_etcd_go_etcd_tests_v3//integration",