Skip to content

localbackend: move to ingestctrl and rename package #68382

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ti-chi-bot merged 9 commits into from
May 18, 2026
Merged

localbackend: move to ingestctrl and rename package #68382

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
c8f00b3
ingestor,lightning,ddl: move local backend to ingestctrl
D3Hunter May 14, 2026
74e45ad
ingestor,lightning,ddl: rename local package to ingestctrl
D3Hunter May 14, 2026
227897b
bazel
D3Hunter May 14, 2026
e311277
bazel
D3Hunter May 14, 2026
814def4
lint
D3Hunter May 14, 2026
5bfb0f2
lint
D3Hunter May 14, 2026
fa622f7
lint
D3Hunter May 14, 2026
5c3fad9
lightning: allow ingestctrl logs in filter core
D3Hunter May 15, 2026
d9d87c3
lightning: fix ingestctrl caller-path log filter
D3Hunter May 15, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion Makefile
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -556,7 +556,7 @@ mock_import: mockgen
tools/bin/mockgen -package mock github.com/pingcap/tidb/pkg/lightning/backend Backend,EngineWriter,TargetInfoGetter > br/pkg/mock/backend.go
tools/bin/mockgen -package mock github.com/pingcap/tidb/pkg/lightning/common ChunkFlushStatus > br/pkg/mock/common.go
tools/bin/mockgen -package mock github.com/pingcap/tidb/pkg/lightning/backend/encode Encoder,EncodingBuilder,Rows,Row > br/pkg/mock/encode.go
tools/bin/mockgen -package mocklocal github.com/pingcap/tidb/pkg/lightning/backend/local DiskUsage,TiKVModeSwitcher,StoreHelper > br/pkg/mock/mocklocal/local.go
tools/bin/mockgen -package mocklocal github.com/pingcap/tidb/pkg/ingestor/ingestctrl DiskUsage,TiKVModeSwitcher,StoreHelper > br/pkg/mock/mocklocal/local.go
tools/bin/mockgen -package mock github.com/pingcap/tidb/br/pkg/utils TaskRegister > br/pkg/mock/task_register.go
tools/bin/mockgen -package mock github.com/pingcap/tidb/pkg/dxf/framework/taskexecutor TaskTable,TaskExecutor,Extension > pkg/dxf/framework/mock/task_executor_mock.go
tools/bin/mockgen -package mock github.com/pingcap/tidb/pkg/dxf/framework/scheduler Scheduler,CleanUpRoutine,TaskManager > pkg/dxf/framework/mock/scheduler_mock.go
Expand Down
4 changes: 2 additions & 2 deletions br/pkg/mock/mocklocal/local.go
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

8 changes: 4 additions & 4 deletions build/nogo_config.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -423,7 +423,7 @@
"pkg/lightning/mydump/": "ignore mydump code",
"pkg/lightning/checkpoints/checkpoints.go": "ignore checkpoints code",
"pkg/lightning/checkpoints/glue_checkpoint.go": "ignore glue_checkpoint code",
"pkg/lightning/backend/local/": "ignore local code",
"pkg/ingestor/ingestctrl/": "ignore local code",
"pkg/lightning/backend/tidb/tidb.go": "ignore tidb code",
"pkg/lightning/backend/tidb/tidb_test.go": "ignore tidb code",
"pkg/lightning/checkpoints/checkpoints_test.go": "ignore checkpoints code",
Expand Down Expand Up @@ -657,7 +657,7 @@
"pkg/util/chunk": "ignore util/chunk",
"pkg/lightning/mydump/": "more than 50",
"lightning/pkg/importer/": "more than 50",
"pkg/lightning/backend/local/": "more than 50",
"pkg/ingestor/ingestctrl/": "more than 50",
"br/pkg/restore/": "more than 50",
"br/pkg/storage/": "more than 50",
"pkg/ddl/tests/partition/": "more than 50"
Expand Down Expand Up @@ -1065,8 +1065,8 @@
"br/pkg/streamhelper/advancer_cliext.go": "github.com/golang/protobuf deprecated",
"pkg/lightning/checkpoints/checkpoints.go": "cfg.TikvImporter.Addr is deprecated",
"pkg/lightning/checkpoints/glue_checkpoint.go": "cfg.TikvImporter.Addr is deprecated",
"pkg/lightning/backend/local/local.go": "grpc Compressor/Decompressor is deprecated",
"pkg/lightning/backend/local/compress.go": "grpc Compressor/Decompressor is deprecated",
"pkg/ingestor/ingestctrl/local.go": "grpc Compressor/Decompressor is deprecated",
"pkg/ingestor/ingestctrl/compress.go": "grpc Compressor/Decompressor is deprecated",
"pkg/domain/infosync/resource_manager_client.go": "github.com/golang/protobuf deprecated"
},
"only_files": {
Expand Down
4 changes: 2 additions & 2 deletions lightning/pkg/importer/BUILD.bazel
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,12 +33,12 @@ go_library(
"//pkg/config",
"//pkg/ddl",
"//pkg/errno",
"//pkg/ingestor/ingestctrl",
"//pkg/keyspace",
"//pkg/kv",
"//pkg/lightning/backend",
"//pkg/lightning/backend/encode",
"//pkg/lightning/backend/kv",
"//pkg/lightning/backend/local",
"//pkg/lightning/backend/tidb",
"//pkg/lightning/common",
"//pkg/lightning/config",
Expand Down Expand Up @@ -139,11 +139,11 @@ go_test(
"//pkg/config/kerneltype",
"//pkg/ddl",
"//pkg/errno",
"//pkg/ingestor/ingestctrl",
"//pkg/kv",
"//pkg/lightning/backend",
"//pkg/lightning/backend/encode",
"//pkg/lightning/backend/kv",
"//pkg/lightning/backend/local",
"//pkg/lightning/backend/tidb",
"//pkg/lightning/common",
"//pkg/lightning/config",
Expand Down
20 changes: 10 additions & 10 deletions lightning/pkg/importer/checksum_helper.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,8 +19,8 @@ import (

"github.com/pingcap/errors"
"github.com/pingcap/tidb/br/pkg/pdutil"
"github.com/pingcap/tidb/pkg/ingestor/ingestctrl"
Comment thread
coderabbitai[bot] marked this conversation as resolved.
"github.com/pingcap/tidb/pkg/kv"
"github.com/pingcap/tidb/pkg/lightning/backend/local"
"github.com/pingcap/tidb/pkg/lightning/common"
"github.com/pingcap/tidb/pkg/lightning/config"
"github.com/pingcap/tidb/pkg/lightning/importdef"
Expand All @@ -31,7 +31,7 @@ import (
)

// NewChecksumManager creates a new checksum manager.
func NewChecksumManager(ctx context.Context, rc *Controller, store kv.Storage) (local.ChecksumManager, error) {
func NewChecksumManager(ctx context.Context, rc *Controller, store kv.Storage) (ingestctrl.ChecksumManager, error) {
// if we don't need checksum, just return nil
if rc.cfg.TikvImporter.Backend == config.BackendTiDB || rc.cfg.PostRestore.Checksum == config.OpLevelOff {
return nil, nil
Expand All @@ -43,30 +43,30 @@ func NewChecksumManager(ctx context.Context, rc *Controller, store kv.Storage) (
}

// for v4.0.0 or upper, we can use the gc ttl api
var manager local.ChecksumManager
var manager ingestctrl.ChecksumManager
if pdVersion.Major >= 4 && !rc.cfg.PostRestore.ChecksumViaSQL {
backoffWeight, err := common.GetBackoffWeightFromDB(ctx, rc.db)
// only set backoff weight when it's smaller than default value
if err == nil && backoffWeight >= local.DefaultBackoffWeight {
if err == nil && backoffWeight >= ingestctrl.DefaultBackoffWeight {
logutil.Logger(ctx).Info("get tidb_backoff_weight", zap.Int("backoff_weight", backoffWeight))
} else {
logutil.Logger(ctx).Info("set tidb_backoff_weight to default", zap.Int("backoff_weight", local.DefaultBackoffWeight))
backoffWeight = local.DefaultBackoffWeight
logutil.Logger(ctx).Info("set tidb_backoff_weight to default", zap.Int("backoff_weight", ingestctrl.DefaultBackoffWeight))
backoffWeight = ingestctrl.DefaultBackoffWeight
}

manager = local.NewTiKVChecksumManager(store.GetClient(), rc.pdCli, uint(rc.cfg.TiDB.DistSQLScanConcurrency), backoffWeight, rc.resourceGroupName, rc.taskType)
manager = ingestctrl.NewTiKVChecksumManager(store.GetClient(), rc.pdCli, uint(rc.cfg.TiDB.DistSQLScanConcurrency), backoffWeight, rc.resourceGroupName, rc.taskType)
} else {
manager = local.NewTiDBChecksumExecutor(rc.db)
manager = ingestctrl.NewTiDBChecksumExecutor(rc.db)
}

return manager, nil
}

// DoChecksum do checksum for tables.
// table should be in <db>.<table>, format. e.g. foo.bar
func DoChecksum(ctx context.Context, table *importdef.TableInfo) (*local.RemoteChecksum, error) {
func DoChecksum(ctx context.Context, table *importdef.TableInfo) (*ingestctrl.RemoteChecksum, error) {
var err error
manager, ok := ctx.Value(&checksumManagerKey).(local.ChecksumManager)
manager, ok := ctx.Value(&checksumManagerKey).(ingestctrl.ChecksumManager)
if !ok {
return nil, errors.New("No gcLifeTimeManager found in context, check context initialization")
}
Expand Down
6 changes: 3 additions & 3 deletions lightning/pkg/importer/get_pre_info.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,10 +30,10 @@ import (
ropts "github.com/pingcap/tidb/lightning/pkg/importer/opts"
"github.com/pingcap/tidb/pkg/ddl"
"github.com/pingcap/tidb/pkg/errno"
"github.com/pingcap/tidb/pkg/ingestor/ingestctrl"
"github.com/pingcap/tidb/pkg/lightning/backend"
"github.com/pingcap/tidb/pkg/lightning/backend/encode"
"github.com/pingcap/tidb/pkg/lightning/backend/kv"
"github.com/pingcap/tidb/pkg/lightning/backend/local"
"github.com/pingcap/tidb/pkg/lightning/backend/tidb"
"github.com/pingcap/tidb/pkg/lightning/common"
"github.com/pingcap/tidb/pkg/lightning/config"
Expand Down Expand Up @@ -145,7 +145,7 @@ func NewTargetInfoGetterImpl(
case config.BackendTiDB:
backendTargetInfoGetter = tidb.NewTargetInfoGetter(targetDB)
case config.BackendLocal:
backendTargetInfoGetter = local.NewTargetInfoGetter(tls, targetDB, pdHTTPCli)
backendTargetInfoGetter = ingestctrl.NewTargetInfoGetter(tls, targetDB, pdHTTPCli)
default:
return nil, common.ErrUnknownBackend.GenWithStackByArgs(cfg.TikvImporter.Backend)
}
Expand Down Expand Up @@ -299,7 +299,7 @@ func NewPreImportInfoGetter(
case config.BackendTiDB:
encBuilder = tidb.NewEncodingBuilder()
case config.BackendLocal:
encBuilder = local.NewEncodingBuilder(context.Background())
encBuilder = ingestctrl.NewEncodingBuilder(context.Background())
default:
return nil, common.ErrUnknownBackend.GenWithStackByArgs(cfg.TikvImporter.Backend)
}
Expand Down
30 changes: 15 additions & 15 deletions lightning/pkg/importer/import.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,11 +40,11 @@ import (
"github.com/pingcap/tidb/lightning/pkg/errormanager"
"github.com/pingcap/tidb/lightning/pkg/progress"
tidbconfig "github.com/pingcap/tidb/pkg/config"
"github.com/pingcap/tidb/pkg/ingestor/ingestctrl"
"github.com/pingcap/tidb/pkg/keyspace"
tidbkv "github.com/pingcap/tidb/pkg/kv"
"github.com/pingcap/tidb/pkg/lightning/backend"
"github.com/pingcap/tidb/pkg/lightning/backend/encode"
"github.com/pingcap/tidb/pkg/lightning/backend/local"
"github.com/pingcap/tidb/pkg/lightning/backend/tidb"
"github.com/pingcap/tidb/pkg/lightning/common"
"github.com/pingcap/tidb/pkg/lightning/config"
Expand Down Expand Up @@ -240,7 +240,7 @@ type Controller struct {
preInfoGetter PreImportInfoGetter
precheckItemBuilder *PrecheckItemBuilder
encBuilder encode.EncodingBuilder
tikvModeSwitcher local.TiKVModeSwitcher
tikvModeSwitcher ingestctrl.TiKVModeSwitcher

keyspaceName string
resourceGroupName string
Expand Down Expand Up @@ -369,19 +369,19 @@ func NewImportControllerWithPauser(
encodingBuilder = tidb.NewEncodingBuilder()
backendObj = tidb.NewTiDBBackend(ctx, db, cfg, errorMgr)
case config.BackendLocal:
var rLimit local.RlimT
rLimit, err = local.GetSystemRLimit()
var rLimit ingestctrl.RlimT
rLimit, err = ingestctrl.GetSystemRLimit()
if err != nil {
return nil, err
}
maxOpenFiles := int(rLimit / local.RlimT(cfg.App.TableConcurrency))
maxOpenFiles := int(rLimit / ingestctrl.RlimT(cfg.App.TableConcurrency))
// check overflow
if maxOpenFiles < 0 {
maxOpenFiles = math.MaxInt32
}

addrs := strings.Split(cfg.TiDB.PdAddr, ",")
pdCli, err = pd.NewClientWithContext(ctx, componentName, addrs, tls.ToPDSecurityOption(), local.PDClientOptions()...)
pdCli, err = pd.NewClientWithContext(ctx, componentName, addrs, tls.ToPDSecurityOption(), ingestctrl.PDClientOptions()...)
if err != nil {
return nil, errors.Trace(err)
}
Expand All @@ -402,14 +402,14 @@ func NewImportControllerWithPauser(
}

// simple wraps PD client, so no need to close it separately.
pdCliForTiKV, err := local.NewCodecPDClient(pdCli, p.KeyspaceName)
pdCliForTiKV, err := ingestctrl.NewCodecPDClient(pdCli, p.KeyspaceName)
if err != nil {
return nil, common.ErrCreatePDClient.Wrap(err).GenWithStackByArgs()
}

initGlobalConfig(tls.ToTiKVSecurityConfig())

encodingBuilder = local.NewEncodingBuilder(ctx)
encodingBuilder = ingestctrl.NewEncodingBuilder(ctx)

// get resource group name.
exec := common.SQLWithRetry{
Expand Down Expand Up @@ -445,8 +445,8 @@ func NewImportControllerWithPauser(
if isRaftKV2 {
raftKV2SwitchModeDuration = cfg.Cron.SwitchMode.Duration
}
backendConfig := local.NewBackendConfig(cfg, maxOpenFiles, p.KeyspaceName, p.ResourceGroupName, p.TaskType, raftKV2SwitchModeDuration)
backendObj, err = local.NewBackend(ctx, tls, backendConfig, pdCliForTiKV)
backendConfig := ingestctrl.NewBackendConfig(cfg, maxOpenFiles, p.KeyspaceName, p.ResourceGroupName, p.TaskType, raftKV2SwitchModeDuration)
backendObj, err = ingestctrl.NewBackend(ctx, tls, backendConfig, pdCliForTiKV)
if err != nil {
return nil, common.NormalizeOrWrapErr(common.ErrUnknown, err)
}
Expand Down Expand Up @@ -479,7 +479,7 @@ func NewImportControllerWithPauser(

var wrapper backend.TargetInfoGetter
if cfg.TikvImporter.Backend == config.BackendLocal {
wrapper = local.NewTargetInfoGetter(tls, db, pdHTTPCli)
wrapper = ingestctrl.NewTargetInfoGetter(tls, db, pdHTTPCli)
} else {
wrapper = tidb.NewTargetInfoGetter(db)
}
Expand Down Expand Up @@ -544,7 +544,7 @@ func NewImportControllerWithPauser(
preInfoGetter: preInfoGetter,
precheckItemBuilder: preCheckBuilder,
encBuilder: encodingBuilder,
tikvModeSwitcher: local.NewTiKVModeSwitcher(tls.TLSConfig(), pdHTTPCli, logutil.Logger(ctx)),
tikvModeSwitcher: ingestctrl.NewTiKVModeSwitcher(tls.TLSConfig(), pdHTTPCli, logutil.Logger(ctx)),

keyspaceName: p.KeyspaceName,
resourceGroupName: p.ResourceGroupName,
Expand Down Expand Up @@ -760,7 +760,7 @@ func verifyLocalFile(ctx context.Context, cpdb checkpoints.DB, dir string) error
for tableName, engineIDs := range targetTables {
for _, engineID := range engineIDs {
_, eID := backend.MakeUUID(tableName, int64(engineID))
file := local.Engine{UUID: eID}
file := ingestctrl.Engine{UUID: eID}
err := file.Exist(dir)
if err != nil {
logutil.Logger(ctx).Error("can't find local file",
Expand Down Expand Up @@ -1705,7 +1705,7 @@ func (rc *Controller) enforceDiskQuota(ctx context.Context) {
return
}

localBackend := rc.backend.(*local.Backend)
localBackend := rc.backend.(*ingestctrl.Backend)
go func() {
// locker is assigned when we detect the disk quota is exceeded.
// before the disk quota is confirmed exceeded, we keep the diskQuotaLock
Expand Down Expand Up @@ -1733,7 +1733,7 @@ func (rc *Controller) enforceDiskQuota(ctx context.Context) {
}

quota := int64(rc.cfg.TikvImporter.DiskQuota)
largeEngines, inProgressLargeEngines, totalDiskSize, totalMemSize := local.CheckDiskQuota(localBackend, quota)
largeEngines, inProgressLargeEngines, totalDiskSize, totalMemSize := ingestctrl.CheckDiskQuota(localBackend, quota)
if m, ok := metric.FromContext(ctx); ok {
m.LocalStorageUsageBytesGauge.WithLabelValues("disk").Set(float64(totalDiskSize))
m.LocalStorageUsageBytesGauge.WithLabelValues("mem").Set(float64(totalMemSize))
Expand Down
10 changes: 5 additions & 5 deletions lightning/pkg/importer/meta_manager_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,8 +24,8 @@ import (
"github.com/DATA-DOG/go-sqlmock"
"github.com/pingcap/errors"
"github.com/pingcap/tidb/pkg/ddl"
"github.com/pingcap/tidb/pkg/ingestor/ingestctrl"
"github.com/pingcap/tidb/pkg/kv"
"github.com/pingcap/tidb/pkg/lightning/backend/local"
"github.com/pingcap/tidb/pkg/lightning/common"
"github.com/pingcap/tidb/pkg/lightning/importdef"
"github.com/pingcap/tidb/pkg/lightning/log"
Expand Down Expand Up @@ -361,7 +361,7 @@ func (s *metaMgrSuite) prepareMockInner(rowsVal [][]driver.Value, nextRowID *int
s.mockDB.ExpectExec("\\QUPDATE `test`.`table_meta` SET total_kvs_base = ?, total_bytes_base = ?, checksum_base = ?, status = ? WHERE table_id = ? AND task_id = ?\\E").
WithArgs(checksum.SumKVS(), checksum.SumSize(), checksum.Sum(), metaStatusRestoreStarted.String(), int64(1), int64(1)).
WillReturnResult(sqlmock.NewResult(int64(0), int64(1)))
s.checksumMgr.checksum = local.RemoteChecksum{
s.checksumMgr.checksum = ingestctrl.RemoteChecksum{
TotalBytes: checksum.SumSize(),
TotalKVs: checksum.SumKVS(),
Checksum: checksum.Sum(),
Expand Down Expand Up @@ -438,13 +438,13 @@ func TestCheckTasksExclusively(t *testing.T) {
}

type testChecksumMgr struct {
checksum local.RemoteChecksum
checksum ingestctrl.RemoteChecksum
callCnt int
}

var _ local.ChecksumManager = (*testChecksumMgr)(nil)
var _ ingestctrl.ChecksumManager = (*testChecksumMgr)(nil)

func (t *testChecksumMgr) Checksum(ctx context.Context, tableInfo *importdef.TableInfo) (*local.RemoteChecksum, error) {
func (t *testChecksumMgr) Checksum(ctx context.Context, tableInfo *importdef.TableInfo) (*ingestctrl.RemoteChecksum, error) {
t.callCnt++
return &t.checksum, nil
}
Expand Down
Loading
Loading