diff --git a/br/pkg/lightning/importer/BUILD.bazel b/br/pkg/lightning/importer/BUILD.bazel index c3e804cc77397..a49a1db60caf1 100644 --- a/br/pkg/lightning/importer/BUILD.bazel +++ b/br/pkg/lightning/importer/BUILD.bazel @@ -83,6 +83,8 @@ go_library( "@com_github_tikv_pd_client//:client", "@io_etcd_go_etcd_client_v3//:client", "@org_golang_google_grpc//:grpc", + "@org_golang_google_grpc//codes", + "@org_golang_google_grpc//status", "@org_golang_x_exp//maps", "@org_golang_x_exp//slices", "@org_golang_x_sync//errgroup", diff --git a/br/pkg/lightning/importer/table_import.go b/br/pkg/lightning/importer/table_import.go index 25ddc004dde59..29408dcf115d0 100644 --- a/br/pkg/lightning/importer/table_import.go +++ b/br/pkg/lightning/importer/table_import.go @@ -54,6 +54,8 @@ import ( "go.uber.org/multierr" "go.uber.org/zap" "golang.org/x/exp/slices" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" ) // TableImporter is a helper struct to import a table. @@ -988,15 +990,26 @@ func (tr *TableImporter) postProcess( var remoteChecksum *local.RemoteChecksum remoteChecksum, err = DoChecksum(ctx, tr.tableInfo) + failpoint.Inject("checksum-error", func() { + tr.logger.Info("failpoint checksum-error injected.") + remoteChecksum = nil + err = status.Error(codes.Unknown, "Checksum meets error.") + }) if err != nil { - return false, err + if rc.cfg.PostRestore.Checksum != config.OpLevelOptional { + return false, err + } + tr.logger.Warn("do checksum failed, will skip this error and go on", log.ShortError(err)) + err = nil } - err = tr.compareChecksum(remoteChecksum, localChecksum) - // with post restore level 'optional', we will skip checksum error - if rc.cfg.PostRestore.Checksum == config.OpLevelOptional { - if err != nil { - tr.logger.Warn("compare checksum failed, will skip this error and go on", log.ShortError(err)) - err = nil + if remoteChecksum != nil { + err = tr.compareChecksum(remoteChecksum, localChecksum) + // with post restore level 'optional', we will skip checksum error + if rc.cfg.PostRestore.Checksum == config.OpLevelOptional { + if err != nil { + tr.logger.Warn("compare checksum failed, will skip this error and go on", log.ShortError(err)) + err = nil + } } } } else { @@ -1054,11 +1067,12 @@ func (tr *TableImporter) postProcess( case forcePostProcess || !rc.cfg.PostRestore.PostProcessAtLast: err := tr.analyzeTable(ctx, rc.db) // witch post restore level 'optional', we will skip analyze error - if rc.cfg.PostRestore.Analyze == config.OpLevelOptional { - if err != nil { - tr.logger.Warn("analyze table failed, will skip this error and go on", log.ShortError(err)) - err = nil + if err != nil { + if rc.cfg.PostRestore.Analyze != config.OpLevelOptional { + return false, err } + tr.logger.Warn("analyze table failed, will skip this error and go on", log.ShortError(err)) + err = nil } saveCpErr := rc.saveStatusCheckpoint(ctx, tr.tableName, checkpoints.WholeTableEngineID, err, checkpoints.CheckpointStatusAnalyzed) if err = firstErr(err, saveCpErr); err != nil { diff --git a/br/tests/lightning_routes/config.toml b/br/tests/lightning_routes/config.toml index bb54609dd03b1..74913091c5916 100644 --- a/br/tests/lightning_routes/config.toml +++ b/br/tests/lightning_routes/config.toml @@ -8,3 +8,6 @@ schema-pattern = "routes_a*" table-pattern = "t*" target-schema = "routes_b" target-table = "u" + +[post-restore] +checksum = "optional" diff --git a/br/tests/lightning_routes/run.sh b/br/tests/lightning_routes/run.sh index 1db0ce2035021..5ae757eb0bd43 100755 --- a/br/tests/lightning_routes/run.sh +++ b/br/tests/lightning_routes/run.sh @@ -4,12 +4,17 @@ set -eux +echo "testing checksum-error..." +export GO_FAILPOINTS="github.com/pingcap/tidb/br/pkg/lightning/importer/checksum-error=1*return()" + run_sql 'DROP DATABASE IF EXISTS routes_a0;' run_sql 'DROP DATABASE IF EXISTS routes_a1;' run_sql 'DROP DATABASE IF EXISTS routes_b;' run_lightning +echo "test checksum-error success!" + run_sql 'SELECT count(1), sum(x) FROM routes_b.u;' check_contains 'count(1): 4' check_contains 'sum(x): 259'