Release 6.5.11 hotfix 20260410 - #67726
Closed
HunDunDM wants to merge 12 commits into
Closed
Conversation
ti-chi-bot
Bot
added
do-not-merge/invalid-title
release-note-none
|
[APPROVALNOTIFIER] This PR is NOT APPROVED This pull-request has been approved by: The full list of commands accepted by this bot can be found here. DetailsNeeds approval from an approver in each of these files:Approvers can indicate their approval by writing |
|
Hi @HunDunDM. Thanks for your PR. PRs from untrusted users cannot be marked as trusted with I understand the commands that are listed here. DetailsInstructions for interacting with me using PR comments are available here. If you have questions or suggestions related to my behavior, please file an issue against the kubernetes-sigs/prow repository. |
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
📝 WalkthroughWalkthroughAdds PD force-merge reporting and a background merge-empty-regions scanner with checkpointing; integrates force-merge reporting into delete-range/GC flows; refactors DDL running-job tracking and owner-retire hooks; fixes EXCHANGE PARTITION _tidb_rowid behavior; exposes an HTTP reset API; introduces sysvar and multiple tests. Changes
Sequence Diagram(s)sequenceDiagram
rect rgba(200,200,255,0.5)
participant GCWorker as GC Worker
participant DeleteRange as Delete-Range Emulator
participant DDL as DDL (range computation)
participant InfoSync as InfoSync (PD client)
participant PD as PD Server
end
GCWorker->>DeleteRange: start deleteRanges(ctx)
DeleteRange->>DeleteRange: doDelRangeWork(ctx)
DeleteRange->>DDL: GetForceMergeRangesForGCDeleteRange(historyJob, task, cache)
DDL-->>DeleteRange: []ForceMergeKeyRange
DeleteRange->>InfoSync: AddForceMergeRanges(ctx, ranges)
InfoSync->>PD: POST /regions/force-merge (batched)
PD-->>InfoSync: response
InfoSync->>InfoSync: sleep between batches (cancellable)
sequenceDiagram
rect rgba(200,255,200,0.5)
participant Ticker as Ticker
participant Domain as Domain
participant Meta as Meta Store
participant DDL as DDL (merge-empty-regions)
participant InfoSync as InfoSync
participant PD as PD Server
end
Ticker->>Domain: tick -> doMergeEmptyRegions()
Domain->>Meta: loadOrInitMergeEmptyRegionsMinTableID()
Meta-->>Domain: minTableID
Domain->>DDL: GetMergeEmptyRegionsKeyRanges(infoschema, minTableID)
DDL-->>Domain: (maxTableID, ranges)
alt ranges non-empty
Domain->>InfoSync: AddForceMergeRanges(ctx, ranges)
InfoSync->>PD: POST /regions/force-merge
PD-->>InfoSync: success
end
Domain->>Meta: storeMergeEmptyRegionsMinTableIDIfUnchanged(expected, next)
Meta-->>Domain: updated?
Estimated code review effort🎯 4 (Complex) | ⏱️ ~75 minutes Possibly related PRs
Suggested labels
Suggested reviewers
Poem
🚥 Pre-merge checks | ❌ 3❌ Failed checks (2 warnings, 1 inconclusive)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 5
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
store/gcworker/gc_worker.go (1)
723-739:⚠️ Potential issue | 🟠 MajorReport force-merge ranges before completing the delete-range task.
After Line 723 succeeds, this record is no longer retried by
deleteRanges(). IfdoGCForceMergeRanges()then fails, the new PD notification is only logged andredoDeleteRanges()never replays it, so a transient error can permanently drop the force-merge request for that table.Suggested fix
- err = util.CompleteDeleteRange(se, r) - if err != nil { - logutil.Logger(ctx).Error("[gc worker] failed to mark delete range task done", - zap.String("uuid", w.uuid), - zap.Stringer("startKey", startKey), - zap.Stringer("endKey", endKey), - zap.Error(err)) - metrics.GCUnsafeDestroyRangeFailuresCounterVec.WithLabelValues("save").Inc() - } - if err := w.doGCForceMergeRanges(ctx, se, r, gcForceMergeTableCache); err != nil { logutil.Logger(ctx).Error("[gc worker] report force merge ranges failed on range", zap.String("uuid", w.uuid), zap.Int64("jobID", r.JobID), zap.Int64("elementID", r.ElementID), zap.Error(err)) + continue + } + + err = util.CompleteDeleteRange(se, r) + if err != nil { + logutil.Logger(ctx).Error("[gc worker] failed to mark delete range task done", + zap.String("uuid", w.uuid), + zap.Stringer("startKey", startKey), + zap.Stringer("endKey", endKey), + zap.Error(err)) + metrics.GCUnsafeDestroyRangeFailuresCounterVec.WithLabelValues("save").Inc() }As per coding guidelines "Keep error handling actionable and contextual; avoid silently swallowing errors."
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@store/gcworker/gc_worker.go` around lines 723 - 739, The current flow calls util.CompleteDeleteRange(se, r) before invoking w.doGCForceMergeRanges(ctx, se, r, gcForceMergeTableCache), which means a failure in doGCForceMergeRanges will be lost because the delete-range task is already marked done; move the call to w.doGCForceMergeRanges so it executes before util.CompleteDeleteRange, and if doGCForceMergeRanges returns an error, log/metric it and return that error (do not call CompleteDeleteRange) so deleteRanges()/redoDeleteRanges() can retry the task; update the surrounding error handling to preserve context (include w.uuid, r.JobID, r.ElementID) and increment appropriate failure metrics when retryable.
🧹 Nitpick comments (1)
testkit/result.go (1)
148-148: Add a doc comment for exportedStringmethod.Please add a short Go doc comment above
String()to satisfy exported-symbol documentation requirements.As per coding guidelines "Keep exported-symbol doc comments, and prefer semantic constraints over name restatement."
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@testkit/result.go` at line 148, Add a concise Go doc comment above the exported method Result.String that describes what the method returns and its semantics (e.g., the human-readable or formatted representation of the Result), avoiding a restatement of the method name; place the comment immediately above "func (res *Result) String() string" and follow standard Go doc style (Sentence starting with "String" is acceptable but prefer describing the content/format returned).
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@ddl/ddl.go`:
- Around line 772-774: The retire-owner hook currently replaces the shared
pointer d.runningJobs (registered via d.ownerManager.SetRetireOwnerHook), which
races with workers and wait4Switch() that may still reference the old instance;
instead implement and call a reset method on the runningJobs type (e.g.,
RunningJobs.Reset or Clear) that acquires its internal lock and zeroes/clears
its internal maps/slices in-place, then change the hook to call that reset
method rather than assigning newRunningJobs(); also ensure the hook registration
happens early enough relative to Start()/EnableDDL() so an early owner
retirement cannot miss the reset.
In `@domain/infosync/info.go`:
- Around line 425-433: doRequest currently sets bodyFactory to return the same
io.Reader instance on each retry, which reuses a consumed reader; change
doRequest so it buffers the original body (e.g., read entire body into a []byte
once) and make bodyFactory return a fresh io.Reader (like bytes.NewReader(buf))
for each invocation before calling doRequestWithBodyFactory, referencing the
doRequest function and the bodyFactory variable to implement the buffering and
fresh-reader creation.
In `@executor/update.go`:
- Around line 85-93: The current fix toggles skipMultipleChangesOnSameRow but
leaves merge() and exec() logic partition-unaware, causing mergedRowData and
updatedRowKeys to coalesce different physical rows when _tidb_rowid collides;
update the code so that when handling partitioned non-clustered tables (where
skipMultipleChangesOnSameRow is set false) you either 1) make mergedRowData and
updatedRowKeys keys partition-aware (e.g., include partition ID + handle)
wherever mergedRowData[TblID].Get(handle) and updatedRowKeys[TblID] are used so
distinct physical rows do not collapse, or 2) on detecting a collision (where
e.changed[i] is set false and multiUpdateOnSameTable[TblID] is true) explicitly
skip both merge() and the exec()/matching-count path (avoid incrementing
e.matches and the Rows matched counter) for that conflicting row; implement one
of these two fixes touching the logic around skipMultipleChangesOnSameRow,
merge(), mergedRowData, updatedRowKeys, e.changed and e.matches so partition
collisions are handled correctly.
In `@owner/manager.go`:
- Around line 62-63: Clear the manager's election state (m.elec) before invoking
the retire hook so that any call to IsOwner() within the hook returns false;
specifically, in the retire sequence in owner/manager.go update the order so
m.elec is set to nil/cleared prior to calling the function provided to
SetRetireOwnerHook, and adjust the surrounding comment to document the invariant
("hook runs after m.elec cleared so IsOwner() == false"); ensure the same change
is applied to the other retire-related invocation sites referenced around the
current retire logic.
In `@owner/mock.go`:
- Around line 105-108: The mockManager stores a retire callback via
SetRetireOwnerHook but RetireOwner() currently ignores it; update
mockManager.RetireOwner to check m.retireHook and invoke it (e.g., if
m.retireHook != nil { m.retireHook() }) so the registered cleanup runs under the
mock; ensure you reference the mockManager.retireHook field and call it from the
RetireOwner method, preserving any existing return behavior.
---
Outside diff comments:
In `@store/gcworker/gc_worker.go`:
- Around line 723-739: The current flow calls util.CompleteDeleteRange(se, r)
before invoking w.doGCForceMergeRanges(ctx, se, r, gcForceMergeTableCache),
which means a failure in doGCForceMergeRanges will be lost because the
delete-range task is already marked done; move the call to
w.doGCForceMergeRanges so it executes before util.CompleteDeleteRange, and if
doGCForceMergeRanges returns an error, log/metric it and return that error (do
not call CompleteDeleteRange) so deleteRanges()/redoDeleteRanges() can retry the
task; update the surrounding error handling to preserve context (include w.uuid,
r.JobID, r.ElementID) and increment appropriate failure metrics when retryable.
---
Nitpick comments:
In `@testkit/result.go`:
- Line 148: Add a concise Go doc comment above the exported method Result.String
that describes what the method returns and its semantics (e.g., the
human-readable or formatted representation of the Result), avoiding a
restatement of the method name; place the comment immediately above "func (res
*Result) String() string" and follow standard Go doc style (Sentence starting
with "String" is acceptable but prefer describing the content/format returned).
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro
Run ID: 0ee86e6a-6b85-4f64-af00-2fbc6e22ac4c
📒 Files selected for processing (31)
ddl/BUILD.bazelddl/db_partition_test.goddl/ddl.goddl/ddl_running_jobs.goddl/ddl_running_jobs_test.goddl/delete_range.goddl/pkdb_force_merge.goddl/pkdb_force_merge_test.godomain/BUILD.bazeldomain/domain.godomain/infosync/BUILD.bazeldomain/infosync/info.godomain/infosync/pkdb_force_merge.godomain/infosync/pkdb_force_merge_test.godomain/pkdb_force_merge.godomain/pkdb_force_merge_test.goexecutor/update.goinfoschema/builder.gometa/meta.gometa/meta_test.goowner/manager.goowner/mock.goserver/http_handler.goserver/http_handler_test.goserver/http_status.gosessionctx/variable/BUILD.bazelsessionctx/variable/pkdb_force_merge_test.gosessionctx/variable/sysvar.gosessionctx/variable/tidb_vars.gostore/gcworker/gc_worker.gotestkit/result.go
👮 Files not reviewed due to content moderation or server errors (8)
- meta/meta.go
- sessionctx/variable/pkdb_force_merge_test.go
- ddl/delete_range.go
- domain/pkdb_force_merge.go
- domain/pkdb_force_merge_test.go
- domain/infosync/pkdb_force_merge.go
- ddl/ddl_running_jobs.go
- ddl/pkdb_force_merge_test.go
Comment on lines
+772
to
+774
| d.ownerManager.SetRetireOwnerHook(func() { | ||
| d.runningJobs = newRunningJobs() | ||
| }) |
There was a problem hiding this comment.
Reset runningJobs in place instead of swapping the shared pointer.
This hook currently replaces d.runningJobs while workers and wait4Switch() can still dereference it concurrently, which creates a real race and can make later checks observe an empty tracker while old goroutines are still updating the previous instance. Also, this registration comes after the EnableDDL() path in Start(), so an early owner retirement can miss the reset entirely.
Proposed fix
d.ownerManager.SetBeOwnerHook(func() {
var err error
d.ddlSeqNumMu.seqNum, err = d.GetNextDDLSeqNum()
if err != nil {
logutil.BgLogger().Error("error when getting the ddl history count", zap.Error(err))
}
d.ddlCtx.setOwnerTS(time.Now().Unix())
})
+ d.ownerManager.SetRetireOwnerHook(func() {
+ d.runningJobs.reset()
+ })
d.delRangeMgr = d.newDeleteRangeManager(ctxPool == nil)
@@
d.wg.Run(d.PollTiFlashRoutine)
ingest.InitGlobalLightningEnv()
- d.ownerManager.SetRetireOwnerHook(func() {
- d.runningJobs = newRunningJobs()
- })
return nil
}And add a reset helper on runningJobs that clears the existing instance under its lock instead of replacing the pointer.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@ddl/ddl.go` around lines 772 - 774, The retire-owner hook currently replaces
the shared pointer d.runningJobs (registered via
d.ownerManager.SetRetireOwnerHook), which races with workers and wait4Switch()
that may still reference the old instance; instead implement and call a reset
method on the runningJobs type (e.g., RunningJobs.Reset or Clear) that acquires
its internal lock and zeroes/clears its internal maps/slices in-place, then
change the hook to call that reset method rather than assigning
newRunningJobs(); also ensure the hook registration happens early enough
relative to Start()/EnableDDL() so an early owner retirement cannot miss the
reset.
Comment on lines
425
to
+433
| func doRequest(ctx context.Context, apiName string, addrs []string, route, method string, body io.Reader) ([]byte, error) { | ||
| var bodyFactory func() io.Reader | ||
| if body != nil { | ||
| bodyFactory = func() io.Reader { | ||
| return body | ||
| } | ||
| } | ||
| return doRequestWithBodyFactory(ctx, apiName, addrs, route, method, bodyFactory) | ||
| } |
There was a problem hiding this comment.
Retry path still reuses a consumed request body.
At Line 428-Line 430, bodyFactory returns the same io.Reader instance each attempt. After the first failed send, subsequent retries can send an empty/partial payload.
💡 Proposed fix
func doRequest(ctx context.Context, apiName string, addrs []string, route, method string, body io.Reader) ([]byte, error) {
- var bodyFactory func() io.Reader
- if body != nil {
- bodyFactory = func() io.Reader {
- return body
- }
- }
- return doRequestWithBodyFactory(ctx, apiName, addrs, route, method, bodyFactory)
+ if body == nil {
+ return doRequestWithBodyFactory(ctx, apiName, addrs, route, method, nil)
+ }
+ bodyBytes, err := io.ReadAll(body)
+ if err != nil {
+ return nil, errors.Trace(err)
+ }
+ return doRequestWithBodyBytes(ctx, apiName, addrs, route, method, bodyBytes)
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| func doRequest(ctx context.Context, apiName string, addrs []string, route, method string, body io.Reader) ([]byte, error) { | |
| var bodyFactory func() io.Reader | |
| if body != nil { | |
| bodyFactory = func() io.Reader { | |
| return body | |
| } | |
| } | |
| return doRequestWithBodyFactory(ctx, apiName, addrs, route, method, bodyFactory) | |
| } | |
| func doRequest(ctx context.Context, apiName string, addrs []string, route, method string, body io.Reader) ([]byte, error) { | |
| if body == nil { | |
| return doRequestWithBodyFactory(ctx, apiName, addrs, route, method, nil) | |
| } | |
| bodyBytes, err := io.ReadAll(body) | |
| if err != nil { | |
| return nil, errors.Trace(err) | |
| } | |
| return doRequestWithBodyBytes(ctx, apiName, addrs, route, method, bodyBytes) | |
| } |
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@domain/infosync/info.go` around lines 425 - 433, doRequest currently sets
bodyFactory to return the same io.Reader instance on each retry, which reuses a
consumed reader; change doRequest so it buffers the original body (e.g., read
entire body into a []byte once) and make bodyFactory return a fresh io.Reader
(like bytes.NewReader(buf)) for each invocation before calling
doRequestWithBodyFactory, referencing the doRequest function and the bodyFactory
variable to implement the buffering and fresh-reader creation.
Comment on lines
+85
to
+93
| // Partitioned tables can have duplicate _tidb_rowid between different partitions | ||
| // due to EXCHANGE PARTITION, if so, do not optimize skipping rows with multiple changes | ||
| skipMultipleChangesOnSameRow := true | ||
| tbl := e.tblID2table[content.TblID] | ||
| if _, ok := tbl.(table.PartitionedTable); ok { | ||
| if !tbl.Meta().HasClusteredIndex() { | ||
| skipMultipleChangesOnSameRow = false | ||
| } | ||
| } |
There was a problem hiding this comment.
❓ Verification inconclusive
Script executed:
find . -name "update.go" -path "*/executor/*" -type fRepository: pingcap/tidb
Repository: pingcap/tidb
Exit code: 0
stdout:
./executor/update.go
Script executed:
wc -l executor/update.goRepository: pingcap/tidb
Repository: pingcap/tidb
Exit code: 0
stdout:
572 executor/update.go
Script executed:
sed -n '80,130p' executor/update.go | cat -nRepository: pingcap/tidb
Repository: pingcap/tidb
Exit code: 0
stdout:
1 e.handles = e.handles[:0]
2 e.tableUpdatable = e.tableUpdatable[:0]
3 e.changed = e.changed[:0]
4 e.matches = e.matches[:0]
5 for _, content := range e.tblColPosInfos {
6 // Partitioned tables can have duplicate _tidb_rowid between different partitions
7 // due to EXCHANGE PARTITION, if so, do not optimize skipping rows with multiple changes
8 skipMultipleChangesOnSameRow := true
9 tbl := e.tblID2table[content.TblID]
10 if _, ok := tbl.(table.PartitionedTable); ok {
11 if !tbl.Meta().HasClusteredIndex() {
12 skipMultipleChangesOnSameRow = false
13 }
14 }
15
16 if e.updatedRowKeys[content.Start] == nil {
17 e.updatedRowKeys[content.Start] = kv.NewMemAwareHandleMap[bool]()
18 }
19 handle, err := content.HandleCols.BuildHandleByDatums(row)
20 if err != nil {
21 return err
22 }
23 e.handles = append(e.handles, handle)
24
25 updatable := false
26 flags := e.assignFlag[content.Start:content.End]
27 for _, flag := range flags {
28 if flag >= 0 {
29 updatable = true
30 break
31 }
32 }
33 if unmatchedOuterRow(content, row) {
34 updatable = false
35 }
36 e.tableUpdatable = append(e.tableUpdatable, updatable)
37
38 changed, ok := e.updatedRowKeys[content.Start].Get(handle)
39 if ok {
40 e.changed = append(e.changed, changed && skipMultipleChangesOnSameRow)
41 e.matches = append(e.matches, false)
42 } else {
43 e.changed = append(e.changed, false)
44 e.matches = append(e.matches, true)
45 }
46 }
47 return nil
48 }
49
50 func (e *UpdateExec) merge(row, newData []types.Datum, mergeGenerated bool) error {
51 if e.mergedRowData == nil {
Script executed:
# Search for mergedRowData declaration and usage
rg -n "mergedRowData\s*:=" executor/update.go
rg -n "\.mergedRowData" executor/update.go -A 2Repository: pingcap/tidb
Repository: pingcap/tidb
Exit code: 0
stdout:
130: if e.mergedRowData == nil {
131: e.mergedRowData = make(map[int64]*kv.MemAwareHandleMap[[]types.Datum])
132- }
133- var mergedData []types.Datum
--
151: if e.mergedRowData[content.TblID] == nil {
152: e.mergedRowData[content.TblID] = kv.NewMemAwareHandleMap[[]types.Datum]()
153- }
154- tbl := e.tblID2table[content.TblID]
--
157: if v, ok := e.mergedRowData[content.TblID].Get(handle); ok {
158- mergedData = v
159- for i, flag := range flags {
--
174: memDelta := e.mergedRowData[content.TblID].Set(handle, mergedData)
175- memDelta += types.EstimatedMemUsage(mergedData, 1) + int64(handle.ExtraMemSize())
176- e.memTracker.Consume(memDelta)
Script executed:
# Search for updatedRowKeys usage
rg -n "updatedRowKeys" executor/update.go -A 2Repository: pingcap/tidb
Repository: pingcap/tidb
Exit code: 0
stdout:
43: // updatedRowKeys is a map for unique (TableAlias, handle) pair.
44- // The value is true if the row is changed, or false otherwise
45: updatedRowKeys map[int]*kv.MemAwareHandleMap[bool]
46- tblID2table map[int64]table.Table
47- // mergedRowData is a map for unique (Table, handle) pair.
--
77: if e.updatedRowKeys == nil {
78: e.updatedRowKeys = make(map[int]*kv.MemAwareHandleMap[bool])
79- }
80- e.handles = e.handles[:0]
--
95: if e.updatedRowKeys[content.Start] == nil {
96: e.updatedRowKeys[content.Start] = kv.NewMemAwareHandleMap[bool]()
97- }
98- handle, err := content.HandleCols.BuildHandleByDatums(row)
--
117: changed, ok := e.updatedRowKeys[content.Start].Get(handle)
118- if ok {
119- e.changed = append(e.changed, changed && skipMultipleChangesOnSameRow)
--
212: _, exist := e.updatedRowKeys[content.Start].Get(handle)
213: memDelta := e.updatedRowKeys[content.Start].Set(handle, changed)
214- if !exist {
215- memDelta += int64(handle.ExtraMemSize())
Script executed:
# Look for merge() method definition
rg -n "func.*merge" executor/update.goRepository: pingcap/tidb
Repository: pingcap/tidb
Exit code: 0
stdout:
129:func (e *UpdateExec) merge(row, newData []types.Datum, mergeGenerated bool) error {
Script executed:
# Check e.matches usage
rg -n "e\.matches" executor/update.go -B 1 -A 1Repository: pingcap/tidb
Repository: pingcap/tidb
Exit code: 0
stdout:
82- e.changed = e.changed[:0]
83: e.matches = e.matches[:0]
84- for _, content := range e.tblColPosInfos {
--
119- e.changed = append(e.changed, changed && skipMultipleChangesOnSameRow)
120: e.matches = append(e.matches, false)
121- } else {
122- e.changed = append(e.changed, false)
123: e.matches = append(e.matches, true)
124- }
--
195- }
196: if e.matches[i] {
197- // Row is matched for the first time, increment `matched` counter
Script executed:
# Check the updateRecord method to see how e.changed is used
rg -n "func.*updateRecord" executor/update.go -A 20Repository: pingcap/tidb
Repository: pingcap/tidb
Exit code: 1
Script executed:
# Check how mergedRowData is populated and used in the main update flow
sed -n '129,250p' executor/update.go | cat -nRepository: pingcap/tidb
Repository: pingcap/tidb
Exit code: 0
stdout:
1 func (e *UpdateExec) merge(row, newData []types.Datum, mergeGenerated bool) error {
2 if e.mergedRowData == nil {
3 e.mergedRowData = make(map[int64]*kv.MemAwareHandleMap[[]types.Datum])
4 }
5 var mergedData []types.Datum
6 // merge updates from and into mergedRowData
7 for i, content := range e.tblColPosInfos {
8 if !e.multiUpdateOnSameTable[content.TblID] {
9 // No need to merge if not multi-updated
10 continue
11 }
12 if !e.tableUpdatable[i] {
13 // If there's nothing to update, we can just skip current row
14 continue
15 }
16 if e.changed[i] {
17 // Each matched row is updated once, even if it matches the conditions multiple times.
18 continue
19 }
20 handle := e.handles[i]
21 flags := e.assignFlag[content.Start:content.End]
22
23 if e.mergedRowData[content.TblID] == nil {
24 e.mergedRowData[content.TblID] = kv.NewMemAwareHandleMap[[]types.Datum]()
25 }
26 tbl := e.tblID2table[content.TblID]
27 oldData := row[content.Start:content.End]
28 newTableData := newData[content.Start:content.End]
29 if v, ok := e.mergedRowData[content.TblID].Get(handle); ok {
30 mergedData = v
31 for i, flag := range flags {
32 if tbl.WritableCols()[i].IsGenerated() != mergeGenerated {
33 continue
34 }
35 mergedData[i].Copy(&oldData[i])
36 if flag >= 0 {
37 newTableData[i].Copy(&mergedData[i])
38 } else {
39 mergedData[i].Copy(&newTableData[i])
40 }
41 }
42 } else {
43 mergedData = append([]types.Datum{}, newTableData...)
44 }
45
46 memDelta := e.mergedRowData[content.TblID].Set(handle, mergedData)
47 memDelta += types.EstimatedMemUsage(mergedData, 1) + int64(handle.ExtraMemSize())
48 e.memTracker.Consume(memDelta)
49 }
50 return nil
51 }
52
53 func (e *UpdateExec) exec(ctx context.Context, schema *expression.Schema, row, newData []types.Datum) error {
54 defer trace.StartRegion(ctx, "UpdateExec").End()
55 bAssignFlag := make([]bool, len(e.assignFlag))
56 for i, flag := range e.assignFlag {
57 bAssignFlag[i] = flag >= 0
58 }
59 for i, content := range e.tblColPosInfos {
60 if !e.tableUpdatable[i] {
61 // If there's nothing to update, we can just skip current row
62 continue
63 }
64 if e.changed[i] {
65 // Each matched row is updated once, even if it matches the conditions multiple times.
66 continue
67 }
68 if e.matches[i] {
69 // Row is matched for the first time, increment `matched` counter
70 e.matched++
71 }
72 tbl := e.tblID2table[content.TblID]
73 handle := e.handles[i]
74
75 oldData := row[content.Start:content.End]
76 newTableData := newData[content.Start:content.End]
77 flags := bAssignFlag[content.Start:content.End]
78
79 // Update row
80 fkChecks := e.fkChecks[content.TblID]
81 fkCascades := e.fkCascades[content.TblID]
82 changed, err1 := updateRecord(ctx, e.ctx, handle, oldData, newTableData, flags, tbl, false, e.memTracker, fkChecks, fkCascades)
83 if err1 == nil {
84 _, exist := e.updatedRowKeys[content.Start].Get(handle)
85 memDelta := e.updatedRowKeys[content.Start].Set(handle, changed)
86 if !exist {
87 memDelta += int64(handle.ExtraMemSize())
88 }
89 e.memTracker.Consume(memDelta)
90 continue
91 }
92
93 sc := e.ctx.GetSessionVars().StmtCtx
94 if kv.ErrKeyExists.Equal(err1) && sc.DupKeyAsWarning {
95 sc.AppendWarning(err1)
96 continue
97 }
98 return err1
99 }
100 return nil
101 }
102
103 // unmatchedOuterRow checks the tableCols of a record to decide whether that record
104 // can not be updated. The handle is NULL only when it is the inner side of an
105 // outer join: the outer row can not match any inner rows, and in this scenario
106 // the inner handle field is filled with a NULL value.
107 //
108 // This fixes: https://github.com/pingcap/tidb/issues/7176.
109 func unmatchedOuterRow(tblPos plannercore.TblColPosInfo, waitUpdateRow []types.Datum) bool {
110 firstHandleIdx := tblPos.HandleCols.GetCol(0)
111 return waitUpdateRow[firstHandleIdx.Index].IsNull()
112 }
113
114 // Next implements the Executor Next interface.
115 func (e *UpdateExec) Next(ctx context.Context, req *chunk.Chunk) error {
116 req.Reset()
117 if !e.drained {
118 if e.collectRuntimeStatsEnabled() {
119 ctx = context.WithValue(ctx, autoid.AllocatorRuntimeStatsCtxKey, e.stats.AllocatorRuntimeStats)
120 }
121 numRows, err := e.updateRows(ctx)
122 if err != nil {
Script executed:
# Look for any partition-aware logic in merge or related methods
rg -n "Partition" executor/update.go -B 2 -A 2Repository: pingcap/tidb
Repository: pingcap/tidb
Exit code: 0
stdout:
83- e.matches = e.matches[:0]
84- for _, content := range e.tblColPosInfos {
85: // Partitioned tables can have duplicate _tidb_rowid between different partitions
86- // due to EXCHANGE PARTITION, if so, do not optimize skipping rows with multiple changes
87- skipMultipleChangesOnSameRow := true
88- tbl := e.tblID2table[content.TblID]
89: if _, ok := tbl.(table.PartitionedTable); ok {
90- if !tbl.Meta().HasClusteredIndex() {
91- skipMultipleChangesOnSameRow = false
Script executed:
# Check the full prepare method to understand the flow
sed -n '75,128p' executor/update.go | cat -nRepository: pingcap/tidb
Repository: pingcap/tidb
Exit code: 0
stdout:
1 // prepare `handles`, `tableUpdatable`, `changed` to avoid re-computations.
2 func (e *UpdateExec) prepare(row []types.Datum) (err error) {
3 if e.updatedRowKeys == nil {
4 e.updatedRowKeys = make(map[int]*kv.MemAwareHandleMap[bool])
5 }
6 e.handles = e.handles[:0]
7 e.tableUpdatable = e.tableUpdatable[:0]
8 e.changed = e.changed[:0]
9 e.matches = e.matches[:0]
10 for _, content := range e.tblColPosInfos {
11 // Partitioned tables can have duplicate _tidb_rowid between different partitions
12 // due to EXCHANGE PARTITION, if so, do not optimize skipping rows with multiple changes
13 skipMultipleChangesOnSameRow := true
14 tbl := e.tblID2table[content.TblID]
15 if _, ok := tbl.(table.PartitionedTable); ok {
16 if !tbl.Meta().HasClusteredIndex() {
17 skipMultipleChangesOnSameRow = false
18 }
19 }
20
21 if e.updatedRowKeys[content.Start] == nil {
22 e.updatedRowKeys[content.Start] = kv.NewMemAwareHandleMap[bool]()
23 }
24 handle, err := content.HandleCols.BuildHandleByDatums(row)
25 if err != nil {
26 return err
27 }
28 e.handles = append(e.handles, handle)
29
30 updatable := false
31 flags := e.assignFlag[content.Start:content.End]
32 for _, flag := range flags {
33 if flag >= 0 {
34 updatable = true
35 break
36 }
37 }
38 if unmatchedOuterRow(content, row) {
39 updatable = false
40 }
41 e.tableUpdatable = append(e.tableUpdatable, updatable)
42
43 changed, ok := e.updatedRowKeys[content.Start].Get(handle)
44 if ok {
45 e.changed = append(e.changed, changed && skipMultipleChangesOnSameRow)
46 e.matches = append(e.matches, false)
47 } else {
48 e.changed = append(e.changed, false)
49 e.matches = append(e.matches, true)
50 }
51 }
52 return nil
53 }
54
Script executed:
# Check where mergedRowData is read/used after being populated
rg -n "mergedRowData" executor/update.goRepository: pingcap/tidb
Repository: pingcap/tidb
Exit code: 0
stdout:
47: // mergedRowData is a map for unique (Table, handle) pair.
49: mergedRowData map[int64]*kv.MemAwareHandleMap[[]types.Datum]
130: if e.mergedRowData == nil {
131: e.mergedRowData = make(map[int64]*kv.MemAwareHandleMap[[]types.Datum])
134: // merge updates from and into mergedRowData
151: if e.mergedRowData[content.TblID] == nil {
152: e.mergedRowData[content.TblID] = kv.NewMemAwareHandleMap[[]types.Datum]()
157: if v, ok := e.mergedRowData[content.TblID].Get(handle); ok {
174: memDelta := e.mergedRowData[content.TblID].Set(handle, mergedData)
Script executed:
# Check the updateRows method to understand the flow
sed -n '300,400p' executor/update.go | cat -nRepository: pingcap/tidb
Repository: pingcap/tidb
Exit code: 0
stdout:
1 for rowIdx := 0; rowIdx < chk.NumRows(); rowIdx++ {
2 chunkRow := chk.GetRow(rowIdx)
3 datumRow := chunkRow.GetDatumRow(fields)
4 // precomputes handles
5 if err := e.prepare(datumRow); err != nil {
6 return 0, err
7 }
8 // compose non-generated columns
9 newRow, err := composeFunc(globalRowIdx, datumRow, colsInfo)
10 if err != nil {
11 return 0, err
12 }
13 // merge non-generated columns
14 if err := e.merge(datumRow, newRow, false); err != nil {
15 return 0, err
16 }
17 if e.virtualAssignmentsOffset < len(e.OrderedList) {
18 // compose generated columns
19 newRow, err = e.composeGeneratedColumns(globalRowIdx, newRow, colsInfo)
20 if err != nil {
21 return 0, err
22 }
23 // merge generated columns
24 if err := e.merge(datumRow, newRow, true); err != nil {
25 return 0, err
26 }
27 }
28 // write to table
29 if err := e.exec(ctx, e.children[0].Schema(), datumRow, newRow); err != nil {
30 return 0, err
31 }
32 }
33 totalNumRows += chk.NumRows()
34 chk = chunk.Renew(chk, e.maxChunkSize)
35 }
36 return totalNumRows, nil
37 }
38
39 func (e *UpdateExec) handleErr(colName model.CIStr, rowIdx int, err error) error {
40 if err == nil {
41 return nil
42 }
43
44 if types.ErrDataTooLong.Equal(err) {
45 return resetErrDataTooLong(colName.O, rowIdx+1, err)
46 }
47
48 if types.ErrOverflow.Equal(err) {
49 return types.ErrWarnDataOutOfRange.GenWithStackByArgs(colName.O, rowIdx+1)
50 }
51
52 return err
53 }
54
55 func (e *UpdateExec) fastComposeNewRow(rowIdx int, oldRow []types.Datum, cols []*table.Column) ([]types.Datum, error) {
56 newRowData := types.CloneRow(oldRow)
57 for _, assign := range e.OrderedList {
58 tblIdx := e.assignFlag[assign.Col.Index]
59 if tblIdx >= 0 && !e.tableUpdatable[tblIdx] {
60 continue
61 }
62 con := assign.Expr.(*expression.Constant)
63 val, err := con.Eval(emptyRow)
64 if err = e.handleErr(assign.ColName, rowIdx, err); err != nil {
65 return nil, err
66 }
67
68 // info of `_tidb_rowid` column is nil.
69 // No need to cast `_tidb_rowid` column value.
70 if cols[assign.Col.Index] != nil {
71 val, err = table.CastValue(e.ctx, val, cols[assign.Col.Index].ColumnInfo, false, false)
72 if err = e.handleErr(assign.ColName, rowIdx, err); err != nil {
73 return nil, err
74 }
75 }
76
77 val.Copy(&newRowData[assign.Col.Index])
78 }
79 return newRowData, nil
80 }
81
82 func (e *UpdateExec) composeNewRow(rowIdx int, oldRow []types.Datum, cols []*table.Column) ([]types.Datum, error) {
83 newRowData := types.CloneRow(oldRow)
84 e.evalBuffer.SetDatums(newRowData...)
85 for _, assign := range e.OrderedList[:e.virtualAssignmentsOffset] {
86 tblIdx := e.assignFlag[assign.Col.Index]
87 if tblIdx >= 0 && !e.tableUpdatable[tblIdx] {
88 continue
89 }
90 val, err := assign.Expr.Eval(e.evalBuffer.ToRow())
91 if err != nil {
92 return nil, err
93 }
94
95 // info of `_tidb_rowid` column is nil.
96 // No need to cast `_tidb_rowid` column value.
97 if cols[assign.Col.Index] != nil {
98 val, err = table.CastValue(e.ctx, val, cols[assign.Col.Index].ColumnInfo, false, false)
99 if err = e.handleErr(assign.ColName, rowIdx, err); err != nil {
100 return nil, err
101 }
Script executed:
# Look for all uses of merged data after merge() is called
rg -n "func.*updateRows" executor/update.go -A 50 | head -100Repository: pingcap/tidb
Repository: pingcap/tidb
Exit code: 0
stdout:
259:func (e *UpdateExec) updateRows(ctx context.Context) (int, error) {
260- fields := retTypes(e.children[0])
261- colsInfo := plannercore.GetUpdateColumnsInfo(e.tblID2table, e.tblColPosInfos, len(fields))
262- globalRowIdx := 0
263- chk := tryNewCacheChunk(e.children[0])
264- if !e.allAssignmentsAreConstant {
265- e.evalBuffer = chunk.MutRowFromTypes(fields)
266- }
267- composeFunc := e.fastComposeNewRow
268- if !e.allAssignmentsAreConstant {
269- composeFunc = e.composeNewRow
270- }
271- memUsageOfChk := int64(0)
272- totalNumRows := 0
273- for {
274- e.memTracker.Consume(-memUsageOfChk)
275- err := Next(ctx, e.children[0], chk)
276- if err != nil {
277- return 0, err
278- }
279-
280- if chk.NumRows() == 0 {
281- break
282- }
283- memUsageOfChk = chk.MemoryUsage()
284- e.memTracker.Consume(memUsageOfChk)
285- if e.collectRuntimeStatsEnabled() {
286- txn, err := e.ctx.Txn(true)
287- if err == nil && txn.GetSnapshot() != nil {
288- txn.GetSnapshot().SetOption(kv.CollectRuntimeStats, e.stats.SnapshotRuntimeStats)
289- }
290- }
291- txn, err := e.ctx.Txn(true)
292- if err == nil {
293- sc := e.ctx.GetSessionVars().StmtCtx
294- txn.SetOption(kv.ResourceGroupTagger, sc.GetResourceGroupTagger())
295- if sc.KvExecCounter != nil {
296- // Bind an interceptor for client-go to count the number of SQL executions of each TiKV.
297- txn.SetOption(kv.RPCInterceptor, sc.KvExecCounter.RPCInterceptor())
298- }
299- }
300- for rowIdx := 0; rowIdx < chk.NumRows(); rowIdx++ {
301- chunkRow := chk.GetRow(rowIdx)
302- datumRow := chunkRow.GetDatumRow(fields)
303- // precomputes handles
304- if err := e.prepare(datumRow); err != nil {
305- return 0, err
306- }
307- // compose non-generated columns
308- newRow, err := composeFunc(globalRowIdx, datumRow, colsInfo)
309- if err != nil {
Script executed:
# Check where multiUpdateOnSameTable is set and what it means
rg -n "multiUpdateOnSameTable" executor/update.go -B 2 -A 2Repository: pingcap/tidb
Repository: pingcap/tidb
Exit code: 0
stdout:
48- // The value is cached table row
49- mergedRowData map[int64]*kv.MemAwareHandleMap[[]types.Datum]
50: multiUpdateOnSameTable map[int64]bool
51-
52- matched uint64 // a counter of matched rows during update
--
134- // merge updates from and into mergedRowData
135- for i, content := range e.tblColPosInfos {
136: if !e.multiUpdateOnSameTable[content.TblID] {
137- // No need to merge if not multi-updated
138- continue
Script executed:
# Check the struct definition to understand the state
sed -n '1,60p' executor/update.go | cat -nRepository: pingcap/tidb
Repository: pingcap/tidb
Exit code: 0
stdout:
1 // Copyright 2018 PingCAP, Inc.
2 //
3 // Licensed under the Apache License, Version 2.0 (the "License");
4 // you may not use this file except in compliance with the License.
5 // You may obtain a copy of the License at
6 //
7 // http://www.apache.org/licenses/LICENSE-2.0
8 //
9 // Unless required by applicable law or agreed to in writing, software
10 // distributed under the License is distributed on an "AS IS" BASIS,
11 // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 // See the License for the specific language governing permissions and
13 // limitations under the License.
14
15 package executor
16
17 import (
18 "bytes"
19 "context"
20 "fmt"
21 "runtime/trace"
22
23 "github.com/pingcap/tidb/expression"
24 "github.com/pingcap/tidb/kv"
25 "github.com/pingcap/tidb/meta/autoid"
26 "github.com/pingcap/tidb/parser/model"
27 "github.com/pingcap/tidb/parser/mysql"
28 plannercore "github.com/pingcap/tidb/planner/core"
29 "github.com/pingcap/tidb/table"
30 "github.com/pingcap/tidb/types"
31 "github.com/pingcap/tidb/util/chunk"
32 "github.com/pingcap/tidb/util/execdetails"
33 "github.com/pingcap/tidb/util/memory"
34 "github.com/tikv/client-go/v2/txnkv/txnsnapshot"
35 )
36
37 // UpdateExec represents a new update executor.
38 type UpdateExec struct {
39 baseExecutor
40
41 OrderedList []*expression.Assignment
42
43 // updatedRowKeys is a map for unique (TableAlias, handle) pair.
44 // The value is true if the row is changed, or false otherwise
45 updatedRowKeys map[int]*kv.MemAwareHandleMap[bool]
46 tblID2table map[int64]table.Table
47 // mergedRowData is a map for unique (Table, handle) pair.
48 // The value is cached table row
49 mergedRowData map[int64]*kv.MemAwareHandleMap[[]types.Datum]
50 multiUpdateOnSameTable map[int64]bool
51
52 matched uint64 // a counter of matched rows during update
53 // tblColPosInfos stores relationship between column ordinal to its table handle.
54 // the columns ordinals is present in ordinal range format, `@see` plannercore.TblColPosInfos
55 tblColPosInfos plannercore.TblColPosInfoSlice
56 assignFlag []int
57 evalBuffer chunk.MutRow
58 allAssignmentsAreConstant bool
59 virtualAssignmentsOffset int
60 drained bool
Script executed:
# Look for the full flow - particularly how multi-update is detected
rg -n "multiUpdateOnSameTable\[" executor/update.goRepository: pingcap/tidb
Repository: pingcap/tidb
Exit code: 0
stdout:
136: if !e.multiUpdateOnSameTable[content.TblID] {
Script executed:
# Check if mergedRowData is flushed or used after updateRows
rg -n "mergedRowData" executor/update.go -B 3 -A 3Repository: pingcap/tidb
Repository: pingcap/tidb
Exit code: 0
stdout:
44- // The value is true if the row is changed, or false otherwise
45- updatedRowKeys map[int]*kv.MemAwareHandleMap[bool]
46- tblID2table map[int64]table.Table
47: // mergedRowData is a map for unique (Table, handle) pair.
48- // The value is cached table row
49: mergedRowData map[int64]*kv.MemAwareHandleMap[[]types.Datum]
50- multiUpdateOnSameTable map[int64]bool
51-
52- matched uint64 // a counter of matched rows during update
--
127-}
128-
129-func (e *UpdateExec) merge(row, newData []types.Datum, mergeGenerated bool) error {
130: if e.mergedRowData == nil {
131: e.mergedRowData = make(map[int64]*kv.MemAwareHandleMap[[]types.Datum])
132- }
133- var mergedData []types.Datum
134: // merge updates from and into mergedRowData
135- for i, content := range e.tblColPosInfos {
136- if !e.multiUpdateOnSameTable[content.TblID] {
137- // No need to merge if not multi-updated
--
148- handle := e.handles[i]
149- flags := e.assignFlag[content.Start:content.End]
150-
151: if e.mergedRowData[content.TblID] == nil {
152: e.mergedRowData[content.TblID] = kv.NewMemAwareHandleMap[[]types.Datum]()
153- }
154- tbl := e.tblID2table[content.TblID]
155- oldData := row[content.Start:content.End]
156- newTableData := newData[content.Start:content.End]
157: if v, ok := e.mergedRowData[content.TblID].Get(handle); ok {
158- mergedData = v
159- for i, flag := range flags {
160- if tbl.WritableCols()[i].IsGenerated() != mergeGenerated {
--
171- mergedData = append([]types.Datum{}, newTableData...)
172- }
173-
174: memDelta := e.mergedRowData[content.TblID].Set(handle, mergedData)
175- memDelta += types.EstimatedMemUsage(mergedData, 1) + int64(handle.ExtraMemSize())
176- e.memTracker.Consume(memDelta)
177- }
Script executed:
# Look at the complete flow - check if mergedRowData is applied somewhere
sed -n '140,180p' executor/update.go | cat -nRepository: pingcap/tidb
Repository: pingcap/tidb
Exit code: 0
stdout:
1 if !e.tableUpdatable[i] {
2 // If there's nothing to update, we can just skip current row
3 continue
4 }
5 if e.changed[i] {
6 // Each matched row is updated once, even if it matches the conditions multiple times.
7 continue
8 }
9 handle := e.handles[i]
10 flags := e.assignFlag[content.Start:content.End]
11
12 if e.mergedRowData[content.TblID] == nil {
13 e.mergedRowData[content.TblID] = kv.NewMemAwareHandleMap[[]types.Datum]()
14 }
15 tbl := e.tblID2table[content.TblID]
16 oldData := row[content.Start:content.End]
17 newTableData := newData[content.Start:content.End]
18 if v, ok := e.mergedRowData[content.TblID].Get(handle); ok {
19 mergedData = v
20 for i, flag := range flags {
21 if tbl.WritableCols()[i].IsGenerated() != mergeGenerated {
22 continue
23 }
24 mergedData[i].Copy(&oldData[i])
25 if flag >= 0 {
26 newTableData[i].Copy(&mergedData[i])
27 } else {
28 mergedData[i].Copy(&newTableData[i])
29 }
30 }
31 } else {
32 mergedData = append([]types.Datum{}, newTableData...)
33 }
34
35 memDelta := e.mergedRowData[content.TblID].Set(handle, mergedData)
36 memDelta += types.EstimatedMemUsage(mergedData, 1) + int64(handle.ExtraMemSize())
37 e.memTracker.Consume(memDelta)
38 }
39 return nil
40 }
41
Script executed:
# Check what happens after the row loop - finalization
sed -n '500,572p' executor/update.go | cat -nRepository: pingcap/tidb
Repository: pingcap/tidb
Exit code: 0
stdout:
1 if e.AllocatorRuntimeStats != nil {
2 stats := e.AllocatorRuntimeStats.String()
3 if stats != "" {
4 if buf.Len() > 0 {
5 buf.WriteString(", ")
6 }
7 buf.WriteString(stats)
8 }
9 }
10 return buf.String()
11 }
12
13 // Clone implements the RuntimeStats interface.
14 func (e *updateRuntimeStats) Clone() execdetails.RuntimeStats {
15 newRs := &updateRuntimeStats{}
16 if e.SnapshotRuntimeStats != nil {
17 snapshotStats := e.SnapshotRuntimeStats.Clone()
18 newRs.SnapshotRuntimeStats = snapshotStats
19 }
20 if e.AllocatorRuntimeStats != nil {
21 newRs.AllocatorRuntimeStats = e.AllocatorRuntimeStats.Clone()
22 }
23 return newRs
24 }
25
26 // Merge implements the RuntimeStats interface.
27 func (e *updateRuntimeStats) Merge(other execdetails.RuntimeStats) {
28 tmp, ok := other.(*updateRuntimeStats)
29 if !ok {
30 return
31 }
32 if tmp.SnapshotRuntimeStats != nil {
33 if e.SnapshotRuntimeStats == nil {
34 snapshotStats := tmp.SnapshotRuntimeStats.Clone()
35 e.SnapshotRuntimeStats = snapshotStats
36 } else {
37 e.SnapshotRuntimeStats.Merge(tmp.SnapshotRuntimeStats)
38 }
39 }
40 if tmp.AllocatorRuntimeStats != nil {
41 if e.AllocatorRuntimeStats == nil {
42 e.AllocatorRuntimeStats = tmp.AllocatorRuntimeStats.Clone()
43 }
44 }
45 }
46
47 // Tp implements the RuntimeStats interface.
48 func (e *updateRuntimeStats) Tp() int {
49 return execdetails.TpUpdateRuntimeStats
50 }
51
52 // GetFKChecks implements WithForeignKeyTrigger interface.
53 func (e *UpdateExec) GetFKChecks() []*FKCheckExec {
54 fkChecks := make([]*FKCheckExec, 0, len(e.fkChecks))
55 for _, fkc := range e.fkChecks {
56 fkChecks = append(fkChecks, fkc...)
57 }
58 return fkChecks
59 }
60
61 // GetFKCascades implements WithForeignKeyTrigger interface.
62 func (e *UpdateExec) GetFKCascades() []*FKCascadeExec {
63 fkCascades := make([]*FKCascadeExec, 0, len(e.fkChecks))
64 for _, fkc := range e.fkCascades {
65 fkCascades = append(fkCascades, fkc...)
66 }
67 return fkCascades
68 }
69
70 // HasFKCascades implements WithForeignKeyTrigger interface.
71 func (e *UpdateExec) HasFKCascades() bool {
72 return len(e.fkCascades) > 0
73 }
Script executed:
# Search for where mergedRowData is actually used after being populated
rg -n "mergedRowData\[" executor/update.goRepository: pingcap/tidb
Repository: pingcap/tidb
Exit code: 0
stdout:
151: if e.mergedRowData[content.TblID] == nil {
152: e.mergedRowData[content.TblID] = kv.NewMemAwareHandleMap[[]types.Datum]()
157: if v, ok := e.mergedRowData[content.TblID].Get(handle); ok {
174: memDelta := e.mergedRowData[content.TblID].Set(handle, mergedData)
Script executed:
# Check if there's a finalization step that applies mergedRowData
rg -n "Close\|finalize" executor/update.go -iRepository: pingcap/tidb
Repository: pingcap/tidb
Exit code: 1
Script executed:
# Look for any test that exercises the multi-update on same table scenario
find . -name "*test.go" -path "*/executor/*" | head -10Repository: pingcap/tidb
Repository: pingcap/tidb
Exit code: 0
stdout:
./executor/adapter_test.go
./executor/admin_test.go
./executor/aggfuncs/aggfunc_test.go
./executor/aggfuncs/export_test.go
./executor/aggfuncs/func_avg_test.go
./executor/aggfuncs/func_bitfuncs_test.go
./executor/aggfuncs/func_count_test.go
./executor/aggfuncs/func_cume_dist_test.go
./executor/aggfuncs/func_first_row_test.go
./executor/aggfuncs/func_group_concat_test.go
Script executed:
# Search for multi-update tests
rg -l "multiUpdate\|multi.*update" tests/ -i | head -5Repository: pingcap/tidb
Repository: pingcap/tidb
Exit code: 0
Script executed:
# Check what happens with mergedRowData - is it ever flushed or applied?
sed -n '125,250p' executor/update.go | cat -nRepository: pingcap/tidb
Repository: pingcap/tidb
Exit code: 0
stdout:
1 }
2 return nil
3 }
4
5 func (e *UpdateExec) merge(row, newData []types.Datum, mergeGenerated bool) error {
6 if e.mergedRowData == nil {
7 e.mergedRowData = make(map[int64]*kv.MemAwareHandleMap[[]types.Datum])
8 }
9 var mergedData []types.Datum
10 // merge updates from and into mergedRowData
11 for i, content := range e.tblColPosInfos {
12 if !e.multiUpdateOnSameTable[content.TblID] {
13 // No need to merge if not multi-updated
14 continue
15 }
16 if !e.tableUpdatable[i] {
17 // If there's nothing to update, we can just skip current row
18 continue
19 }
20 if e.changed[i] {
21 // Each matched row is updated once, even if it matches the conditions multiple times.
22 continue
23 }
24 handle := e.handles[i]
25 flags := e.assignFlag[content.Start:content.End]
26
27 if e.mergedRowData[content.TblID] == nil {
28 e.mergedRowData[content.TblID] = kv.NewMemAwareHandleMap[[]types.Datum]()
29 }
30 tbl := e.tblID2table[content.TblID]
31 oldData := row[content.Start:content.End]
32 newTableData := newData[content.Start:content.End]
33 if v, ok := e.mergedRowData[content.TblID].Get(handle); ok {
34 mergedData = v
35 for i, flag := range flags {
36 if tbl.WritableCols()[i].IsGenerated() != mergeGenerated {
37 continue
38 }
39 mergedData[i].Copy(&oldData[i])
40 if flag >= 0 {
41 newTableData[i].Copy(&mergedData[i])
42 } else {
43 mergedData[i].Copy(&newTableData[i])
44 }
45 }
46 } else {
47 mergedData = append([]types.Datum{}, newTableData...)
48 }
49
50 memDelta := e.mergedRowData[content.TblID].Set(handle, mergedData)
51 memDelta += types.EstimatedMemUsage(mergedData, 1) + int64(handle.ExtraMemSize())
52 e.memTracker.Consume(memDelta)
53 }
54 return nil
55 }
56
57 func (e *UpdateExec) exec(ctx context.Context, schema *expression.Schema, row, newData []types.Datum) error {
58 defer trace.StartRegion(ctx, "UpdateExec").End()
59 bAssignFlag := make([]bool, len(e.assignFlag))
60 for i, flag := range e.assignFlag {
61 bAssignFlag[i] = flag >= 0
62 }
63 for i, content := range e.tblColPosInfos {
64 if !e.tableUpdatable[i] {
65 // If there's nothing to update, we can just skip current row
66 continue
67 }
68 if e.changed[i] {
69 // Each matched row is updated once, even if it matches the conditions multiple times.
70 continue
71 }
72 if e.matches[i] {
73 // Row is matched for the first time, increment `matched` counter
74 e.matched++
75 }
76 tbl := e.tblID2table[content.TblID]
77 handle := e.handles[i]
78
79 oldData := row[content.Start:content.End]
80 newTableData := newData[content.Start:content.End]
81 flags := bAssignFlag[content.Start:content.End]
82
83 // Update row
84 fkChecks := e.fkChecks[content.TblID]
85 fkCascades := e.fkCascades[content.TblID]
86 changed, err1 := updateRecord(ctx, e.ctx, handle, oldData, newTableData, flags, tbl, false, e.memTracker, fkChecks, fkCascades)
87 if err1 == nil {
88 _, exist := e.updatedRowKeys[content.Start].Get(handle)
89 memDelta := e.updatedRowKeys[content.Start].Set(handle, changed)
90 if !exist {
91 memDelta += int64(handle.ExtraMemSize())
92 }
93 e.memTracker.Consume(memDelta)
94 continue
95 }
96
97 sc := e.ctx.GetSessionVars().StmtCtx
98 if kv.ErrKeyExists.Equal(err1) && sc.DupKeyAsWarning {
99 sc.AppendWarning(err1)
100 continue
101 }
102 return err1
103 }
104 return nil
105 }
106
107 // unmatchedOuterRow checks the tableCols of a record to decide whether that record
108 // can not be updated. The handle is NULL only when it is the inner side of an
109 // outer join: the outer row can not match any inner rows, and in this scenario
110 // the inner handle field is filled with a NULL value.
111 //
112 // This fixes: https://github.com/pingcap/tidb/issues/7176.
113 func unmatchedOuterRow(tblPos plannercore.TblColPosInfo, waitUpdateRow []types.Datum) bool {
114 firstHandleIdx := tblPos.HandleCols.GetCol(0)
115 return waitUpdateRow[firstHandleIdx.Index].IsNull()
116 }
117
118 // Next implements the Executor Next interface.
119 func (e *UpdateExec) Next(ctx context.Context, req *chunk.Chunk) error {
120 req.Reset()
121 if !e.drained {
122 if e.collectRuntimeStatsEnabled() {
123 ctx = context.WithValue(ctx, autoid.AllocatorRuntimeStatsCtxKey, e.stats.AllocatorRuntimeStats)
124 }
125 numRows, err := e.updateRows(ctx)
126 if err != nil {
Script executed:
# Check if mergedRowData is used anywhere else in the codebase
rg -n "mergedRowData" . --type goRepository: pingcap/tidb
Repository: pingcap/tidb
Exit code: 0
stdout:
./executor/update.go:47: // mergedRowData is a map for unique (Table, handle) pair.
./executor/update.go:49: mergedRowData map[int64]*kv.MemAwareHandleMap[[]types.Datum]
./executor/update.go:130: if e.mergedRowData == nil {
./executor/update.go:131: e.mergedRowData = make(map[int64]*kv.MemAwareHandleMap[[]types.Datum])
./executor/update.go:134: // merge updates from and into mergedRowData
./executor/update.go:151: if e.mergedRowData[content.TblID] == nil {
./executor/update.go:152: e.mergedRowData[content.TblID] = kv.NewMemAwareHandleMap[[]types.Datum]()
./executor/update.go:157: if v, ok := e.mergedRowData[content.TblID].Get(handle); ok {
./executor/update.go:174: memDelta := e.mergedRowData[content.TblID].Set(handle, mergedData)
Script executed:
# Look at git history/context for this update.go file
git log --oneline executor/update.go | head -20Repository: pingcap/tidb
Repository: pingcap/tidb
Exit code: 0
stdout:
752e495b ddl/server: add async force merge for empty table-id regions (`#684`)
Script executed:
# Search for tests that exercise multi-update scenarios
rg -l "UPDATE.*UPDATE\|multiUpdate" tests/ --type go | head -3Repository: pingcap/tidb
Repository: pingcap/tidb
Exit code: 0
Script executed:
# Check the exec() function more carefully - does newTableData get modified?
sed -n '57,105p' executor/update.go | cat -nRepository: pingcap/tidb
Repository: pingcap/tidb
Exit code: 0
stdout:
1 evalBuffer chunk.MutRow
2 allAssignmentsAreConstant bool
3 virtualAssignmentsOffset int
4 drained bool
5 memTracker *memory.Tracker
6
7 stats *updateRuntimeStats
8
9 handles []kv.Handle
10 tableUpdatable []bool
11 changed []bool
12 matches []bool
13 // fkChecks contains the foreign key checkers. the map is tableID -> []*FKCheckExec
14 fkChecks map[int64][]*FKCheckExec
15 // fkCascades contains the foreign key cascade. the map is tableID -> []*FKCascadeExec
16 fkCascades map[int64][]*FKCascadeExec
17 }
18
19 // prepare `handles`, `tableUpdatable`, `changed` to avoid re-computations.
20 func (e *UpdateExec) prepare(row []types.Datum) (err error) {
21 if e.updatedRowKeys == nil {
22 e.updatedRowKeys = make(map[int]*kv.MemAwareHandleMap[bool])
23 }
24 e.handles = e.handles[:0]
25 e.tableUpdatable = e.tableUpdatable[:0]
26 e.changed = e.changed[:0]
27 e.matches = e.matches[:0]
28 for _, content := range e.tblColPosInfos {
29 // Partitioned tables can have duplicate _tidb_rowid between different partitions
30 // due to EXCHANGE PARTITION, if so, do not optimize skipping rows with multiple changes
31 skipMultipleChangesOnSameRow := true
32 tbl := e.tblID2table[content.TblID]
33 if _, ok := tbl.(table.PartitionedTable); ok {
34 if !tbl.Meta().HasClusteredIndex() {
35 skipMultipleChangesOnSameRow = false
36 }
37 }
38
39 if e.updatedRowKeys[content.Start] == nil {
40 e.updatedRowKeys[content.Start] = kv.NewMemAwareHandleMap[bool]()
41 }
42 handle, err := content.HandleCols.BuildHandleByDatums(row)
43 if err != nil {
44 return err
45 }
46 e.handles = append(e.handles, handle)
47
48 updatable := false
49 flags := e.assignFlag[content.Start:content.End]
Script executed:
# Let me trace the actual data flow - does merge() modify newTableData?
sed -n '155,178p' executor/update.go | cat -nRepository: pingcap/tidb
Repository: pingcap/tidb
Exit code: 0
stdout:
1 oldData := row[content.Start:content.End]
2 newTableData := newData[content.Start:content.End]
3 if v, ok := e.mergedRowData[content.TblID].Get(handle); ok {
4 mergedData = v
5 for i, flag := range flags {
6 if tbl.WritableCols()[i].IsGenerated() != mergeGenerated {
7 continue
8 }
9 mergedData[i].Copy(&oldData[i])
10 if flag >= 0 {
11 newTableData[i].Copy(&mergedData[i])
12 } else {
13 mergedData[i].Copy(&newTableData[i])
14 }
15 }
16 } else {
17 mergedData = append([]types.Datum{}, newTableData...)
18 }
19
20 memDelta := e.mergedRowData[content.TblID].Set(handle, mergedData)
21 memDelta += types.EstimatedMemUsage(mergedData, 1) + int64(handle.ExtraMemSize())
22 e.memTracker.Consume(memDelta)
23 }
24 return nil
Script executed:
# Let's look at the complete picture - how does mergedRowData affect the final update?
sed -n '150,175p' executor/update.go | cat -nRepository: pingcap/tidb
Repository: pingcap/tidb
Exit code: 0
stdout:
1
2 if e.mergedRowData[content.TblID] == nil {
3 e.mergedRowData[content.TblID] = kv.NewMemAwareHandleMap[[]types.Datum]()
4 }
5 tbl := e.tblID2table[content.TblID]
6 oldData := row[content.Start:content.End]
7 newTableData := newData[content.Start:content.End]
8 if v, ok := e.mergedRowData[content.TblID].Get(handle); ok {
9 mergedData = v
10 for i, flag := range flags {
11 if tbl.WritableCols()[i].IsGenerated() != mergeGenerated {
12 continue
13 }
14 mergedData[i].Copy(&oldData[i])
15 if flag >= 0 {
16 newTableData[i].Copy(&mergedData[i])
17 } else {
18 mergedData[i].Copy(&newTableData[i])
19 }
20 }
21 } else {
22 mergedData = append([]types.Datum{}, newTableData...)
23 }
24
25 memDelta := e.mergedRowData[content.TblID].Set(handle, mergedData)
26 memDelta += types.EstimatedMemUsage(mergedData, 1) + int64(handle.ExtraMemSize())
Script executed:
# Check how exec() uses the newTableData that was potentially modified by merge()
sed -n '53,105p' executor/update.go | cat -nRepository: pingcap/tidb
Repository: pingcap/tidb
Exit code: 0
stdout:
1 // tblColPosInfos stores relationship between column ordinal to its table handle.
2 // the columns ordinals is present in ordinal range format, `@see` plannercore.TblColPosInfos
3 tblColPosInfos plannercore.TblColPosInfoSlice
4 assignFlag []int
5 evalBuffer chunk.MutRow
6 allAssignmentsAreConstant bool
7 virtualAssignmentsOffset int
8 drained bool
9 memTracker *memory.Tracker
10
11 stats *updateRuntimeStats
12
13 handles []kv.Handle
14 tableUpdatable []bool
15 changed []bool
16 matches []bool
17 // fkChecks contains the foreign key checkers. the map is tableID -> []*FKCheckExec
18 fkChecks map[int64][]*FKCheckExec
19 // fkCascades contains the foreign key cascade. the map is tableID -> []*FKCascadeExec
20 fkCascades map[int64][]*FKCascadeExec
21 }
22
23 // prepare `handles`, `tableUpdatable`, `changed` to avoid re-computations.
24 func (e *UpdateExec) prepare(row []types.Datum) (err error) {
25 if e.updatedRowKeys == nil {
26 e.updatedRowKeys = make(map[int]*kv.MemAwareHandleMap[bool])
27 }
28 e.handles = e.handles[:0]
29 e.tableUpdatable = e.tableUpdatable[:0]
30 e.changed = e.changed[:0]
31 e.matches = e.matches[:0]
32 for _, content := range e.tblColPosInfos {
33 // Partitioned tables can have duplicate _tidb_rowid between different partitions
34 // due to EXCHANGE PARTITION, if so, do not optimize skipping rows with multiple changes
35 skipMultipleChangesOnSameRow := true
36 tbl := e.tblID2table[content.TblID]
37 if _, ok := tbl.(table.PartitionedTable); ok {
38 if !tbl.Meta().HasClusteredIndex() {
39 skipMultipleChangesOnSameRow = false
40 }
41 }
42
43 if e.updatedRowKeys[content.Start] == nil {
44 e.updatedRowKeys[content.Start] = kv.NewMemAwareHandleMap[bool]()
45 }
46 handle, err := content.HandleCols.BuildHandleByDatums(row)
47 if err != nil {
48 return err
49 }
50 e.handles = append(e.handles, handle)
51
52 updatable := false
53 flags := e.assignFlag[content.Start:content.End]
Script executed:
# Look for any recent fixes related to partition handle collision
git log --all --oneline --grep="partition\|handle\|EXCHANGE" -- executor/update.go | head -20Repository: pingcap/tidb
Repository: pingcap/tidb
Exit code: 0
Script executed:
# Search for tests that check multi-update on partitioned tables
rg -n "EXCHANGE PARTITION" tests/ -A 5 -B 5 | head -80Repository: pingcap/tidb
Repository: pingcap/tidb
Exit code: 0
Script executed:
# Verify the complete flow in updateRows
sed -n '259,340p' executor/update.go | cat -nRepository: pingcap/tidb
Repository: pingcap/tidb
Exit code: 0
stdout:
1 func (e *UpdateExec) updateRows(ctx context.Context) (int, error) {
2 fields := retTypes(e.children[0])
3 colsInfo := plannercore.GetUpdateColumnsInfo(e.tblID2table, e.tblColPosInfos, len(fields))
4 globalRowIdx := 0
5 chk := tryNewCacheChunk(e.children[0])
6 if !e.allAssignmentsAreConstant {
7 e.evalBuffer = chunk.MutRowFromTypes(fields)
8 }
9 composeFunc := e.fastComposeNewRow
10 if !e.allAssignmentsAreConstant {
11 composeFunc = e.composeNewRow
12 }
13 memUsageOfChk := int64(0)
14 totalNumRows := 0
15 for {
16 e.memTracker.Consume(-memUsageOfChk)
17 err := Next(ctx, e.children[0], chk)
18 if err != nil {
19 return 0, err
20 }
21
22 if chk.NumRows() == 0 {
23 break
24 }
25 memUsageOfChk = chk.MemoryUsage()
26 e.memTracker.Consume(memUsageOfChk)
27 if e.collectRuntimeStatsEnabled() {
28 txn, err := e.ctx.Txn(true)
29 if err == nil && txn.GetSnapshot() != nil {
30 txn.GetSnapshot().SetOption(kv.CollectRuntimeStats, e.stats.SnapshotRuntimeStats)
31 }
32 }
33 txn, err := e.ctx.Txn(true)
34 if err == nil {
35 sc := e.ctx.GetSessionVars().StmtCtx
36 txn.SetOption(kv.ResourceGroupTagger, sc.GetResourceGroupTagger())
37 if sc.KvExecCounter != nil {
38 // Bind an interceptor for client-go to count the number of SQL executions of each TiKV.
39 txn.SetOption(kv.RPCInterceptor, sc.KvExecCounter.RPCInterceptor())
40 }
41 }
42 for rowIdx := 0; rowIdx < chk.NumRows(); rowIdx++ {
43 chunkRow := chk.GetRow(rowIdx)
44 datumRow := chunkRow.GetDatumRow(fields)
45 // precomputes handles
46 if err := e.prepare(datumRow); err != nil {
47 return 0, err
48 }
49 // compose non-generated columns
50 newRow, err := composeFunc(globalRowIdx, datumRow, colsInfo)
51 if err != nil {
52 return 0, err
53 }
54 // merge non-generated columns
55 if err := e.merge(datumRow, newRow, false); err != nil {
56 return 0, err
57 }
58 if e.virtualAssignmentsOffset < len(e.OrderedList) {
59 // compose generated columns
60 newRow, err = e.composeGeneratedColumns(globalRowIdx, newRow, colsInfo)
61 if err != nil {
62 return 0, err
63 }
64 // merge generated columns
65 if err := e.merge(datumRow, newRow, true); err != nil {
66 return 0, err
67 }
68 }
69 // write to table
70 if err := e.exec(ctx, e.children[0].Schema(), datumRow, newRow); err != nil {
71 return 0, err
72 }
73 }
74 totalNumRows += chk.NumRows()
75 chk = chunk.Renew(chk, e.maxChunkSize)
76 }
77 return totalNumRows, nil
78 }
79
80 func (e *UpdateExec) handleErr(colName model.CIStr, rowIdx int, err error) error {
81 if err == nil {
82 return nil
Script executed:
# Double-check the changed flag behavior in prepare and its downstream effects
sed -n '116,128p' executor/update.go | cat -nRepository: pingcap/tidb
Repository: pingcap/tidb
Exit code: 0
stdout:
1
2 changed, ok := e.updatedRowKeys[content.Start].Get(handle)
3 if ok {
4 e.changed = append(e.changed, changed && skipMultipleChangesOnSameRow)
5 e.matches = append(e.matches, false)
6 } else {
7 e.changed = append(e.changed, false)
8 e.matches = append(e.matches, true)
9 }
10 }
11 return nil
12 }
13
Script executed:
# Verify how multiUpdateOnSameTable is set - is it table-specific or partition-specific?
rg -n "multiUpdateOnSameTable\[" . --type go -B 2 -A 2Repository: pingcap/tidb
Repository: pingcap/tidb
Exit code: 0
stdout:
./executor/update.go-134- // merge updates from and into mergedRowData
./executor/update.go-135- for i, content := range e.tblColPosInfos {
./executor/update.go:136: if !e.multiUpdateOnSameTable[content.TblID] {
./executor/update.go-137- // No need to merge if not multi-updated
./executor/update.go-138- continue
--
./executor/builder.go-2295- tbl, _ := b.is.TableByID(info.TblID)
./executor/builder.go-2296- if _, ok := tblID2table[info.TblID]; ok {
./executor/builder.go:2297: multiUpdateOnSameTable[info.TblID] = true
./executor/builder.go-2298- }
./executor/builder.go-2299- tblID2table[info.TblID] = tbl
Script executed:
# Check what BuildHandleByDatums actually produces - is it partition-aware?
rg -n "BuildHandleByDatums" . --type go | head -5Repository: pingcap/tidb
Repository: pingcap/tidb
Exit code: 0
stdout:
./planner/core/handle_cols.go:38: // BuildHandleByDatums builds a Handle from a datum slice.
./planner/core/handle_cols.go:39: BuildHandleByDatums(row []types.Datum) (kv.Handle, error)
./planner/core/handle_cols.go:96:// BuildHandleByDatums implements the kv.HandleCols interface.
./planner/core/handle_cols.go:97:func (cb *CommonHandleCols) BuildHandleByDatums(row []types.Datum) (kv.Handle, error) {
./planner/core/handle_cols.go:222:// BuildHandleByDatums implements the kv.HandleCols interface.
Script executed:
# Final check: if skipMultipleChangesOnSameRow=false, what path does merge() take?
sed -n '136,178p' executor/update.goRepository: pingcap/tidb
Repository: pingcap/tidb
Exit code: 0
stdout:
if !e.multiUpdateOnSameTable[content.TblID] {
// No need to merge if not multi-updated
continue
}
if !e.tableUpdatable[i] {
// If there's nothing to update, we can just skip current row
continue
}
if e.changed[i] {
// Each matched row is updated once, even if it matches the conditions multiple times.
continue
}
handle := e.handles[i]
flags := e.assignFlag[content.Start:content.End]
if e.mergedRowData[content.TblID] == nil {
e.mergedRowData[content.TblID] = kv.NewMemAwareHandleMap[[]types.Datum]()
}
tbl := e.tblID2table[content.TblID]
oldData := row[content.Start:content.End]
newTableData := newData[content.Start:content.End]
if v, ok := e.mergedRowData[content.TblID].Get(handle); ok {
mergedData = v
for i, flag := range flags {
if tbl.WritableCols()[i].IsGenerated() != mergeGenerated {
continue
}
mergedData[i].Copy(&oldData[i])
if flag >= 0 {
newTableData[i].Copy(&mergedData[i])
} else {
mergedData[i].Copy(&newTableData[i])
}
}
} else {
mergedData = append([]types.Datum{}, newTableData...)
}
memDelta := e.mergedRowData[content.TblID].Set(handle, mergedData)
memDelta += types.EstimatedMemUsage(mergedData, 1) + int64(handle.ExtraMemSize())
e.memTracker.Consume(memDelta)
}
return nil
Script executed:
# Check if there are any tests for the partition handle collision scenario
rg -n "EXCHANGE PARTITION" . --type go -A 10 | grep -i "update\|test" | head -20Repository: pingcap/tidb
Repository: pingcap/tidb
Exit code: 0
stdout:
./parser/parser_test.go:3089: {"alter table t with validation, exchange partition p with table nt without validation;", true, "ALTER TABLE `t` WITH VALIDATION, EXCHANGE PARTITION `p` WITH TABLE `nt` WITHOUT VALIDATION"},
./parser/parser_test.go:3090: {"alter table t exchange partition p with table nt with validation;", true, "ALTER TABLE `t` EXCHANGE PARTITION `p` WITH TABLE `nt`"},
./parser/parser_test.go-3091-
./parser/parser_test.go-3092- // For reorganize partition statement
./parser/parser_test.go-3093- {"alter table t reorganize partition;", true, "ALTER TABLE `t` REORGANIZE PARTITION"},
./parser/parser_test.go-3094- {"alter table t reorganize partition local;", true, "ALTER TABLE `t` REORGANIZE PARTITION NO_WRITE_TO_BINLOG"},
./parser/parser_test.go-3095- {"alter table t reorganize partition no_write_to_binlog;", true, "ALTER TABLE `t` REORGANIZE PARTITION NO_WRITE_TO_BINLOG"},
./parser/parser_test.go-3096- {"ALTER TABLE members REORGANIZE PARTITION n0 INTO (PARTITION s0 VALUES LESS THAN (1960), PARTITION s1 VALUES LESS THAN (1970));", true, "ALTER TABLE `members` REORGANIZE PARTITION `n0` INTO (PARTITION `s0` VALUES LESS THAN (1960), PARTITION `s1` VALUES LESS THAN (1970))"},
./parser/parser_test.go-3097- {"ALTER TABLE members REORGANIZE PARTITION LOCAL n0 INTO (PARTITION s0 VALUES LESS THAN (1960), PARTITION s1 VALUES LESS THAN (1970));", true, "ALTER TABLE `members` REORGANIZE PARTITION NO_WRITE_TO_BINLOG `n0` INTO (PARTITION `s0` VALUES LESS THAN (1960), PARTITION `s1` VALUES LESS THAN (1970))"},
./parser/parser_test.go-3098- {"ALTER TABLE members REORGANIZE PARTITION p1,p2,p3 INTO ( PARTITION s0 VALUES LESS THAN (1960), PARTITION s1 VALUES LESS THAN (1970));", true, "ALTER TABLE `members` REORGANIZE PARTITION `p1`,`p2`,`p3` INTO (PARTITION `s0` VALUES LESS THAN (1960), PARTITION `s1` VALUES LESS THAN (1970))"},
./parser/parser_test.go-3099- {"alter table t reorganize partition remove partition;", false, ""},
./parser/parser_test.go-3100- {"alter table t reorganize partition no_write_to_binlog remove into (partition p0 VALUES LESS THAN (1991));", true, "ALTER TABLE `t` REORGANIZE PARTITION NO_WRITE_TO_BINLOG `remove` INTO (PARTITION `p0` VALUES LESS THAN (1991))"},
./ddl/db_partition_test.go:2011: tk.MustExec("ALTER TABLE e EXCHANGE PARTITION p0 WITH TABLE e2")
./ddl/db_partition_test.go-2012- tk.MustQuery("select * from e2").Check(testkit.Rows("16"))
./ddl/db_partition_test.go-2013- tk.MustQuery("select * from e").Check(testkit.Rows("1669", "337", "2005"))
./ddl/db_partition_test.go-2014- // validation test for range partition
./ddl/db_partition_test.go:2015: tk.MustGetErrCode("ALTER TABLE e EXCHANGE PARTITION p1 WITH TABLE e2", errno.ErrRowDoesNotMatchPartition)
./ddl/db_partition_test.go:2016: tk.MustGetErrCode("ALTER TABLE e EXCHANGE PARTITION p2 WITH TABLE e2", errno.ErrRowDoesNotMatchPartition)
./ddl/db_partition_test.go:2017: tk.MustGetErrCode("ALTER TABLE e EXCHANGE PARTITION p3 WITH TABLE e2", errno.ErrRowDoesNotMatchPartition)
./ddl/db_partition_test.go-2018-
Script executed:
# Verify the comment about partition duplicate rowid
rg -n "duplicate.*_tidb_rowid\|_tidb_rowid.*duplicate" . --type go -B 2 -A 2Repository: pingcap/tidb
Repository: pingcap/tidb
Exit code: 1
Script executed:
# Check if there's a test file for update executor
fd update.*test\.go executor/Repository: pingcap/tidb
Repository: pingcap/tidb
Exit code: 0
stdout:
executor/update_test.go
Script executed:
# Look at what skipMultipleChangesOnSameRow actually does - is it the complete fix?
rg -n "skipMultipleChangesOnSameRow" executor/update.go -B 2 -A 2Repository: pingcap/tidb
Repository: pingcap/tidb
Exit code: 0
stdout:
85- // Partitioned tables can have duplicate _tidb_rowid between different partitions
86- // due to EXCHANGE PARTITION, if so, do not optimize skipping rows with multiple changes
87: skipMultipleChangesOnSameRow := true
88- tbl := e.tblID2table[content.TblID]
89- if _, ok := tbl.(table.PartitionedTable); ok {
90- if !tbl.Meta().HasClusteredIndex() {
91: skipMultipleChangesOnSameRow = false
92- }
93- }
--
117- changed, ok := e.updatedRowKeys[content.Start].Get(handle)
118- if ok {
119: e.changed = append(e.changed, changed && skipMultipleChangesOnSameRow)
120- e.matches = append(e.matches, false)
121- } else {
The handle-collision fix is still incomplete.
This patch disables the skipMultipleChangesOnSameRow optimization for partitioned tables without clustered index, but the actual merging and row-deduplication logic remains partition-unaware.
When a partitioned nonclustered table has duplicate _tidb_rowid after EXCHANGE PARTITION and multiUpdateOnSameTable[TblID] is true:
- On the second collision,
e.changed[i]is set tofalse(line 119) - This causes
merge()to still execute (line 144: it skips only ife.changed[i]is true) - The merge then coalesces data from two different physical rows via
mergedRowData[TblID].Get(handle)(line 157) - Additionally,
e.matches[i]remainsfalseon collision (line 120), so theRows matchedcounter is not incremented for the second row
To fix this correctly, either:
- Extend the row identity to be partition-aware in
mergedRowDataandupdatedRowKeyskeys, or - Skip both
merge()and theexec()counting path when a partition-collision is detected on nonclustered tables
Also applies to: 117–120
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@executor/update.go` around lines 85 - 93, The current fix toggles
skipMultipleChangesOnSameRow but leaves merge() and exec() logic
partition-unaware, causing mergedRowData and updatedRowKeys to coalesce
different physical rows when _tidb_rowid collides; update the code so that when
handling partitioned non-clustered tables (where skipMultipleChangesOnSameRow is
set false) you either 1) make mergedRowData and updatedRowKeys keys
partition-aware (e.g., include partition ID + handle) wherever
mergedRowData[TblID].Get(handle) and updatedRowKeys[TblID] are used so distinct
physical rows do not collapse, or 2) on detecting a collision (where
e.changed[i] is set false and multiUpdateOnSameTable[TblID] is true) explicitly
skip both merge() and the exec()/matching-count path (avoid incrementing
e.matches and the Rows matched counter) for that conflicting row; implement one
of these two fixes touching the logic around skipMultipleChangesOnSameRow,
merge(), mergedRowData, updatedRowKeys, e.changed and e.matches so partition
collisions are handled correctly.
Comment on lines
+62
to
+63
| // SetRetireOwnerHook will be called after retiring the owner. | ||
| SetRetireOwnerHook(hook func()) |
There was a problem hiding this comment.
Clear m.elec before invoking the retire hook.
The new contract says this hook runs after retirement, but Line 196 still calls it while IsOwner() can return true. That makes hook behavior order-dependent and easy to misuse.
Suggested fix
func (m *ownerManager) RetireOwner() {
- if m.retireOwnerHook != nil {
- m.retireOwnerHook()
- }
atomic.StorePointer(&m.elec, nil)
+ if m.retireOwnerHook != nil {
+ m.retireOwnerHook()
+ }
}As per coding guidelines "Comments SHOULD explain non-obvious intent, constraints, invariants, concurrency guarantees, SQL/compatibility contracts, or important performance trade-offs, and SHOULD NOT restate what the code already makes clear."
Also applies to: 195-199
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@owner/manager.go` around lines 62 - 63, Clear the manager's election state
(m.elec) before invoking the retire hook so that any call to IsOwner() within
the hook returns false; specifically, in the retire sequence in owner/manager.go
update the order so m.elec is set to nil/cleared prior to calling the function
provided to SetRetireOwnerHook, and adjust the surrounding comment to document
the invariant ("hook runs after m.elec cleared so IsOwner() == false"); ensure
the same change is applied to the other retire-related invocation sites
referenced around the current retire logic.
Comment on lines
+105
to
+108
| // SetRetireOwnerHook implements Manager.SetRetireOwnerHook interface. | ||
| func (m *mockManager) SetRetireOwnerHook(hook func()) { | ||
| m.retireHook = hook | ||
| } |
There was a problem hiding this comment.
Invoke the registered retire hook from RetireOwner().
This setter stores the callback, but mockManager.RetireOwner() still ignores it. That leaves mock/local-store flows out of sync with owner/manager.go, so ddl.Start()'s new owner-retirement cleanup never runs under the mock.
Proposed fix
func (m *mockManager) RetireOwner() {
+ if m.retireHook != nil {
+ m.retireHook()
+ }
atomic.StoreInt32(&m.owner, 0)
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| // SetRetireOwnerHook implements Manager.SetRetireOwnerHook interface. | |
| func (m *mockManager) SetRetireOwnerHook(hook func()) { | |
| m.retireHook = hook | |
| } | |
| func (m *mockManager) RetireOwner() { | |
| if m.retireHook != nil { | |
| m.retireHook() | |
| } | |
| atomic.StoreInt32(&m.owner, 0) | |
| } |
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@owner/mock.go` around lines 105 - 108, The mockManager stores a retire
callback via SetRetireOwnerHook but RetireOwner() currently ignores it; update
mockManager.RetireOwner to check m.retireHook and invoke it (e.g., if
m.retireHook != nil { m.retireHook() }) so the registered cleanup runs under the
mock; ensure you reference the mockManager.retireHook field and call it from the
RetireOwner method, preserving any existing return behavior.
HunDunDM
force-pushed
the
hundundm/260409-empty-region-6511
branch
from
April 13, 2026 08:52
752e495 to
7841750
Compare
There was a problem hiding this comment.
Actionable comments posted: 1
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (2)
store/gcworker/gc_worker.go (1)
723-739:⚠️ Potential issue | 🟠 MajorDon't complete the delete-range task before the force-merge report succeeds.
CompleteDeleteRangeremoves this task from futuredeleteRangesruns, butdoGCForceMergeRangesis only best-effort afterwards. A transient PD/network error here permanently loses the new force-merge side effect, becauseredoDeleteRangesnever retries it.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@store/gcworker/gc_worker.go` around lines 723 - 739, The delete-range task is being marked complete before the best-effort force-merge report runs, which can permanently lose the force-merge side-effect on transient failures; change the call order so util.CompleteDeleteRange(se, r) is only invoked after w.doGCForceMergeRanges(ctx, se, r, gcForceMergeTableCache) succeeds, and if doGCForceMergeRanges returns an error, propagate or return that error (do not call CompleteDeleteRange) so redoDeleteRanges can retry; update logging/metrics accordingly around the CompleteDeleteRange and w.doGCForceMergeRanges calls.ddl/delete_range.go (1)
250-257:⚠️ Potential issue | 🟠 MajorMake force-merge reporting durable before deleting the only retry record.
CompleteDeleteRangeruns first, and any laterreportForceMergeRangesfailure is just logged. That means a transient PD/infosync error — or the new shutdown cancellation path onctx— permanently drops the force-merge report for this task because themysql.gc_delete_rangerow is already gone. Please either make completion contingent on successful reporting, or persist a separate retryable state before removing the task.Also applies to: 274-287
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@ddl/delete_range.go` around lines 250 - 257, The current flow calls CompleteDeleteRange(sctx, r) before dr.reportForceMergeRanges(ctx, sctx, r, gcForceMergeTableCache), risking loss of the force-merge report if reporting fails; either reorder so reportForceMergeRanges is called and succeeds (and any transient failures are retried/persisted) before calling CompleteDeleteRange, or add a durable retryable flag/column in the mysql.gc_delete_range row and set that flag (persisting the intent to report) prior to calling CompleteDeleteRange, then clear it only after reportForceMergeRanges returns success; update the logic in the delete-range completion path (the block using CompleteDeleteRange and reportForceMergeRanges) and mirror the same fix for the analogous block around lines 274–287.
♻️ Duplicate comments (1)
domain/infosync/info.go (1)
425-432:⚠️ Potential issue | 🟠 MajorRetries still reuse a consumed request body.
bodyFactorycloses over the originalbodyreader, so a retry to the next PD address still reuses the same stream. Any caller that still goes throughdoRequestwith a non-nil body can send an empty/partial payload after the first attempt.Run this to inspect the implementation and list remaining call sites that still route through
doRequest:#!/bin/bash set -euo pipefail echo "Current doRequest implementation:" sed -n '425,442p' domain/infosync/info.go echo echo "All doRequest call sites:" rg -nP '\bdoRequest\s*\(' --type go echo echo "Potential non-nil-body call sites (manual inspection target):" python - <<'PY' import pathlib, re for path in pathlib.Path(".").rglob("*.go"): text = path.read_text(errors="ignore") for m in re.finditer(r'\bdoRequest\s*\((.*?)\)', text, re.S): args = " ".join(m.group(1).split()) if re.search(r',\s*nil\s*$', args): continue line = text.count("\n", 0, m.start()) + 1 print(f"{path}:{line}: {args[:220]}") PY🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@domain/infosync/info.go` around lines 425 - 432, doRequest currently builds bodyFactory that closes over the original io.Reader so retries reuse a consumed stream; fix by materializing the request payload once and returning fresh readers for each retry: when body != nil, read the full payload (io.ReadAll) into a []byte buffer and set bodyFactory to return bytes.NewReader(buf) on each call (so doRequestWithBodyFactory and retry logic get a fresh reader per attempt); reference symbols: doRequest, bodyFactory, doRequestWithBodyFactory. Ensure you propagate read errors and only buffer when body != nil.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@domain/pkdb_force_merge.go`:
- Around line 104-121: The helper loadOrInitMergeEmptyRegionsMinTableID
currently uses context.Background() when calling kv.RunInNewTxn which prevents
cancellation from mergeEmptyRegionsLoop; change the function signature to accept
a ctx (e.g., ctx context.Context) and propagate that ctx into kv.RunInNewTxn
(using kv.WithInternalSourceType(ctx, ...)) so the transaction honors the
caller's cancellation. Update all callers (notably mergeEmptyRegionsLoop) to
pass their loop context into loadOrInitMergeEmptyRegionsMinTableID, and ensure
error paths and returns remain unchanged.
---
Outside diff comments:
In `@ddl/delete_range.go`:
- Around line 250-257: The current flow calls CompleteDeleteRange(sctx, r)
before dr.reportForceMergeRanges(ctx, sctx, r, gcForceMergeTableCache), risking
loss of the force-merge report if reporting fails; either reorder so
reportForceMergeRanges is called and succeeds (and any transient failures are
retried/persisted) before calling CompleteDeleteRange, or add a durable
retryable flag/column in the mysql.gc_delete_range row and set that flag
(persisting the intent to report) prior to calling CompleteDeleteRange, then
clear it only after reportForceMergeRanges returns success; update the logic in
the delete-range completion path (the block using CompleteDeleteRange and
reportForceMergeRanges) and mirror the same fix for the analogous block around
lines 274–287.
In `@store/gcworker/gc_worker.go`:
- Around line 723-739: The delete-range task is being marked complete before the
best-effort force-merge report runs, which can permanently lose the force-merge
side-effect on transient failures; change the call order so
util.CompleteDeleteRange(se, r) is only invoked after
w.doGCForceMergeRanges(ctx, se, r, gcForceMergeTableCache) succeeds, and if
doGCForceMergeRanges returns an error, propagate or return that error (do not
call CompleteDeleteRange) so redoDeleteRanges can retry; update logging/metrics
accordingly around the CompleteDeleteRange and w.doGCForceMergeRanges calls.
---
Duplicate comments:
In `@domain/infosync/info.go`:
- Around line 425-432: doRequest currently builds bodyFactory that closes over
the original io.Reader so retries reuse a consumed stream; fix by materializing
the request payload once and returning fresh readers for each retry: when body
!= nil, read the full payload (io.ReadAll) into a []byte buffer and set
bodyFactory to return bytes.NewReader(buf) on each call (so
doRequestWithBodyFactory and retry logic get a fresh reader per attempt);
reference symbols: doRequest, bodyFactory, doRequestWithBodyFactory. Ensure you
propagate read errors and only buffer when body != nil.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro
Run ID: 890cfa49-f19a-450c-8fad-f2ee804b4b72
📒 Files selected for processing (22)
ddl/BUILD.bazelddl/delete_range.goddl/pkdb_force_merge.goddl/pkdb_force_merge_test.godomain/BUILD.bazeldomain/domain.godomain/infosync/BUILD.bazeldomain/infosync/info.godomain/infosync/pkdb_force_merge.godomain/infosync/pkdb_force_merge_test.godomain/pkdb_force_merge.godomain/pkdb_force_merge_test.gometa/meta.gometa/meta_test.goserver/http_handler.goserver/http_handler_test.goserver/http_status.gosessionctx/variable/BUILD.bazelsessionctx/variable/pkdb_force_merge_test.gosessionctx/variable/sysvar.gosessionctx/variable/tidb_vars.gostore/gcworker/gc_worker.go
✅ Files skipped from review due to trivial changes (5)
- sessionctx/variable/BUILD.bazel
- domain/BUILD.bazel
- ddl/BUILD.bazel
- domain/pkdb_force_merge_test.go
- ddl/pkdb_force_merge.go
🚧 Files skipped from review as they are similar to previous changes (7)
- server/http_status.go
- sessionctx/variable/tidb_vars.go
- meta/meta_test.go
- server/http_handler.go
- meta/meta.go
- domain/infosync/pkdb_force_merge.go
- sessionctx/variable/sysvar.go
Comment on lines
+104
to
+121
| func (do *Domain) loadOrInitMergeEmptyRegionsMinTableID() (int64, error) { | ||
| minTableID := mergeEmptyRegionsInitMinTableID | ||
| err := kv.RunInNewTxn(kv.WithInternalSourceType(context.Background(), kv.InternalTxnDDL), do.store, true, func(ctx context.Context, txn kv.Transaction) error { | ||
| t := meta.NewMeta(txn) | ||
| tableID, ok, err := t.GetMergeEmptyRegionsMinTableID() | ||
| if err != nil { | ||
| return errors.Trace(err) | ||
| } | ||
| if !ok || tableID < mergeEmptyRegionsInitMinTableID { | ||
| tableID = mergeEmptyRegionsInitMinTableID | ||
| if err := t.SetMergeEmptyRegionsMinTableID(tableID); err != nil { | ||
| return errors.Trace(err) | ||
| } | ||
| } | ||
| minTableID = tableID | ||
| return nil | ||
| }) | ||
| return minTableID, errors.Trace(err) |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
find . -name "pkdb_force_merge.go" -type fRepository: pingcap/tidb
Length of output: 148
🏁 Script executed:
cat -n ./domain/pkdb_force_merge.go | head -200Repository: pingcap/tidb
Length of output: 6472
🏁 Script executed:
# Search for other uses of kv.RunInNewTxn to understand the pattern
rg -A 3 'kv\.RunInNewTxn.*context\.Background' --type go | head -80Repository: pingcap/tidb
Length of output: 5293
🏁 Script executed:
# Look for other background loops in domain package to see how they handle context
rg -B 5 -A 15 'func \(do \*Domain\).*Loop\(ctx context\.Context\)' domain/ --type go | head -150Repository: pingcap/tidb
Length of output: 1220
🏁 Script executed:
# Check what other background loops exist in domain.go
rg 'func \(do \*Domain\).*Loop\(' domain/*.go --type go -B 2 -A 10 | head -200Repository: pingcap/tidb
Length of output: 9868
🏁 Script executed:
# Check how mergeEmptyRegionsLoop is called and what context is passed
rg 'mergeEmptyRegionsLoop' domain/ --type go -B 5 -A 5Repository: pingcap/tidb
Length of output: 1399
🏁 Script executed:
# Check the initialization of Domain to see what context is passed to mergeEmptyRegionsLoop
rg -B 15 'go do\.mergeEmptyRegionsLoop' domain/domain.goRepository: pingcap/tidb
Length of output: 523
🏁 Script executed:
# Find where Domain.EnableAutoAnalyze or similar initialization happens to understand the context lifecycle
rg -B 20 'go do\.mergeEmptyRegionsLoop' domain/domain.go | head -50Repository: pingcap/tidb
Length of output: 706
🏁 Script executed:
# Find the function that contains this code to understand the ctx parameter
rg -B 30 'go do\.mergeEmptyRegionsLoop' domain/domain.go | grep -E '(func|ctx context)' | head -10Repository: pingcap/tidb
Length of output: 38
🏁 Script executed:
# Get the function signature for the code containing mergeEmptyRegionsLoop
rg -B 50 'go do\.mergeEmptyRegionsLoop' domain/domain.go | grep -A 50 'func.*ctx context.Context'Repository: pingcap/tidb
Length of output: 38
🏁 Script executed:
# Find the function name and signature
cat -n domain/domain.go | grep -B 100 'go do\.mergeEmptyRegionsLoop' | grep 'func ' | tail -1Repository: pingcap/tidb
Length of output: 38
🏁 Script executed:
# Find the line number of mergeEmptyRegionsLoop call
grep -n 'go do\.mergeEmptyRegionsLoop' domain/domain.goRepository: pingcap/tidb
Length of output: 94
🏁 Script executed:
# Get context around line 1098 to find the function signature
sed -n '1060,1100p' domain/domain.goRepository: pingcap/tidb
Length of output: 1239
🏁 Script executed:
# Get the function signature
sed -n '1040,1100p' domain/domain.go | head -20Repository: pingcap/tidb
Length of output: 716
🏁 Script executed:
# Find the function definition that contains line 1098
sed -n '1000,1098p' domain/domain.go | grep -E '(^func|^})' | tail -5Repository: pingcap/tidb
Length of output: 38
🏁 Script executed:
# Get the function signature by looking backward from line 1098
awk 'NR<=1098 {line=$0} NR==1098 {for(i=NR; i>=1; i--) {getline < FILENAME; if(/^func/) {print; break}}}' domain/domain.goRepository: pingcap/tidb
Length of output: 86
Thread the loop context into these metadata transactions.
These helpers discard the caller's ctx and recreate the transaction context from context.Background(). If the store stalls during metadata reads or writes, mergeEmptyRegionsLoop can no longer be canceled promptly and domain shutdown has to wait for these operations to complete anyway. Pass ctx through the helper signatures so kv.RunInNewTxn and its commit path honor loop cancellation.
Proposed direction
-func (do *Domain) doMergeEmptyRegions(ctx context.Context) error {
- minTableID, err := do.loadOrInitMergeEmptyRegionsMinTableID()
+func (do *Domain) doMergeEmptyRegions(ctx context.Context) error {
+ minTableID, err := do.loadOrInitMergeEmptyRegionsMinTableID(ctx)
if err != nil {
return errors.Trace(err)
}
@@
- updated, err := do.storeMergeEmptyRegionsMinTableIDIfUnchanged(minTableID, maxTableID)
+ updated, err := do.storeMergeEmptyRegionsMinTableIDIfUnchanged(ctx, minTableID, maxTableID)-func (do *Domain) loadOrInitMergeEmptyRegionsMinTableID() (int64, error) {
+func (do *Domain) loadOrInitMergeEmptyRegionsMinTableID(ctx context.Context) (int64, error) {
minTableID := mergeEmptyRegionsInitMinTableID
- err := kv.RunInNewTxn(kv.WithInternalSourceType(context.Background(), kv.InternalTxnDDL), do.store, true, func(ctx context.Context, txn kv.Transaction) error {
+ err := kv.RunInNewTxn(kv.WithInternalSourceType(ctx, kv.InternalTxnDDL), do.store, true, func(ctx context.Context, txn kv.Transaction) error {
t := meta.NewMeta(txn)-func (do *Domain) storeMergeEmptyRegionsMinTableID(tableID int64) error {
+func (do *Domain) storeMergeEmptyRegionsMinTableID(ctx context.Context, tableID int64) error {
@@
- return errors.Trace(kv.RunInNewTxn(kv.WithInternalSourceType(context.Background(), kv.InternalTxnDDL), do.store, true, func(ctx context.Context, txn kv.Transaction) error {
+ return errors.Trace(kv.RunInNewTxn(kv.WithInternalSourceType(ctx, kv.InternalTxnDDL), do.store, true, func(ctx context.Context, txn kv.Transaction) error {
return errors.Trace(meta.NewMeta(txn).SetMergeEmptyRegionsMinTableID(tableID))
}))
}
-func (do *Domain) storeMergeEmptyRegionsMinTableIDIfUnchanged(expectedTableID, nextTableID int64) (bool, error) {
+func (do *Domain) storeMergeEmptyRegionsMinTableIDIfUnchanged(ctx context.Context, expectedTableID, nextTableID int64) (bool, error) {
@@
- err := kv.RunInNewTxn(kv.WithInternalSourceType(context.Background(), kv.InternalTxnDDL), do.store, true, func(ctx context.Context, txn kv.Transaction) error {
+ err := kv.RunInNewTxn(kv.WithInternalSourceType(ctx, kv.InternalTxnDDL), do.store, true, func(ctx context.Context, txn kv.Transaction) error {
t := meta.NewMeta(txn) func (do *Domain) ResetMergeEmptyRegionsMinTableID() error {
- return errors.Trace(do.storeMergeEmptyRegionsMinTableID(mergeEmptyRegionsInitMinTableID))
+ return errors.Trace(do.storeMergeEmptyRegionsMinTableID(context.Background(), mergeEmptyRegionsInitMinTableID))
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| func (do *Domain) loadOrInitMergeEmptyRegionsMinTableID() (int64, error) { | |
| minTableID := mergeEmptyRegionsInitMinTableID | |
| err := kv.RunInNewTxn(kv.WithInternalSourceType(context.Background(), kv.InternalTxnDDL), do.store, true, func(ctx context.Context, txn kv.Transaction) error { | |
| t := meta.NewMeta(txn) | |
| tableID, ok, err := t.GetMergeEmptyRegionsMinTableID() | |
| if err != nil { | |
| return errors.Trace(err) | |
| } | |
| if !ok || tableID < mergeEmptyRegionsInitMinTableID { | |
| tableID = mergeEmptyRegionsInitMinTableID | |
| if err := t.SetMergeEmptyRegionsMinTableID(tableID); err != nil { | |
| return errors.Trace(err) | |
| } | |
| } | |
| minTableID = tableID | |
| return nil | |
| }) | |
| return minTableID, errors.Trace(err) | |
| func (do *Domain) loadOrInitMergeEmptyRegionsMinTableID(ctx context.Context) (int64, error) { | |
| minTableID := mergeEmptyRegionsInitMinTableID | |
| err := kv.RunInNewTxn(kv.WithInternalSourceType(ctx, kv.InternalTxnDDL), do.store, true, func(ctx context.Context, txn kv.Transaction) error { | |
| t := meta.NewMeta(txn) | |
| tableID, ok, err := t.GetMergeEmptyRegionsMinTableID() | |
| if err != nil { | |
| return errors.Trace(err) | |
| } | |
| if !ok || tableID < mergeEmptyRegionsInitMinTableID { | |
| tableID = mergeEmptyRegionsInitMinTableID | |
| if err := t.SetMergeEmptyRegionsMinTableID(tableID); err != nil { | |
| return errors.Trace(err) | |
| } | |
| } | |
| minTableID = tableID | |
| return nil | |
| }) | |
| return minTableID, errors.Trace(err) |
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@domain/pkdb_force_merge.go` around lines 104 - 121, The helper
loadOrInitMergeEmptyRegionsMinTableID currently uses context.Background() when
calling kv.RunInNewTxn which prevents cancellation from mergeEmptyRegionsLoop;
change the function signature to accept a ctx (e.g., ctx context.Context) and
propagate that ctx into kv.RunInNewTxn (using kv.WithInternalSourceType(ctx,
...)) so the transaction honors the caller's cancellation. Update all callers
(notably mergeEmptyRegionsLoop) to pass their loop context into
loadOrInitMergeEmptyRegionsMinTableID, and ensure error paths and returns remain
unchanged.
Contributor
Author
|
/test check-dev |
Contributor
Author
|
/test unit-test |
|
@HunDunDM: PRs from untrusted users cannot be marked as trusted with DetailsIn response to this:
Instructions for interacting with me using PR comments are available here. If you have questions or suggestions related to my behavior, please file an issue against the kubernetes-sigs/prow repository. |
|
@HunDunDM: PRs from untrusted users cannot be marked as trusted with DetailsIn response to this:
Instructions for interacting with me using PR comments are available here. If you have questions or suggestions related to my behavior, please file an issue against the kubernetes-sigs/prow repository. |
HunDunDM
force-pushed
the
hundundm/260409-empty-region-6511
branch
from
April 13, 2026 10:09
7841750 to
981deea
Compare
There was a problem hiding this comment.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
ddl/delete_range.go (1)
251-257:⚠️ Potential issue | 🟠 MajorDon't acknowledge the delete-range task before the PD report succeeds.
util.CompleteDeleteRangeruns beforereportForceMergeRanges, and any report failure is only logged. IfGetHistoryJobByIDorinfosync.AddForceMergeRangesfails once—orctxis canceled during shutdown—thegc_delete_rangerow is already gone, so this force-merge request is lost permanently. That is especially risky here becauseAddForceMergeRangesstops on the first failed batch, and the logical-table cache may already suppress later emission for the same table in this scan. Please keep the task retriable until reporting succeeds, or persist separate retry state for the force-merge side effect.As per coding guidelines, "Keep error handling actionable and contextual; avoid silently swallowing errors."
Also applies to: 274-287
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@ddl/delete_range.go` around lines 251 - 257, The current flow calls util.CompleteDeleteRange(sctx, r) before dr.reportForceMergeRanges(...), which can permanently lose the force-merge request if reportForceMergeRanges (or underlying GetHistoryJobByID / infosync.AddForceMergeRanges) fails or context is canceled; change the logic so the delete-range task is not acknowledged/removed until reportForceMergeRanges returns success — either by moving the CompleteDeleteRange call after a successful report or by persisting retry state for the force-merge side effect so the task remains retriable; ensure functions referenced (util.CompleteDeleteRange, dr.reportForceMergeRanges, GetHistoryJobByID, infosync.AddForceMergeRanges) are updated accordingly to propagate/report errors so failures are actionable and the task is retried rather than silently dropped.
♻️ Duplicate comments (1)
domain/infosync/info.go (1)
425-432:⚠️ Potential issue | 🟠 Major
doRequeststill reuses a consumed body across retries.Line 428 closes over the original
io.Reader, so a retry to the next PD address re-sends the same exhausted reader. That still breaks current callers such asdomain/infosync/schedule_manager.go:60-65,domain/infosync/placement_manager.go:76-79, anddomain/infosync/tiflash_manager.go:244-250, all of which pass one-shot readers intodoRequest(...). Buffer the body once here and delegate todoRequestWithBodyBytes(...)so each attempt gets a fresh reader.💡 Proposed fix
func doRequest(ctx context.Context, apiName string, addrs []string, route, method string, body io.Reader) ([]byte, error) { - var bodyFactory func() io.Reader - if body != nil { - bodyFactory = func() io.Reader { - return body - } - } - return doRequestWithBodyFactory(ctx, apiName, addrs, route, method, bodyFactory) + if body == nil { + return doRequestWithBodyFactory(ctx, apiName, addrs, route, method, nil) + } + bodyBytes, err := io.ReadAll(body) + if err != nil { + return nil, errors.Trace(err) + } + return doRequestWithBodyBytes(ctx, apiName, addrs, route, method, bodyBytes) }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@domain/infosync/info.go` around lines 425 - 432, doRequest currently closes over and reuses a one-shot io.Reader across retries (in doRequestWithBodyFactory), causing exhausted bodies on subsequent attempts; fix by reading and buffering the entire body bytes once in doRequest when body != nil, then call doRequestWithBodyBytes(ctx, apiName, addrs, route, method, bodyBytes) so each retry gets a fresh reader/reader factory derived from the buffered bytes; update references to doRequestWithBodyFactory to use doRequestWithBodyBytes and ensure nil body handling remains unchanged.
🧹 Nitpick comments (1)
sessionctx/variable/tidb_vars.go (1)
1224-1225: Add a doc comment for the new exported flag.
EnableDropTableForceMergeis a new exported package-global. A short comment explaining that it is the in-memory mirror oftidb_enable_drop_table_force_mergewould make its ownership and intended readers/writers clearer.As per coding guidelines, "Keep exported-symbol doc comments, and prefer semantic constraints over name restatement."
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@sessionctx/variable/tidb_vars.go` around lines 1224 - 1225, Add a concise exported doc comment for the package-global EnableDropTableForceMerge variable explaining that it is the in-memory mirror of the session/system variable tidb_enable_drop_table_force_merge, who should read/write it (e.g., system session management code), and any relevant semantic constraint (default behavior or expected lifecycle) rather than restating the name; place the comment immediately above the EnableDropTableForceMerge declaration near EnableConcurrentDDL to satisfy exported-symbol documentation guidelines.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Outside diff comments:
In `@ddl/delete_range.go`:
- Around line 251-257: The current flow calls util.CompleteDeleteRange(sctx, r)
before dr.reportForceMergeRanges(...), which can permanently lose the
force-merge request if reportForceMergeRanges (or underlying GetHistoryJobByID /
infosync.AddForceMergeRanges) fails or context is canceled; change the logic so
the delete-range task is not acknowledged/removed until reportForceMergeRanges
returns success — either by moving the CompleteDeleteRange call after a
successful report or by persisting retry state for the force-merge side effect
so the task remains retriable; ensure functions referenced
(util.CompleteDeleteRange, dr.reportForceMergeRanges, GetHistoryJobByID,
infosync.AddForceMergeRanges) are updated accordingly to propagate/report errors
so failures are actionable and the task is retried rather than silently dropped.
---
Duplicate comments:
In `@domain/infosync/info.go`:
- Around line 425-432: doRequest currently closes over and reuses a one-shot
io.Reader across retries (in doRequestWithBodyFactory), causing exhausted bodies
on subsequent attempts; fix by reading and buffering the entire body bytes once
in doRequest when body != nil, then call doRequestWithBodyBytes(ctx, apiName,
addrs, route, method, bodyBytes) so each retry gets a fresh reader/reader
factory derived from the buffered bytes; update references to
doRequestWithBodyFactory to use doRequestWithBodyBytes and ensure nil body
handling remains unchanged.
---
Nitpick comments:
In `@sessionctx/variable/tidb_vars.go`:
- Around line 1224-1225: Add a concise exported doc comment for the
package-global EnableDropTableForceMerge variable explaining that it is the
in-memory mirror of the session/system variable
tidb_enable_drop_table_force_merge, who should read/write it (e.g., system
session management code), and any relevant semantic constraint (default behavior
or expected lifecycle) rather than restating the name; place the comment
immediately above the EnableDropTableForceMerge declaration near
EnableConcurrentDDL to satisfy exported-symbol documentation guidelines.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro
Run ID: d45a1bbe-8a56-458c-9bb9-dd3b6833d838
📒 Files selected for processing (22)
ddl/BUILD.bazelddl/delete_range.goddl/pkdb_force_merge.goddl/pkdb_force_merge_test.godomain/BUILD.bazeldomain/domain.godomain/infosync/BUILD.bazeldomain/infosync/info.godomain/infosync/pkdb_force_merge.godomain/infosync/pkdb_force_merge_test.godomain/pkdb_force_merge.godomain/pkdb_force_merge_test.gometa/meta.gometa/meta_test.goserver/http_handler.goserver/http_handler_test.goserver/http_status.gosessionctx/variable/BUILD.bazelsessionctx/variable/pkdb_force_merge_test.gosessionctx/variable/sysvar.gosessionctx/variable/tidb_vars.gostore/gcworker/gc_worker.go
✅ Files skipped from review due to trivial changes (8)
- sessionctx/variable/BUILD.bazel
- domain/BUILD.bazel
- ddl/BUILD.bazel
- domain/infosync/BUILD.bazel
- domain/infosync/pkdb_force_merge.go
- sessionctx/variable/sysvar.go
- sessionctx/variable/pkdb_force_merge_test.go
- ddl/pkdb_force_merge.go
🚧 Files skipped from review as they are similar to previous changes (8)
- domain/domain.go
- server/http_status.go
- server/http_handler_test.go
- meta/meta_test.go
- store/gcworker/gc_worker.go
- domain/pkdb_force_merge_test.go
- domain/infosync/pkdb_force_merge_test.go
- domain/pkdb_force_merge.go
HunDunDM
force-pushed
the
hundundm/260409-empty-region-6511
branch
from
April 13, 2026 10:24
981deea to
af7312b
Compare
Contributor
Author
|
/test mysql-test |
|
@HunDunDM: PRs from untrusted users cannot be marked as trusted with DetailsIn response to this:
Instructions for interacting with me using PR comments are available here. If you have questions or suggestions related to my behavior, please file an issue against the kubernetes-sigs/prow repository. |
Contributor
Author
|
/test mysql-test |
|
@HunDunDM: PRs from untrusted users cannot be marked as trusted with DetailsIn response to this:
Instructions for interacting with me using PR comments are available here. If you have questions or suggestions related to my behavior, please file an issue against the kubernetes-sigs/prow repository. |
Contributor
Author
|
/test check-dev |
|
@HunDunDM: PRs from untrusted users cannot be marked as trusted with DetailsIn response to this:
Instructions for interacting with me using PR comments are available here. If you have questions or suggestions related to my behavior, please file an issue against the kubernetes-sigs/prow repository. |
Contributor
Author
|
/test mysql-test |
|
@HunDunDM: PRs from untrusted users cannot be marked as trusted with DetailsIn response to this:
Instructions for interacting with me using PR comments are available here. If you have questions or suggestions related to my behavior, please file an issue against the kubernetes-sigs/prow repository. |
Contributor
Author
|
/test mysql-test |
|
@HunDunDM: PRs from untrusted users cannot be marked as trusted with DetailsIn response to this:
Instructions for interacting with me using PR comments are available here. If you have questions or suggestions related to my behavior, please file an issue against the kubernetes-sigs/prow repository. |
Contributor
Author
|
/test mysql-test |
|
@HunDunDM: PRs from untrusted users cannot be marked as trusted with DetailsIn response to this:
Instructions for interacting with me using PR comments are available here. If you have questions or suggestions related to my behavior, please file an issue against the kubernetes-sigs/prow repository. |
Contributor
Author
|
/test mysql-test |
|
@HunDunDM: PRs from untrusted users cannot be marked as trusted with DetailsIn response to this:
Instructions for interacting with me using PR comments are available here. If you have questions or suggestions related to my behavior, please file an issue against the kubernetes-sigs/prow repository. |
There was a problem hiding this comment.
♻️ Duplicate comments (1)
domain/pkdb_force_merge.go (1)
112-167:⚠️ Potential issue | 🟠 MajorPropagate the loop context into these checkpoint transactions.
doMergeEmptyRegionsis already wired for cancellation, but Line 114, Line 136, and Line 150 rebuild the transaction context fromcontext.Background(). If the store blocks here,mergeEmptyRegionsLoopstill can't stop promptly during shutdown. Please threadctxthrough these helper signatures and usekv.WithInternalSourceType(ctx, kv.InternalTxnDDL)instead of discarding it.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@domain/pkdb_force_merge.go` around lines 112 - 167, The three helper functions loadOrInitMergeEmptyRegionsMinTableID, storeMergeEmptyRegionsMinTableID, and storeMergeEmptyRegionsMinTableIDIfUnchanged recreate transactions with context.Background(), preventing cancellation; change each signature to accept a context.Context parameter, propagate the ctx from doMergeEmptyRegions (or mergeEmptyRegionsLoop) into their calls, and replace kv.WithInternalSourceType(context.Background(), kv.InternalTxnDDL) with kv.WithInternalSourceType(ctx, kv.InternalTxnDDL) so the checkpoint transactions honor shutdown/cancellation signals.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Duplicate comments:
In `@domain/pkdb_force_merge.go`:
- Around line 112-167: The three helper functions
loadOrInitMergeEmptyRegionsMinTableID, storeMergeEmptyRegionsMinTableID, and
storeMergeEmptyRegionsMinTableIDIfUnchanged recreate transactions with
context.Background(), preventing cancellation; change each signature to accept a
context.Context parameter, propagate the ctx from doMergeEmptyRegions (or
mergeEmptyRegionsLoop) into their calls, and replace
kv.WithInternalSourceType(context.Background(), kv.InternalTxnDDL) with
kv.WithInternalSourceType(ctx, kv.InternalTxnDDL) so the checkpoint transactions
honor shutdown/cancellation signals.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro
Run ID: d2842318-b3e1-4af4-b9ae-4b195256eb82
📒 Files selected for processing (2)
ddl/delete_range.godomain/pkdb_force_merge.go
🚧 Files skipped from review as they are similar to previous changes (1)
- ddl/delete_range.go
…#684) Signed-off-by: HunDunDM <hundundm@gmail.com>
Signed-off-by: HunDunDM <hundundm@gmail.com>
HunDunDM
force-pushed
the
hundundm/260409-empty-region-6511
branch
from
April 14, 2026 07:40
753e26f to
3790020
Compare
|
[FORMAT CHECKER NOTIFICATION] Notice: To remove the 📖 For more info, you can check the "Contribute Code" section in the development guide. |
|
tangenta seems not to be a GitHub user. You need a GitHub account to be able to sign the CLA. If you have already a GitHub account, please add the email address used for this commit to your account. You have signed the CLA already but the status is still pending? Let us recheck it. |
|
@HunDunDM: The following test failed, say
Full PR test history. Your PR dashboard. DetailsInstructions for interacting with me using PR comments are available here. If you have questions or suggestions related to my behavior, please file an issue against the kubernetes-sigs/prow repository. I understand the commands that are listed here. |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
4 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Only for CI
What problem does this PR solve?
Issue Number: close #xxx
Problem Summary:
What changed and how does it work?
Check List
Tests
Side effects
Documentation
Release note
Please refer to Release Notes Language Style Guide to write a quality release note.
Summary by CodeRabbit
New Features
Bug Fixes
Tests