keyspace: refine the split and merge details #6707
There are no files selected for viewing
This file contains 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
This file contains 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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -536,6 +536,16 @@ func (m *GroupManager) SplitKeyspaceGroupByID( | |
| if splitSourceKg.IsMerging() { | ||
| return ErrKeyspaceGroupInMerging(splitSourceID) | ||
| } | ||
| // Build the new keyspace groups for split source and target. | ||
| var startKeyspaceID, endKeyspaceID uint32 | ||
| if len(keyspaceIDRange) >= 2 { | ||
| startKeyspaceID, endKeyspaceID = keyspaceIDRange[0], keyspaceIDRange[1] | ||
| } | ||
| splitSourceKeyspaces, splitTargetKeyspaces, err := buildSplitKeyspaces( | ||
| splitSourceKg.Keyspaces, keyspaces, startKeyspaceID, endKeyspaceID) | ||
| if err != nil { | ||
| return err | ||
| } | ||
| // Check if the source keyspace group has enough replicas. | ||
| if len(splitSourceKg.Members) < utils.DefaultKeyspaceGroupReplicaCount { | ||
| return ErrKeyspaceGroupNotEnoughReplicas | ||
|
|
@@ -548,15 +558,6 @@ func (m *GroupManager) SplitKeyspaceGroupByID( | |
| if splitTargetKg != nil { | ||
| return ErrKeyspaceGroupExists | ||
| } | ||
| // Update the old keyspace group. | ||
| splitSourceKg.Keyspaces = splitSourceKeyspaces | ||
| splitSourceKg.SplitState = &endpoint.SplitState{ | ||
|
|
@@ -606,6 +607,9 @@ func buildSplitKeyspaces( | |
| oldKeyspaceMap[keyspace] = struct{}{} | ||
| } | ||
| for _, keyspace := range new { | ||
| if keyspace == utils.DefaultKeyspaceID { | ||
| return nil, nil, ErrModifyDefaultKeyspace | ||
| } | ||
| if _, ok := oldKeyspaceMap[keyspace]; !ok { | ||
| return nil, nil, ErrKeyspaceNotInKeyspaceGroup | ||
| } | ||
|
|
@@ -618,15 +622,7 @@ func buildSplitKeyspaces( | |
| oldSplit = append(oldSplit, keyspace) | ||
| } | ||
| } | ||
|
There was a problem hiding this comment. I found that There was a problem hiding this comment. We can't remove this check. I specifically add this block to fix issue #6687. root@serverless-cluster-pd-0:/# ./pd-ctl keyspace-group split 0 2 2 2 {"id":2,"user-kind":"basic","members":[{"address":"http://pd-tso-server-0.tso-service.tidb-serverless.svc:2379","priority":200},{"address":"http://pd-tso-server-1.tso-service.tidb-serverless.svc:2379","priority":300}],"keyspaces":[2,2]} There was a problem hiding this comment. It's my fault, since I signed off this change and previous change didn't add unittest. |
||
| return oldSplit, new, nil | ||
| } | ||
| // Split according to the start and end keyspace ID. | ||
| if startKeyspaceID == 0 && endKeyspaceID == 0 { | ||
|
|
@@ -637,6 +633,9 @@ func buildSplitKeyspaces( | |
| newKeyspaceMap = make(map[uint32]struct{}, newNum) | ||
| ) | ||
| for _, keyspace := range old { | ||
| if keyspace == utils.DefaultKeyspaceID { | ||
|
There was a problem hiding this comment. We can't add this check, because it means that we can't split default keyspace group now. |
||
| return nil, nil, ErrModifyDefaultKeyspace | ||
| } | ||
| if startKeyspaceID <= keyspace && keyspace <= endKeyspaceID { | ||
| newSplit = append(newSplit, keyspace) | ||
| newKeyspaceMap[keyspace] = struct{}{} | ||
|
|
@@ -770,7 +769,9 @@ func (m *GroupManager) AllocNodesForKeyspaceGroup(id uint32, desiredReplicaCount | |
| return nil, err | ||
| } | ||
| m.groups[endpoint.StringUserKind(kg.UserKind)].Put(kg) | ||
| log.Info("alloc nodes for keyspace group", | ||
| zap.Uint32("keyspace-group-id", id), | ||
| zap.Reflect("nodes", nodes)) | ||
| return nodes, nil | ||
| } | ||
|
|
||
|
|
@@ -906,20 +907,17 @@ func (m *GroupManager) MergeKeyspaceGroups(mergeTargetID uint32, mergeList []uin | |
| } | ||
| groups[kgID] = kg | ||
| } | ||
| // Build the new keyspaces for the merge target keyspace group. | ||
| mergeTargetKg = groups[mergeTargetID] | ||
| keyspaces := make(map[uint32]struct{}) | ||
| for _, keyspace := range mergeTargetKg.Keyspaces { | ||
| keyspaces[keyspace] = struct{}{} | ||
| } | ||
| for _, kgID := range mergeList { | ||
| kg := groups[kgID] | ||
| for _, keyspace := range kg.Keyspaces { | ||
| keyspaces[keyspace] = struct{}{} | ||
| } | ||
| } | ||
| mergedKeyspaces := make([]uint32, 0, len(keyspaces)) | ||
| for keyspace := range keyspaces { | ||
|
|
@@ -933,7 +931,17 @@ func (m *GroupManager) MergeKeyspaceGroups(mergeTargetID uint32, mergeList []uin | |
| mergeTargetKg.MergeState = &endpoint.MergeState{ | ||
| MergeList: mergeList, | ||
| } | ||
| err = m.store.SaveKeyspaceGroup(txn, mergeTargetKg) | ||
| if err != nil { | ||
| return err | ||
| } | ||
| // Delete the keyspace groups in merge list and move the keyspaces in it to the target keyspace group. | ||
| for _, kgID := range mergeList { | ||
| if err := m.store.DeleteKeyspaceGroup(txn, kgID); err != nil { | ||
| return err | ||
| } | ||
| } | ||
| return nil | ||
| }); err != nil { | ||
| return err | ||
| } | ||
|
|
@@ -948,7 +956,10 @@ func (m *GroupManager) MergeKeyspaceGroups(mergeTargetID uint32, mergeList []uin | |
|
|
||
| // FinishMergeKeyspaceByID finishes the merging keyspace group by the merge target ID. | ||
| func (m *GroupManager) FinishMergeKeyspaceByID(mergeTargetID uint32) error { | ||
| var ( | ||
| mergeTargetKg *endpoint.KeyspaceGroup | ||
| mergeList []uint32 | ||
| ) | ||
| m.Lock() | ||
| defer m.Unlock() | ||
| if err := m.store.RunInTxn(m.ctx, func(txn kv.Txn) (err error) { | ||
|
|
@@ -974,12 +985,16 @@ func (m *GroupManager) FinishMergeKeyspaceByID(mergeTargetID uint32) error { | |
| return ErrKeyspaceGroupNotInMerging(kgID) | ||
| } | ||
| } | ||
| mergeList = mergeTargetKg.MergeState.MergeList | ||
| mergeTargetKg.MergeState = nil | ||
| return m.store.SaveKeyspaceGroup(txn, mergeTargetKg) | ||
| }); err != nil { | ||
| return err | ||
| } | ||
| // Update the keyspace group cache. | ||
| m.groups[endpoint.StringUserKind(mergeTargetKg.UserKind)].Put(mergeTargetKg) | ||
| log.Info("finish merge keyspace group", | ||
| zap.Uint32("merge-target-id", mergeTargetKg.ID), | ||
| zap.Reflect("merge-list", mergeList)) | ||
| return nil | ||
| } | ||
This file contains 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
This file contains 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
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.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
why move this block?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I adjusted the block to let some errors that could be returned earlier to make the unit test easier to write.