Skip to content

ddl: prevent partial ADD INDEX plans after transient TSO errors #69834

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

Open
D3Hunter wants to merge 2 commits into
base: master
Choose a base branch
from
+82 −22
Open

ddl: prevent partial ADD INDEX plans after transient TSO errors #69834

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
baf3ac1
ddl: prevent partial add-index plans after transient retries
D3Hunter Jul 14, 2026
81ad8c2
ddl: retry merge-temp planning on region gaps
D3Hunter Jul 14, 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 .agents/skills/tidb-test-guidelines/references/ddl-case-map.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,7 @@
### Tests
- `pkg/ddl/affinity_test.go` - ddl: Tests affinity build group definitions table.
- `pkg/ddl/attributes_sql_test.go` - ddl: Tests alter table partition attributes.
- `pkg/ddl/backfilling_dist_scheduler_test.go` - ddl: Tests backfilling scheduler local mode.
- `pkg/ddl/backfilling_dist_scheduler_test.go` - ddl: Tests backfilling scheduler local mode and backfilling plan retries.
- `pkg/ddl/backfilling_test.go` - ddl: Tests done task keeper.
- `pkg/ddl/backfilling_txn_executor_test.go` - ddl: Tests expected ingest worker count.
- `pkg/ddl/bdr_test.go` - ddl: Tests denied by BDR when add column.
Expand Down
35 changes: 26 additions & 9 deletions pkg/ddl/backfilling_dist_scheduler.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -352,7 +352,7 @@ func generatePlanForPhysicalTable(
return nil, errors.Trace(err)
}

subTaskMetas := make([][]byte, 0, 4)
var subTaskMetas [][]byte
backoffer := backoff.NewExponential(scanRegionBackoffBase, 2, scanRegionBackoffMax)
err = handle.RunWithRetry(ctx, 8, backoffer, logutil.DDLLogger(), func(_ context.Context) (bool, error) {
regionCache := store.(helper.Storage).GetRegionCache()
Expand All @@ -364,7 +364,8 @@ func generatePlanForPhysicalTable(
return bytes.Compare(recordRegionMetas[i].StartKey(), recordRegionMetas[j].StartKey()) < 0
})

// Check if regions are continuous.
// LoadRegionsInKeyRange can combine multiple PD scans. A concurrent region
// split or merge can make those scans discontinuous, so retry the full scan.
shouldRetry := false
cur := recordRegionMetas[0]
for _, m := range recordRegionMetas[1:] {
Expand All @@ -374,11 +375,15 @@ func generatePlanForPhysicalTable(
}
cur = m
}
failpoint.Inject("mockPhysicalTableRegionDiscontinuity", func() {
shouldRetry = true
})

if shouldRetry {
return true, nil
return true, errors.New("regions are not continuous")
Comment thread
D3Hunter marked this conversation as resolved.
}

attemptMetas := make([][]byte, 0, 4)
regionBatch := CalculateRegionBatch(len(recordRegionMetas), nodeCnt, !useCloud)
logger.Info("calculate region batch",
zap.Int("totalRegionCnt", len(recordRegionMetas)),
Expand All @@ -391,7 +396,7 @@ func generatePlanForPhysicalTable(
// It should be different for each subtask to determine if there are duplicate entries.
importTS, err := allocNewTS(ctx, store.(kv.StorageWithPD))
if err != nil {
return true, nil
return true, err
}
end := min(i+regionBatch, len(recordRegionMetas))
batch := recordRegionMetas[i:end]
Expand All @@ -411,8 +416,9 @@ func generatePlanForPhysicalTable(
if err != nil {
return false, err
}
subTaskMetas = append(subTaskMetas, metaBytes)
attemptMetas = append(attemptMetas, metaBytes)
}
subTaskMetas = attemptMetas

@ingress-bot ingress-bot Jul 14, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

ℹ️ [Info] No invariant comment guards the per-attempt attemptMetas buffer this PR's fix depends on

The whole point of this diff (per the commit title, prevent partial add-index plans after transient retries) is that subTaskMetas must only be replaced atomically from a fresh attemptMetas slice on a fully successful attempt, never appended to directly across retries.
That invariant is encoded purely by variable naming/placement (attemptMetas local to the retry closure, subTaskMetas in the outer scope) with no comment marking it, so a future edit that reintroduces direct appends to subTaskMetas inside the retry closure (the exact shape of the bug this PR fixes) would compile cleanly and pass review without an obvious signal.

return false, nil
})
if err != nil {
Expand Down Expand Up @@ -537,6 +543,11 @@ func generateGlobalSortIngestPlan(
}

func allocNewTS(ctx context.Context, store kv.StorageWithPD) (uint64, error) {
failpoint.Inject("mockAllocNewTSError", func(val failpoint.Value) {
if val.(bool) {
failpoint.Return(0, errors.New("mock alloc new TS error"))
}
})
pdCli := store.GetPDClient()
p, l, err := pdCli.GetTS(ctx)
if err != nil {
Expand Down Expand Up @@ -890,7 +901,7 @@ func genMergeTempPlanForOneIndex(
pid := tbl.GetPhysicalID()
start, end := encodeTempIndexRange(pid, idxInfo.ID, idxInfo.ID)

subTaskMetas := make([][]byte, 0, 4)
var subTaskMetas [][]byte
backoffer := backoff.NewExponential(scanRegionBackoffBase, 2, scanRegionBackoffMax)
err := handle.RunWithRetry(ctx, 8, backoffer, logutil.DDLLogger(), func(_ context.Context) (bool, error) {
regionCache := store.(helper.Storage).GetRegionCache()
Expand All @@ -902,7 +913,8 @@ func genMergeTempPlanForOneIndex(
return bytes.Compare(regionMetas[i].StartKey(), regionMetas[j].StartKey()) < 0
})

// Check if regions are continuous.
// LoadRegionsInKeyRange can combine multiple PD scans. A concurrent region
// split or merge can make those scans discontinuous, so retry the full scan.
shouldRetry := false
cur := regionMetas[0]
for _, m := range regionMetas[1:] {
Expand All @@ -912,11 +924,15 @@ func genMergeTempPlanForOneIndex(
}
cur = m
}
failpoint.Inject("mockMergeTempIndexRegionDiscontinuity", func() {
shouldRetry = true
})

if shouldRetry {
return true, nil
return true, errors.New("regions are not continuous")
}

attemptMetas := make([][]byte, 0, 4)
regionBatch := calculateTempIndexRegionBatch(len(regionMetas), nodeCnt)
logger.Info("calculate temp index region batch",
zap.Int64("physicalTableID", pid),
Expand Down Expand Up @@ -945,8 +961,9 @@ func genMergeTempPlanForOneIndex(
if err != nil {
return false, err
}
subTaskMetas = append(subTaskMetas, metaBytes)
attemptMetas = append(attemptMetas, metaBytes)
}
subTaskMetas = attemptMetas
return false, nil
})
if err != nil {
Expand Down
67 changes: 55 additions & 12 deletions pkg/ddl/backfilling_dist_scheduler_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,6 @@ import (

"github.com/fsouza/fake-gcs-server/fakestorage"
"github.com/ngaut/pools"
"github.com/pingcap/failpoint"
"github.com/pingcap/tidb/pkg/config/kerneltype"
"github.com/pingcap/tidb/pkg/ddl"
"github.com/pingcap/tidb/pkg/domain"
Expand All @@ -40,6 +39,7 @@ import (
"github.com/pingcap/tidb/pkg/parser/ast"
"github.com/pingcap/tidb/pkg/parser/mysql"
"github.com/pingcap/tidb/pkg/testkit"
"github.com/pingcap/tidb/pkg/testkit/testfailpoint"
tidbutil "github.com/pingcap/tidb/pkg/util"
"github.com/stretchr/testify/require"
"github.com/tikv/client-go/v2/util"
Expand Down Expand Up @@ -70,8 +70,7 @@ func TestBackfillingSchedulerLocalMode(t *testing.T) {
tk := testkit.NewTestKit(t, store)
tk.MustExec("use test")

require.NoError(t, failpoint.Enable("github.com/pingcap/tidb/pkg/ddl/mockWriterMemSizeInKB", "return(1048576)"))
defer failpoint.Disable("github.com/pingcap/tidb/pkg/ddl/mockWriterMemSizeInKB")
testfailpoint.Enable(t, "github.com/pingcap/tidb/pkg/ddl/mockWriterMemSizeInKB", "return(1048576)")
/// 1. test partition table.
tk.MustExec("create table tp1(id int primary key, v int, key idx (id)) PARTITION BY RANGE (id) (\n " +
"PARTITION p0 VALUES LESS THAN (10),\n" +
Expand Down Expand Up @@ -133,7 +132,57 @@ func TestBackfillingSchedulerLocalMode(t *testing.T) {
require.NoError(t, err)
require.Equal(t, 1, len(metas))
require.Equal(t, proto.BackfillStepReadIndex, task.Step)
// 2.2.2 BackfillStepReadIndex

// 2.2.2 transient region discontinuity should be retried.
testfailpoint.Enable(t, "github.com/pingcap/tidb/pkg/ddl/mockPhysicalTableRegionDiscontinuity", "1*return()")
metas, err = sch.OnNextSubtasksBatch(ctx, nil, task, execIDs, task.Step)
require.NoError(t, err)
require.Len(t, metas, 1)

// 2.2.3 a retry after allocating some subtask timestamps should not return a partial or duplicate plan.
tk.MustExec("create table t3(id bigint primary key)")
tk.MustExec("insert into t3 values (1), (9999)")
tk.MustQuery("split table t3 by (5000)").Check(testkit.Rows("1 1"))
multiRegionTask, server := createAddIndexTask(t, dom, "test", "t3", proto.Backfill, false)
require.Nil(t, server)
multiRegionTask.Step = sch.GetNextStep(&multiRegionTask.TaskBase)
testfailpoint.Enable(t, "github.com/pingcap/tidb/pkg/ddl/mockRegionBatch", "return(1)")
expectedMetas, err := sch.OnNextSubtasksBatch(ctx, nil, multiRegionTask, execIDs, multiRegionTask.Step)
require.NoError(t, err)
require.Len(t, expectedMetas, 2)

testfailpoint.Enable(t, "github.com/pingcap/tidb/pkg/ddl/mockAllocNewTSError", "1*return(false)->1*return(true)")
metas, err = sch.OnNextSubtasksBatch(ctx, nil, multiRegionTask, execIDs, multiRegionTask.Step)
require.NoError(t, err)
require.Len(t, metas, len(expectedMetas))
for i := range expectedMetas {
var expected, actual ddl.BackfillSubTaskMeta
require.NoError(t, json.Unmarshal(expectedMetas[i], &expected))
require.NoError(t, json.Unmarshal(metas[i], &actual))
require.Equal(t, expected.PhysicalTableID, actual.PhysicalTableID)
require.Equal(t, expected.RowStart, actual.RowStart)
require.Equal(t, expected.RowEnd, actual.RowEnd)
}

// 2.2.4 transient region discontinuity should also be retried when planning the merge-temp-index step.
tk.MustExec("create table t4(id bigint primary key, v bigint, key idx(v))")
mergeTask, server := createAddIndexTask(t, dom, "test", "t4", proto.Backfill, false)
require.Nil(t, server)
mergeTable, err := dom.InfoSchema().TableByName(context.Background(), ast.NewCIStr("test"), ast.NewCIStr("t4"))
require.NoError(t, err)
var mergeTaskMeta ddl.BackfillTaskMeta
require.NoError(t, json.Unmarshal(mergeTask.Meta, &mergeTaskMeta))
mergeTaskMeta.EleIDs = []int64{mergeTable.Meta().Indices[0].ID}
mergeTaskMeta.MergeTempIndex = true
mergeTask.Meta, err = json.Marshal(&mergeTaskMeta)
require.NoError(t, err)
mergeTask.Step = proto.BackfillStepMergeTempIndex
testfailpoint.Enable(t, "github.com/pingcap/tidb/pkg/ddl/mockMergeTempIndexRegionDiscontinuity", "1*return()")
metas, err = sch.OnNextSubtasksBatch(ctx, nil, mergeTask, execIDs, mergeTask.Step)
require.NoError(t, err)
require.Len(t, metas, 1)

// 2.2.5 BackfillStepReadIndex
task.State = proto.TaskStateRunning
task.Step = sch.GetNextStep(&task.TaskBase)
require.Equal(t, proto.StepDone, task.Step)
Expand Down Expand Up @@ -244,10 +293,7 @@ func TestBackfillingSchedulerGlobalSortMode(t *testing.T) {
require.NoError(t, mgr.FinishSubtask(ctx, s.ExecID, s.ID, sortStepMetaBytes))
}
// 2. to merge-sort stage.
require.NoError(t, failpoint.Enable("github.com/pingcap/tidb/pkg/ddl/forceMergeSort", `return()`))
t.Cleanup(func() {
require.NoError(t, failpoint.Disable("github.com/pingcap/tidb/pkg/ddl/forceMergeSort"))
})
testfailpoint.Enable(t, "github.com/pingcap/tidb/pkg/ddl/forceMergeSort", `return()`)
subtaskMetas, err = ext.OnNextSubtasksBatch(ctx, sch, task, execIDs, ext.GetNextStep(&task.TaskBase))
require.NoError(t, err)
require.Len(t, subtaskMetas, 1)
Expand Down Expand Up @@ -284,10 +330,7 @@ func TestBackfillingSchedulerGlobalSortMode(t *testing.T) {
require.NoError(t, mgr.FinishSubtask(ctx, s.ExecID, s.ID, mergeSortStepMetaBytes))
}
// 3. to write&ingest stage.
require.NoError(t, failpoint.Enable("github.com/pingcap/tidb/pkg/ddl/mockWriteIngest", "return(true)"))
t.Cleanup(func() {
require.NoError(t, failpoint.Disable("github.com/pingcap/tidb/pkg/ddl/mockWriteIngest"))
})
testfailpoint.Enable(t, "github.com/pingcap/tidb/pkg/ddl/mockWriteIngest", "return(true)")
subtaskMetas, err = ext.OnNextSubtasksBatch(ctx, sch, task, execIDs, ext.GetNextStep(&task.TaskBase))
require.NoError(t, err)
require.Len(t, subtaskMetas, 1)
Expand Down
Loading