Skip to content

Commit 042a332

Browse files
CbcWestwolfCbcWestwolf
authored
metrics: add col/idx name(s) for BackfillProgressGauge and BackfillTotalCounter (#58380)
close #58114
1 parent ef7ade7 commit 042a332

File tree

10 files changed

+137
-56
lines changed

10 files changed

+137
-56
lines changed

‎pkg/ddl/backfilling.go

Lines changed: 45 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,7 @@ import (
2020
"encoding/hex"
2121
"fmt"
2222
"strconv"
23+
"strings"
2324
"sync"
2425
"sync/atomic"
2526
"time"
@@ -194,23 +195,49 @@ func newBackfillCtx(id int, rInfo *reorgInfo,
194195
id = int(backfillContextID.Add(1))
195196
}
196197

198+
colOrIdxName := ""
199+
switch rInfo.Job.Type {
200+
case model.ActionAddIndex, model.ActionAddPrimaryKey:
201+
args, err := model.GetModifyIndexArgs(rInfo.Job)
202+
if err != nil {
203+
logutil.DDLLogger().Error("Fail to get ModifyIndexArgs", zap.String("label", label), zap.String("schemaName", schemaName), zap.String("tableName", tbl.Meta().Name.String()))
204+
} else {
205+
colOrIdxName = getIdxNamesFromArgs(args)
206+
}
207+
case model.ActionModifyColumn:
208+
oldCol, _ := getOldAndNewColumnsForUpdateColumn(tbl, rInfo.currElement.ID)
209+
if oldCol != nil {
210+
colOrIdxName = oldCol.Name.String()
211+
}
212+
}
213+
197214
batchCnt := rInfo.ReorgMeta.GetBatchSizeOrDefault(int(variable.GetDDLReorgBatchSize()))
198215
return &backfillCtx{
199-
id: id,
200-
ddlCtx: rInfo.jobCtx.oldDDLCtx,
201-
warnings: warnHandler,
202-
exprCtx: exprCtx,
203-
tblCtx: tblCtx,
204-
loc: exprCtx.GetEvalCtx().Location(),
205-
schemaName: schemaName,
206-
table: tbl,
207-
batchCnt: batchCnt,
208-
jobContext: jobCtx,
209-
metricCounter: metrics.BackfillTotalCounter.WithLabelValues(
210-
metrics.GenerateReorgLabel(label, schemaName, tbl.Meta().Name.String())),
216+
id: id,
217+
ddlCtx: rInfo.jobCtx.oldDDLCtx,
218+
warnings: warnHandler,
219+
exprCtx: exprCtx,
220+
tblCtx: tblCtx,
221+
loc: exprCtx.GetEvalCtx().Location(),
222+
schemaName: schemaName,
223+
table: tbl,
224+
batchCnt: batchCnt,
225+
jobContext: jobCtx,
226+
metricCounter: metrics.GetBackfillTotalByLabel(label, schemaName, tbl.Meta().Name.String(), colOrIdxName),
211227
}, nil
212228
}
213229

230+
func getIdxNamesFromArgs(args *model.ModifyIndexArgs) string {
231+
var sb strings.Builder
232+
for i, idx := range args.IndexArgs {
233+
if i > 0 {
234+
sb.WriteString("+")
235+
}
236+
sb.WriteString(idx.IndexName.O)
237+
}
238+
return sb.String()
239+
}
240+
214241
func updateTxnEntrySizeLimitIfNeeded(txn kv.Transaction) {
215242
if entrySizeLimit := variable.TxnEntrySizeLimit.Load(); entrySizeLimit > 0 {
216243
txn.SetOption(kv.SizeLimits, kv.TxnSizeLimits{
@@ -686,6 +713,7 @@ func (dc *ddlCtx) runAddIndexInLocalIngestMode(
686713
idxCnt := len(reorgInfo.elements)
687714
indexIDs := make([]int64, 0, idxCnt)
688715
indexInfos := make([]*model.IndexInfo, 0, idxCnt)
716+
var indexNames strings.Builder
689717
uniques := make([]bool, 0, idxCnt)
690718
hasUnique := false
691719
for _, e := range reorgInfo.elements {
@@ -699,6 +727,10 @@ func (dc *ddlCtx) runAddIndexInLocalIngestMode(
699727
return errors.Errorf("index info not found: %d", e.ID)
700728
}
701729
indexInfos = append(indexInfos, indexInfo)
730+
if indexNames.Len() > 0 {
731+
indexNames.WriteString("+")
732+
}
733+
indexNames.WriteString(indexInfo.Name.O)
702734
uniques = append(uniques, indexInfo.Unique)
703735
hasUnique = hasUnique || indexInfo.Unique
704736
}
@@ -736,8 +768,7 @@ func (dc *ddlCtx) runAddIndexInLocalIngestMode(
736768
rowCntListener := &localRowCntListener{
737769
prevPhysicalRowCnt: reorgCtx.getRowCount(),
738770
reorgCtx: reorgCtx,
739-
counter: metrics.BackfillTotalCounter.WithLabelValues(
740-
metrics.GenerateReorgLabel("add_idx_rate", job.SchemaName, job.TableName)),
771+
counter: metrics.GetBackfillTotalByLabel(metrics.LblAddIdxRate, job.SchemaName, job.TableName, indexNames.String()),
741772
}
742773

743774
sctx, err := sessPool.Get()

‎pkg/ddl/backfilling_operators.go

Lines changed: 0 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -745,10 +745,6 @@ func NewIndexIngestOperator(
745745
writers = append(writers, writer)
746746
}
747747

748-
indexIDs := make([]int64, len(indexes))
749-
for i := 0; i < len(indexes); i++ {
750-
indexIDs[i] = indexes[i].Meta().ID
751-
}
752748
return &indexIngestLocalWorker{
753749
indexIngestBaseWorker: indexIngestBaseWorker{
754750
ctx: ctx,
@@ -762,7 +758,6 @@ func NewIndexIngestOperator(
762758
srcChunkPool: srcChunkPool,
763759
reorgMeta: reorgMeta,
764760
},
765-
indexIDs: indexIDs,
766761
backendCtx: backendCtx,
767762
rowCntListener: rowCntListener,
768763
cpMgr: cpMgr,
@@ -793,7 +788,6 @@ func (w *indexIngestExternalWorker) HandleTask(ck IndexRecordChunk, send func(In
793788

794789
type indexIngestLocalWorker struct {
795790
indexIngestBaseWorker
796-
indexIDs []int64
797791
backendCtx ingest.BackendCtx
798792
rowCntListener RowCountListener
799793
cpMgr *ingest.CheckpointManager

‎pkg/ddl/backfilling_read_index.go

Lines changed: 17 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@ import (
1818
"context"
1919
"encoding/hex"
2020
"encoding/json"
21+
"strings"
2122
"sync"
2223
"sync/atomic"
2324

@@ -224,9 +225,14 @@ func (r *readIndexExecutor) buildLocalStorePipeline(
224225
d := r.d
225226
indexIDs := make([]int64, 0, len(r.indexes))
226227
uniques := make([]bool, 0, len(r.indexes))
228+
var idxNames strings.Builder
227229
for _, index := range r.indexes {
228230
indexIDs = append(indexIDs, index.ID)
229231
uniques = append(uniques, index.Unique)
232+
if idxNames.Len() > 0 {
233+
idxNames.WriteByte('+')
234+
}
235+
idxNames.WriteString(index.Name.O)
230236
}
231237
engines, err := r.bc.Register(indexIDs, uniques, r.ptbl)
232238
if err != nil {
@@ -236,7 +242,7 @@ func (r *readIndexExecutor) buildLocalStorePipeline(
236242
zap.Int64s("index IDs", indexIDs))
237243
return nil, err
238244
}
239-
rowCntListener := newDistTaskRowCntListener(r.curRowCount, r.job.SchemaName, tbl.Meta().Name.O)
245+
rowCntListener := newDistTaskRowCntListener(r.curRowCount, r.job.SchemaName, tbl.Meta().Name.O, idxNames.String())
240246
return NewAddIndexIngestPipeline(
241247
opCtx,
242248
d.store,
@@ -280,7 +286,14 @@ func (r *readIndexExecutor) buildExternalStorePipeline(
280286
kvMeta.MergeSummary(summary)
281287
s.mu.Unlock()
282288
}
283-
rowCntListener := newDistTaskRowCntListener(r.curRowCount, r.job.SchemaName, tbl.Meta().Name.O)
289+
var idxNames strings.Builder
290+
for _, idx := range r.indexes {
291+
if idxNames.Len() > 0 {
292+
idxNames.WriteByte('+')
293+
}
294+
idxNames.WriteString(idx.Name.O)
295+
}
296+
rowCntListener := newDistTaskRowCntListener(r.curRowCount, r.job.SchemaName, tbl.Meta().Name.O, idxNames.String())
284297
return NewWriteIndexToExternalStoragePipeline(
285298
opCtx,
286299
d.store,
@@ -307,9 +320,8 @@ type distTaskRowCntListener struct {
307320
counter prometheus.Counter
308321
}
309322

310-
func newDistTaskRowCntListener(totalRowCnt *atomic.Int64, dbName, tblName string) *distTaskRowCntListener {
311-
counter := metrics.BackfillTotalCounter.WithLabelValues(
312-
metrics.GenerateReorgLabel("add_idx_rate", dbName, tblName))
323+
func newDistTaskRowCntListener(totalRowCnt *atomic.Int64, dbName, tblName, idxName string) *distTaskRowCntListener {
324+
counter := metrics.GetBackfillTotalByLabel(metrics.LblAddIdxRate, dbName, tblName, idxName)
313325
return &distTaskRowCntListener{
314326
totalRowCount: totalRowCnt,
315327
counter: counter,

‎pkg/ddl/backfilling_scheduler.go

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,7 @@ import (
2828
"github.com/pingcap/tidb/pkg/errctx"
2929
"github.com/pingcap/tidb/pkg/kv"
3030
"github.com/pingcap/tidb/pkg/meta/model"
31+
"github.com/pingcap/tidb/pkg/metrics"
3132
"github.com/pingcap/tidb/pkg/resourcegroup"
3233
"github.com/pingcap/tidb/pkg/sessionctx"
3334
"github.com/pingcap/tidb/pkg/sessionctx/stmtctx"
@@ -271,7 +272,7 @@ func (b *txnBackfillScheduler) adjustWorkerSize() error {
271272
)
272273
switch b.tp {
273274
case typeAddIndexWorker:
274-
backfillCtx, err := newBackfillCtx(i, reorgInfo, job.SchemaName, b.tbl, jc, "add_idx_rate", false, false)
275+
backfillCtx, err := newBackfillCtx(i, reorgInfo, job.SchemaName, b.tbl, jc, metrics.LblAddIdxRate, false, false)
275276
if err != nil {
276277
return err
277278
}
@@ -284,7 +285,7 @@ func (b *txnBackfillScheduler) adjustWorkerSize() error {
284285
runner = newBackfillWorker(b.ctx, idxWorker)
285286
worker = idxWorker
286287
case typeAddIndexMergeTmpWorker:
287-
backfillCtx, err := newBackfillCtx(i, reorgInfo, job.SchemaName, b.tbl, jc, "merge_tmp_idx_rate", false, false)
288+
backfillCtx, err := newBackfillCtx(i, reorgInfo, job.SchemaName, b.tbl, jc, metrics.LblMergeTmpIdxRate, false, false)
288289
if err != nil {
289290
return err
290291
}

‎pkg/ddl/column.go

Lines changed: 17 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -34,6 +34,7 @@ import (
3434
"github.com/pingcap/tidb/pkg/meta"
3535
"github.com/pingcap/tidb/pkg/meta/autoid"
3636
"github.com/pingcap/tidb/pkg/meta/model"
37+
"github.com/pingcap/tidb/pkg/metrics"
3738
"github.com/pingcap/tidb/pkg/parser/ast"
3839
pmodel "github.com/pingcap/tidb/pkg/parser/model"
3940
"github.com/pingcap/tidb/pkg/parser/mysql"
@@ -609,8 +610,22 @@ type updateColumnWorker struct {
609610
checksumNeeded bool
610611
}
611612

613+
func getOldAndNewColumnsForUpdateColumn(t table.Table, currElementID int64) (oldCol, newCol *model.ColumnInfo) {
614+
for _, col := range t.WritableCols() {
615+
if col.ID == currElementID {
616+
changeColumnOrigName := table.FindCol(t.Cols(), getChangingColumnOriginName(col.ColumnInfo))
617+
if changeColumnOrigName != nil {
618+
newCol = col.ColumnInfo
619+
oldCol = changeColumnOrigName.ColumnInfo
620+
return
621+
}
622+
}
623+
}
624+
return
625+
}
626+
612627
func newUpdateColumnWorker(id int, t table.PhysicalTable, decodeColMap map[int64]decoder.Column, reorgInfo *reorgInfo, jc *ReorgContext) (*updateColumnWorker, error) {
613-
bCtx, err := newBackfillCtx(id, reorgInfo, reorgInfo.SchemaName, t, jc, "update_col_rate", false, true)
628+
bCtx, err := newBackfillCtx(id, reorgInfo, reorgInfo.SchemaName, t, jc, metrics.LblUpdateColRate, false, true)
614629
if err != nil {
615630
return nil, err
616631
}
@@ -620,14 +635,7 @@ func newUpdateColumnWorker(id int, t table.PhysicalTable, decodeColMap map[int64
620635
zap.Stringer("reorgInfo", reorgInfo))
621636
return nil, nil
622637
}
623-
var oldCol, newCol *model.ColumnInfo
624-
for _, col := range t.WritableCols() {
625-
if col.ID == reorgInfo.currElement.ID {
626-
newCol = col.ColumnInfo
627-
oldCol = table.FindCol(t.Cols(), getChangingColumnOriginName(newCol)).ColumnInfo
628-
break
629-
}
630-
}
638+
oldCol, newCol := getOldAndNewColumnsForUpdateColumn(t, reorgInfo.currElement.ID)
631639
rowDecoder := decoder.NewRowDecoder(t, t.WritableCols(), decodeColMap)
632640
failpoint.Inject("forceRowLevelChecksumOnUpdateColumnBackfill", func() {
633641
orig := variable.EnableRowLevelChecksum.Load()

‎pkg/ddl/index.go

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2921,7 +2921,7 @@ type cleanUpIndexWorker struct {
29212921
}
29222922

29232923
func newCleanUpIndexWorker(id int, t table.PhysicalTable, decodeColMap map[int64]decoder.Column, reorgInfo *reorgInfo, jc *ReorgContext) (*cleanUpIndexWorker, error) {
2924-
bCtx, err := newBackfillCtx(id, reorgInfo, reorgInfo.SchemaName, t, jc, "cleanup_idx_rate", false, false)
2924+
bCtx, err := newBackfillCtx(id, reorgInfo, reorgInfo.SchemaName, t, jc, metrics.LblCleanupIdxRate, false, false)
29252925
if err != nil {
29262926
return nil, err
29272927
}

‎pkg/ddl/modify_column.go

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -502,7 +502,7 @@ func (w *worker) doModifyColumnTypeWithData(
502502
// Make sure job args change after `updateVersionAndTableInfoWithCheck`, otherwise, the job args will
503503
// be updated in `updateDDLJob` even if it meets an error in `updateVersionAndTableInfoWithCheck`.
504504
job.SchemaState = model.StateDeleteOnly
505-
metrics.GetBackfillProgressByLabel(metrics.LblModifyColumn, job.SchemaName, tblInfo.Name.String()).Set(0)
505+
metrics.GetBackfillProgressByLabel(metrics.LblModifyColumn, job.SchemaName, tblInfo.Name.String(), args.OldColumnName.O).Set(0)
506506
args.ChangingColumn = changingCol
507507
args.ChangingIdxs = changingIdxs
508508
failpoint.InjectCall("modifyColumnTypeWithData", job, args)

‎pkg/ddl/partition.go

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -3334,7 +3334,7 @@ func (w *worker) onReorganizePartition(jobCtx *jobContext, job *model.Job) (ver
33343334
}
33353335

33363336
// Assume we cannot have more than MaxUint64 rows, set the progress to 1/10 of that.
3337-
metrics.GetBackfillProgressByLabel(metrics.LblReorgPartition, job.SchemaName, tblInfo.Name.String()).Set(0.1 / float64(math.MaxUint64))
3337+
metrics.GetBackfillProgressByLabel(metrics.LblReorgPartition, job.SchemaName, tblInfo.Name.String(), "").Set(0.1 / float64(math.MaxUint64))
33383338
job.SchemaState = model.StateDeleteOnly
33393339
tblInfo.Partition.DDLState = job.SchemaState
33403340
ver, err = updateVersionAndTableInfoWithCheck(jobCtx, job, tblInfo, true)
@@ -3398,7 +3398,7 @@ func (w *worker) onReorganizePartition(jobCtx *jobContext, job *model.Job) (ver
33983398
}
33993399
}
34003400
tblInfo.Partition.DDLState = model.StateWriteOnly
3401-
metrics.GetBackfillProgressByLabel(metrics.LblReorgPartition, job.SchemaName, tblInfo.Name.String()).Set(0.2 / float64(math.MaxUint64))
3401+
metrics.GetBackfillProgressByLabel(metrics.LblReorgPartition, job.SchemaName, tblInfo.Name.String(), "").Set(0.2 / float64(math.MaxUint64))
34023402
failpoint.Inject("reorgPartRollback2", func(val failpoint.Value) {
34033403
if val.(bool) {
34043404
err = errors.New("Injected error by reorgPartRollback2")
@@ -3419,7 +3419,7 @@ func (w *worker) onReorganizePartition(jobCtx *jobContext, job *model.Job) (ver
34193419
}
34203420
job.SchemaState = model.StateWriteReorganization
34213421
tblInfo.Partition.DDLState = job.SchemaState
3422-
metrics.GetBackfillProgressByLabel(metrics.LblReorgPartition, job.SchemaName, tblInfo.Name.String()).Set(0.3 / float64(math.MaxUint64))
3422+
metrics.GetBackfillProgressByLabel(metrics.LblReorgPartition, job.SchemaName, tblInfo.Name.String(), "").Set(0.3 / float64(math.MaxUint64))
34233423
ver, err = updateVersionAndTableInfo(jobCtx, job, tblInfo, true)
34243424
case model.StateWriteReorganization:
34253425
physicalTableIDs := getPartitionIDsFromDefinitions(tblInfo.Partition.DroppingDefinitions)
@@ -3813,7 +3813,7 @@ type reorgPartitionWorker struct {
38133813
}
38143814

38153815
func newReorgPartitionWorker(i int, t table.PhysicalTable, decodeColMap map[int64]decoder.Column, reorgInfo *reorgInfo, jc *ReorgContext) (*reorgPartitionWorker, error) {
3816-
bCtx, err := newBackfillCtx(i, reorgInfo, reorgInfo.SchemaName, t, jc, "reorg_partition_rate", false, false)
3816+
bCtx, err := newBackfillCtx(i, reorgInfo, reorgInfo.SchemaName, t, jc, metrics.LblReorgPartitionRate, false, false)
38173817
if err != nil {
38183818
return nil, err
38193819
}

‎pkg/ddl/reorg.go

Lines changed: 17 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -511,12 +511,26 @@ func updateBackfillProgress(w *worker, reorgInfo *reorgInfo, tblInfo *model.Tabl
511511
} else {
512512
label = metrics.LblAddIndex
513513
}
514-
metrics.GetBackfillProgressByLabel(label, reorgInfo.SchemaName, tblInfo.Name.String()).Set(progress * 100)
514+
idxNames := ""
515+
args, err := model.GetModifyIndexArgs(reorgInfo.Job)
516+
if err != nil {
517+
logutil.DDLLogger().Error("Fail to get ModifyIndexArgs", zap.Error(err))
518+
} else {
519+
idxNames = getIdxNamesFromArgs(args)
520+
}
521+
metrics.GetBackfillProgressByLabel(label, reorgInfo.SchemaName, tblInfo.Name.String(), idxNames).Set(progress * 100)
515522
case model.ActionModifyColumn:
516-
metrics.GetBackfillProgressByLabel(metrics.LblModifyColumn, reorgInfo.SchemaName, tblInfo.Name.String()).Set(progress * 100)
523+
colName := ""
524+
args, err := model.GetModifyColumnArgs(reorgInfo.Job)
525+
if err != nil {
526+
logutil.DDLLogger().Error("Fail to get ModifyColumnArgs", zap.Error(err))
527+
} else {
528+
colName = args.OldColumnName.O
529+
}
530+
metrics.GetBackfillProgressByLabel(metrics.LblModifyColumn, reorgInfo.SchemaName, tblInfo.Name.String(), colName).Set(progress * 100)
517531
case model.ActionReorganizePartition, model.ActionRemovePartitioning,
518532
model.ActionAlterTablePartitioning:
519-
metrics.GetBackfillProgressByLabel(metrics.LblReorgPartition, reorgInfo.SchemaName, tblInfo.Name.String()).Set(progress * 100)
533+
metrics.GetBackfillProgressByLabel(metrics.LblReorgPartition, reorgInfo.SchemaName, tblInfo.Name.String(), "").Set(progress * 100)
520534
}
521535
}
522536

0 commit comments

Comments
 (0)