feat: add the WAL messages a shard split is built on - #52583
Open
xiaocai2333 wants to merge 7 commits into
Open
Conversation
sre-ci-robot
added
size/XXL
Contributor
|
@xiaocai2333 This is a feature PR ( How to resolve: |
mergify
Bot
added
do-not-merge/missing-design-doc
kind/feature
Contributor
|
[ci-v2-notice] To rerun ci-v2 checks, comment with:
If you have any questions or requests, please contact @zhikunyao. |
Contributor
xiaocai2333
force-pushed
the
shard-split-wal-messages
branch
from
August 31, 2026 01:55
6335e02 to
1c2e391
Compare
xiaocai2333
added a commit
to xiaocai2333/milvus
that referenced
this pull request
Aug 31, 2026
… changes to it The meta and DDL half of online shard split. A collection stops being described by a channel count and starts describing its own topology. model.Collection gains per-shard ShardInfo -- each shard's residues, lifecycle state and truncate tick -- plus the collection-wide RoutingModulus those residues are taken against and the ShardBy expression naming what gets hashed, all persisted. The shard count is no longer len(VirtualChannelNames): a split retires its sources but the vchannel list only ever grows, so the count is derived from the shards that are actually ROUTABLE. Legacy collections read exactly as before: no residues anywhere and a zero modulus mean hash % shardNum by position, and ShardsNum falls back to the vchannel count for meta written before it was persisted. The modulus is collection-wide rather than per-shard because a split that halves a shard down to its last residue has to double it, which re-expresses every other shard's residues at the same instant. One field committed with the topology says that once; a modulus per shard would say it N times and let the copies disagree. ShardBy is written only when a commit carries one -- only a first split has anything to back-fill, and clearing it on every later commit would drop the expression clients route by. CommitShardSplitRouting is the RPC datacoord uses to apply a routing change. It sends the WHOLE new topology and rootcoord applies it atomically, keyed by shard state so a retry -- or a crash between the commit and the caller's own state advance -- is safe. Before broadcasting, rootcoord derives the routing table the write path will derive, over the same writable-shard filter, and refuses a topology that does not tile the key space: this is the last point at which a bad plan is only a rejected DDL, since a committed gap silently drops the writes of the residues nobody claims and a committed overlap sends one key to two shards. The arrays are parallel and the ShardInfos map is rebuilt from them in lockstep, which is what lets a SHORTER list retire a vchannel later. It takes no collection lock, deliberately: its only caller already holds the collection's exclusive resource key across the whole fence -> create -> commit span, and taking the same key again would deadlock on that caller. collection.shardNum is the user-facing request to change the count. It is declarative: the property records the target and datacoord reconciles toward it, so DescribeCollection's shards_num only reaches the value once the split completes. Deleting the property WITHDRAWS the request, which is how a rehash is cancelled -- and it is answered before the rest of validation, because deletes are applied after sets and a request that both sets and deletes the key in fact asks for nothing. Everything rootcoord can decide synchronously is decided here so the common mistakes land in the AlterCollection response instead of in a background task the caller never sees. Nothing drives any of this yet: no caller invokes CommitShardSplitRouting and no reconciler reads collection.shardNum. This is step 3 of online shard split, on top of the routing abstraction (milvus-io#50495) and the WAL messages (milvus-io#52583); the proxy write path and the split state machine follow. Requires milvus-io/milvus-proto#618. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Signed-off-by: cai.zhang <cai.zhang@zilliz.com>
The vocabulary and the streamingnode half of online shard split. Three
messages, all ExclusiveRequired because each rewrites a vchannel's
registration and nothing may be assigning segments while it does:
SplitShard fences a source vchannel forever. The source streamingnode
auto-flushes every growing segment as of the message's time
tick and embeds their ids in the header, so no separate
ManualFlush is needed, and the append is idempotent: a re-fence
returns SHARD_FENCED carrying the recorded T_switch, so a
coordinator that crashed after fencing recovers it.
CreateVChannel the genesis of a target vchannel, carrying the residues it owns
and a BarrierTimeTick greater than every source's T_switch --
so the new WAL is born strictly after the fence and creation
doubles as activation.
DropVChannel retires a vchannel a split left behind, the inverse of
CreateVChannel. Guarded by the vchannel NAME, not just the
collection id: the shard manager is keyed by collection, one
entry per pchannel, so once a retired source's slot is
reclaimed a later vchannel of the same collection can hold that
entry, and a late or replayed teardown must not delete it.
Each lands in the four places a vchannel's lifecycle is tracked: the shard
interceptor, the shard manager, the recovery storage, and the flusher.
DropVChannel deliberately does not forward to the data sync service -- it is a
V2 message, and forwarding routes it into fromMessageToTsMsgV2, which has no
case for it and panics the process. DropCollection may fall through only
because it is V1, handled by the unmarshaler path.
A split target's key space rides in these messages as the residues it owns,
alongside the routing modulus they are taken against. The modulus is recorded
rather than looked up because a WAL record is permanent while a collection's
modulus moves: a later split that doubles it would leave bare residues
uninterpretable. The routing commit in AlterCollectionMessageUpdates carries the
modulus for the same reason, plus shard_by, which a first split back-fills on a
collection created before it existed.
Also adds SHARD_FENCED (with fenced_time_tick) and ROUTING_STALE streaming
error codes. ROUTING_STALE is reserved and unwired: the routing-version
negotiation was dropped in favour of the SHARD_FENCED reject-refetch loop, and
the code is kept for a possible later fast path.
Nothing appends these yet. This is step 2 of online shard split, after the
routing abstraction (milvus-io#50495); the coordinator that drives them follows.
Requires milvus-io/milvus-proto#618.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Signed-off-by: cai.zhang <cai.zhang@zilliz.com>
xiaocai2333
force-pushed
the
shard-split-wal-messages
branch
from
August 31, 2026 02:28
1c2e391 to
b777439
Compare
xiaocai2333
added a commit
to xiaocai2333/milvus
that referenced
this pull request
Aug 31, 2026
… changes to it The meta and DDL half of online shard split. A collection stops being described by a channel count and starts describing its own topology. model.Collection gains per-shard ShardInfo -- each shard's residues, lifecycle state and truncate tick -- plus the collection-wide RoutingModulus those residues are taken against and the ShardBy expression naming what gets hashed, all persisted. The shard count is no longer len(VirtualChannelNames): a split retires its sources but the vchannel list only ever grows, so the count is derived from the shards that are actually ROUTABLE. Legacy collections read exactly as before: no residues anywhere and a zero modulus mean hash % shardNum by position, and ShardsNum falls back to the vchannel count for meta written before it was persisted. The modulus is collection-wide rather than per-shard because a split that halves a shard down to its last residue has to double it, which re-expresses every other shard's residues at the same instant. One field committed with the topology says that once; a modulus per shard would say it N times and let the copies disagree. ShardBy is written only when a commit carries one -- only a first split has anything to back-fill, and clearing it on every later commit would drop the expression clients route by. CommitShardSplitRouting is the RPC datacoord uses to apply a routing change. It sends the WHOLE new topology and rootcoord applies it atomically, keyed by shard state so a retry -- or a crash between the commit and the caller's own state advance -- is safe. Before broadcasting, rootcoord derives the routing table the write path will derive, over the same writable-shard filter, and refuses a topology that does not tile the key space: this is the last point at which a bad plan is only a rejected DDL, since a committed gap silently drops the writes of the residues nobody claims and a committed overlap sends one key to two shards. The arrays are parallel and the ShardInfos map is rebuilt from them in lockstep, which is what lets a SHORTER list retire a vchannel later. It takes no collection lock, deliberately: its only caller already holds the collection's exclusive resource key across the whole fence -> create -> commit span, and taking the same key again would deadlock on that caller. collection.shardNum is the user-facing request to change the count. It is declarative: the property records the target and datacoord reconciles toward it, so DescribeCollection's shards_num only reaches the value once the split completes. Deleting the property WITHDRAWS the request, which is how a rehash is cancelled -- and it is answered before the rest of validation, because deletes are applied after sets and a request that both sets and deletes the key in fact asks for nothing. Everything rootcoord can decide synchronously is decided here so the common mistakes land in the AlterCollection response instead of in a background task the caller never sees. Nothing drives any of this yet: no caller invokes CommitShardSplitRouting and no reconciler reads collection.shardNum. This is step 3 of online shard split, on top of the routing abstraction (milvus-io#50495) and the WAL messages (milvus-io#52583); the proxy write path and the split state machine follow. Requires milvus-io/milvus-proto#618. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Signed-off-by: cai.zhang <cai.zhang@zilliz.com>
Contributor
Contributor
…segments HandleSplitShard was unreachable. It was defined on msgHandlerImpl and mocked, but never added to the MsgHandler interface, and the dd_node's message switch had no MsgType_SplitShard case -- a switch with no default, so the fence was dropped without a log line. That is not a missing log. The fence carries the ids of the growing segments the source streamingnode sealed at T_switch, and it is their only record: the split appends no ManualFlush, and FlushAndFenceSegmentUntil merely updates the WAL-side assignment state without emitting a Flush message. With no handler the ids never reached the write buffer, so the source shard's last rows were never sealed, never flushed, never reported to DataCoord -- and were discarded when the retired vchannel's data sync service was closed on DropVChannel. Adds the interface method (with the ctx its siblings take, instead of context.Background()), the dd_node case, and a test that fails on exactly that claim: removing the case again makes it red on "the fence must reach the msg handler, or its sealed segments are lost". Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Signed-off-by: cai.zhang <cai.zhang@zilliz.com>
TryIntoStreamingError rebuilt the error from (code, cause) alone, so any field a code attaches was silently zeroed on the way back. SHARD_FENCED attaches T_switch in FencedTimeTick, and the whole crash-recovery story of the split write switch rests on reading it back: a re-fence is rejected with the recorded T_switch so a coordinator that crashed after fencing recovers it. The loss was invisible in-process, where the error IS the *StreamingError and keeps every field, and the existing test mocked RawAppend to return NewShardFenced directly -- so it verified the transform, not the input. Over the wire, i.e. in every cluster deployment where the coordinator and the streamingnode are separate processes, T_switch came back as 0. Clones the whole detail message instead, and adds a test that goes through a real grpc status round trip. Also maps SHARD_FENCED and ROUTING_STALE to FailedPrecondition like every other terminal code; without an entry they fell through to codes.Unknown, which reads as a transport failure rather than a decision the server made. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Signed-off-by: cai.zhang <cai.zhang@zilliz.com>
The registration map is keyed by collection id, one entry per collection per pchannel, and DropVChannel already argued at length that every operation on it must match the vchannel NAME because a successor can take over the slot. The other three operations did not. - CheckIfVChannelCanBeWritten and GetSplitTimeTick answered from whatever entry the collection id found, so they could report the wrong shard's fence state in either direction. Both now name the vchannel, and both take the read lock the rest of this file's readers take rather than the write lock -- the fence check runs on every insert and every delete. - SplitShard fenced whatever entry it found, so a replayed or late fence could fence a successor. - CreateVChannel skipped silently when the entry belonged to another vchannel. That is the failure the DropVChannel header comment describes and it is silent and permanent: the newcomer gets a WAL genesis and a recovery-storage entry but no segment assignment, and inherits a fenced source's state. The append is now refused (ErrVChannelConflict) so the contract -- retire the source before placing a successor on its pchannel -- fails loudly instead. - DropVChannel had no guard at all; a teardown naming a live shard would delete its registration with no way back. It is now refused unless a split fenced the vchannel, while a teardown for a vchannel this pchannel no longer holds still goes through, because the recovery storage and the flusher are keyed by vchannel and still have work to do. Recovery had the same hole from the other side: if the snapshot holds two vchannels of one collection -- a fenced source whose teardown has not been observed yet, plus its successor -- the surviving entry was whichever map iteration reached last, so a restart could leave a live shard unwritable at random and leave it that way. The collision is now resolved deterministically (the shard that can still take writes wins, ties broken by name) and logged. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Signed-off-by: cai.zhang <cai.zhang@zilliz.com>
handleSplitShard flushed by collection id while the DropVChannel path it landed alongside introduced a vchannel-scoped variant and argued for it: a recovery storage covers one pchannel, a collection's other shards can live on it, and sealing them is a collection-wide flush that no message asked for. The same argument applies to the fence, so it now uses the same scope. Also exempts DropVChannel from the "vchannel not found" inconsistency probe. Once a retired vchannel's meta is garbage-collected, a replay from an older checkpoint meets its teardown again -- a replay, not a broken invariant, and DropCollection is exempt for exactly that reason. Adds the tests these paths never had: that a fence and a teardown both leave a sibling shard's growing segment alone, that the teardown is idempotent and survives a vchannel that is gone entirely, that the replay is not reported as an inconsistency while a message that genuinely needs the vchannel still is, that T_switch reaches the persisted snapshot, and that dispatching DropVChannel closes the data sync service without forwarding it (the V2 message the msgpack adaptor has no case for -- the trap the CreateVChannel branch documents and the one the PR called out but never covered). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Signed-off-by: cai.zhang <cai.zhang@zilliz.com>
Every Validate in splitshard.go returned a raw errors.New, which the repo's ruleguard gate rejects. These are Input errors by the blame test -- the request content itself is what forces the branch -- so they become WrapErrParameterMissingMsg / WrapErrParameterInvalidMsg. The gate never fired in CI because the build failed first on the unmerged proto dependency. Also corrects two comments the design outgrew: the doc comment of InitSplitTargetVChannels had two drafts spliced together, both starting from the function name, and BarrierTimeTick was documented as "always greater than T_switch" in one place and "= T_switch" in another. It is a barrier, so any value >= T_switch is correct and the caller passes T_switch; the field now says that once. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Signed-off-by: cai.zhang <cai.zhang@zilliz.com>
The message-semantics table is the repo's register of what a WAL message means, what lock it takes and what order it obeys, and it had no entry for the three messages this branch adds. Adds them, plus the split vchannel lifecycle: the one-vchannel-per-collection-per-pchannel invariant and where it is now enforced, the ordering requirement that DropVChannel precede the coordinator removing the vchannel from the collection, and the rule that neither CreateVChannel nor DropVChannel may be forwarded to the data sync service. Records the same keying rule in the shard-management doc. Also softens the delegator whitelist comment, which claimed a benefit this branch does not yet deliver: opening the filter is necessary but not sufficient, because the querynode filter node still drops SplitShard through its default branch. The delegator-side handling lands with the read path. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Signed-off-by: cai.zhang <cai.zhang@zilliz.com>
Contributor
|
[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 |
sre-ci-robot
added
the
do-not-merge/doc-need-two-approve
Contributor
|
[approval-status] effective-owner-approvals=0 [none]; do-not-merge/disable-approve-self=not-required; do-not-merge/doc-need-two-approve=0/2 blocked; ignored=[none] |
Contributor
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
2 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.
issue: #50463
design doc: docs/design-docs/design_docs/20260610-shard_split.md
Requires milvus-io/milvus-proto#618. CI cannot go green until that merges — the pinned
go-apihas noschema.HashRouting,schema.CollectionShardInfoorcommonpb.MsgType_SplitShard, so the build fails before any check runs.Step 2 of online shard split. Independent of #50495 (the routing abstraction) — this branch builds on plain master, so the two can be reviewed in parallel.
What this adds
Three WAL messages, all
ExclusiveRequiredbecause each rewrites a vchannel's registration and nothing may be assigning segments while it does:SplitShardManualFlushis needed. Idempotent: a re-fence returnsSHARD_FENCEDcarrying the recordedT_switch, so a coordinator that crashed after fencing recovers it.CreateVChannelBarrierTimeTick>= every source'sT_switch— the new WAL is born strictly after the fence, so creation doubles as activation.DropVChannelCreateVChannel.Each lands in the four places a vchannel's lifecycle is tracked: the shard interceptor, the shard manager, the recovery storage, and the flusher. The message semantics are recorded in
docs/agent_guides/streaming-system/message/message-semantic-collection.md.Three details worth the reviewer's attention
Every operation on the shard manager's registration is guarded by the vchannel NAME. The map is keyed by collection id — one entry per collection per pchannel — so once a retired source's slot is reclaimed, a later vchannel of the same collection can hold that entry. Reads (
CheckIfVChannelCanBeWritten,GetSplitTimeTick), the fence (SplitShard) and both lifecycle edges all match on the name, or they would answer for, fence, or tear down the wrong shard.That invariant is also enforced, not assumed:
CreateVChannelrefuses the append when another vchannel of the collection already holds the pchannel (ErrVChannelConflict), andDropVChannelrefuses to retire a vchannel no split has fenced. The failure being prevented is silent and permanent — the newcomer would get a WAL genesis and a recovery-storage entry but no segment assignment, and would inherit a fenced source's state. Recovery resolves the same collision deterministically instead of by map-iteration order.Neither
CreateVChannelnorDropVChannelis forwarded to the data sync service. Both are V2 messages, and forwarding routes them intofromMessageToTsMsgV2, which has no case for either and panics the process.DropCollectionmay fall through only because it is V1, handled by the unmarshaler path.Error codes
Adds
SHARD_FENCED(withfenced_time_tick) andROUTING_STALE. Both map toFailedPrecondition.fenced_time_ticktravels on the wire:TryIntoStreamingErrorclones the whole detail message rather than rebuilding it from(code, cause), which is what makes the crash-recovery path work when the coordinator and the streamingnode are separate processes.Reserved and unwired in this PR
ROUTING_STALE— the routing-version negotiation was dropped in favour of theSHARD_FENCEDreject-refetch loop. No producer emits it; kept for a possible later fast path.AlterCollectionMessageUpdatesfields 10–14 (virtual_channel_names,physical_channel_names,shard_infos,routing_modulus,shard_by) andFieldMaskCollectionShardSplitRouting— the routing commit these describe is applied by the coordinator PR; the proto surface lands here so the sibling branches share one generatedmessages.pb.go.SplitShard— opening the WAL scanner's filter is necessary but not sufficient. The querynode's filter node still drops the message through its default branch; the delegator-side handling lands with the read path.Not wired up yet
Nothing appends these messages. The coordinator that drives them follows in a later PR, all behind
dataCoord.shardSplit.enable.🤖 Generated with Claude Code