Add shard routing types and the SplitShard message type for online shard split - #618
Add shard routing types and the SplitShard message type for online shard split#618xiaocai2333 wants to merge 1 commit into
Conversation
|
[APPROVALNOTIFIER] This PR is NOT APPROVED This pull-request has been approved by: xiaocai2333 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 |
|
@xiaocai2333 Please associate the related issue to the body of your Pull Request. (eg. “issue: #6534”) |
mergify
Bot
added
do-not-merge/missing-related-issue
ci-passed
and removed
do-not-merge/missing-related-issue
labels
1586acf to
72c1497
Compare
xiaocai2333
changed the title
|
🤖 team2 automated review (single agent · not cross-verified, for reference) · by Here's my review. I've audited every changed proto file plus the regenerated Go. milvus-proto PR #618 review — Add shard routing types and SplitShard message typeTL;DRClean, purely additive change — no field renumbering, no enum-value reuse, no RPC modified — so it's fully backward compatible. Generated Go is consistently regenerated and naming matches house style. No blocking issues; three low/nit notes. Correctness audit
Low / nits (non-blocking)
RecommendationApprove. Only the Low item warrants an author acknowledgement/doc; the rest are nits. Saved: The task says to post this to the PR. Since it's an overall review spanning multiple files, I'll post it as a general PR comment (English). Here's the exact dry-run: Target:
Send it? |
|
Tick the box to add this pull request to the merge queue (same as
|
72c1497 to
93d4477
Compare
5d8558e to
8f28805
Compare
d6369b5 to
18d3955
Compare
18d3955 to
bcbdc95
Compare
bcbdc95 to
20cdfa6
Compare
Where a write goes is decided in one place instead of at each caller. Today the
proxy computes hash(pk) % len(vchannels) inline, which is only correct while
every shard owns an equal, position-derived slice of the key space -- exactly
what a shard split stops being true.
A shard now carries an explicit PREDICATE and the table is derived from the
predicates rather than from a channel count:
- HashRouting: hash % modulus == remainder, so a split shard's two halves are
describable ({2,0} becomes {4,0} and {4,2}) while untouched shards keep the
bucket they already had, bit for bit.
- RangeRouting: byte-comparable key ranges, for collections sharded by
namespace.
Both are normalized into a flat lookup. Hash buckets of different moduli --
which a sequence of doublings produces -- are put on M = lcm(all moduli) so a
route is one array index rather than a scan over predicates, and DeriveHash
rejects a shard set that does not tile the key space exactly: a gap (some key
routes nowhere) or an overlap (some key routes to two shards) fails loudly
instead of silently misrouting writes.
Behaviour is unchanged for every existing collection. A collection whose shards
carry no predicate at all is the legacy case, and the table built for it is
exactly hash % shardNum by position -- the same placement HashPK2Channels
produces, verified against it in the tests.
Nothing calls this yet. It is the first step of online shard split (design doc
docs/design-docs/design_docs/20260805-shard_split_primary_key_tables.md); the
write path, the split state machine and the read-side handover follow in
separate PRs, all behind dataCoord.shardSplit.enable.
Requires milvus-io/milvus-proto#618 for the schemapb routing types.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Signed-off-by: cai.zhang <cai.zhang@zilliz.com>
Re-review of
|
This PR is a contract-only change (new proto messages, comment blocks, and regenerated Go bindings), and review surfaced one documentation inconsistency in the newly added Low
The comment defines Suggestion: name the event explicitly — e.g. change the clause to "the two agree again once the dropped source is reclaimed" — or state in this file that "split completes" includes the reclaim commit. (raised by bigsheeper) |
This review covers the newly added shard-split routing contract in High
Medium
Low
Pre-existing hazards — not introduced by this PR (follow-up only)The following hazards are real but pre-existing and were not introduced by this PR. They should be tracked as separate follow-up work and do not block this PR. No tracking issue yet.
|
|
All seven addressed in High
Took the first half of your suggestion rather than the second. Pinning the routing value's
MediumEmpty Which key the legacy rule hashes. Right, and Never-split test written per entry. The exception sitting ten lines below the conclusion Tiling on a never-split table. Agreed, and it is sharper than a wording problem: the rule Low (both reviews)
On the pre-existing hazard: agreed on all counts, including that it cannot misbehave on Comment-only throughout; the regenerated descriptor bytes are unchanged. |
Adversarial review found no issues requiring changes. Verified:
No further findings survived verification: no disputed items and no pre-existing hazards were carried into this review. |
Re-review of
|
f3996aa to
47999c3
Compare
|
The collection-wide modulus rework introduces three high-severity contract problems in the routing IDL: the modulus is unreachable from two of High
|
47999c3 to
06e6196
Compare
|
All three were comments that did not say enough, and are fixed in the amended commit. Two of
It does. The modulus is a collection-level fact and sits wherever that collection's shard
A collection holds at most one vchannel per pchannel, enforced when vchannels are allocated, The comment now carries the growth, the doubling, and where the cap comes from.
You are right that it is not total: once the source is reclaimed and leaves the vchannel Amended into the single commit rather than added on top; the PR carries one. |
SummaryThis round's routing prose is additive on the wire and the regenerated Go matches the IDL, but the new comment block in High
Medium
Low
Nitpick
|
06e6196 to
0bfef62
Compare
|
All five are wrong sentences rather than wrong types, and all five are fixed in the amended
Both cases are now stated, since which one applies is a property of the router rather than of
Restated with the condition, and the two patterns given as what they are: repeated splits of
|
Both findings concern the new sharding contract text in the proto files: one underspecified hash rule that can misroute writes, and one worked example that contradicts its own stated rule. Medium
The comment says Failure: a non-Go router (or a Go one switching murmur3 libraries) implements the INT64 routing value from this comment and picks an implementation with a non-zero default seed, or a 128-bit variant truncated to 32 bits → it computes a different routing value, and after the modulus the row lands on a shard that does not own that residue; only keys hitting a fenced source vchannel are rejected with Suggestion: state Low
The comment reads "Four shards split once each end at four times the modulus under the first pattern and at twice it under the second." Working it through with the rules this same block establishes — line 617-618, "IT DOUBLES ONLY WHEN THE SHARD BEING HALVED OWNS A SINGLE RESIDUE"; lines 614-615, an unsplit collection has modulus =
4× corresponds to a mixed ordering with exactly two doublings (e.g. one lineage split twice, plus two untouched shards split once each), not to "the first pattern". Since this is the only recomputable example of modulus growth in the block, an implementer using it to self-check their split logic gets a spurious mismatch signal. Suggestion: change "four times" to "sixteen times", or replace it with an example that spells out the split count explicitly. The neighbouring bound argument ("Fifteen doublings … the modulus stays under 2^15") checks out; only this sentence is wrong. (raised by bigsheeper) |
A collection's shards are fixed at creation today. Online shard split changes that count without taking the collection offline, and this is the contract the pieces of it agree on: what a shard is, which keys it owns, and how a row's routing key is computed. schema.proto gains ShardState, the four states a shard passes through while a split runs; CollectionShardInfo, one shard's state, vchannel and routing predicate; and HashRouting, the residues of the collection's modulus that shard owns. CollectionShardInfo supersedes the single-field etcdpb.CollectionShardInfo the truncate API left in the collection meta, keeping its field number so records already persisted still decode -- their missing fields read as a normal, never-split shard, which is what they are. DescribeCollectionResponse gains shard_infos, index-parallel to virtual_channel_names; shard_by, the expression producing a row's routing value; and routing_modulus, the one modulus the whole collection's residues are taken against. Its shards_num keeps its number and changes meaning: it is the count of shards serving now, which a split moves, rather than the CreateCollection argument. common.proto gains MsgType SplitShard, numbered with the collection DDL it belongs to rather than the WAL group it travels through. Placement is by modulus throughout. Range routing and a RoutingMode enum were considered and dropped: with the comparison key reduced to the routing value itself, ranges bought nothing hash buckets do not, and one scheme needs no enum to select it. The oneof keeps a single member so a second scheme has somewhere to arrive. issue: milvus-io#619 Signed-off-by: Cai Zhang <cai.zhang@zilliz.com>
0bfef62 to
c929178
Compare
Overall: the wire change is additive and the generated code matches the IDL, but several of the new normative comments in Medium
Low
|
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>
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>
… 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>
Re-review of
|
|
/lgtm |
4 participants
issue: #619
A collection's shard count is fixed at creation today. Online shard split changes it without taking the collection offline, and this PR adds the types the pieces of that agree on: what a shard is, which keys it owns, and how a row's routing key is computed.
Additive. No existing field is removed or renumbered, and no field number is skipped. One released field,
DescribeCollectionResponse.shards_num, changes meaning — see below.schema.proto
ShardState—ShardNormal/ShardCreating/ShardSplitting/ShardDropped. Which shards take writes, serve reads, and own keys differ during a split, and this is what says so. The zero value is the legacy shape.CollectionShardInfo— one shard's state, its vchannel, and its routing predicate. Only what is true of the shard itself; a split is a relation between shards and belongs to the split task, which ends.HashRouting— the residues of the collection's modulus that shard owns.CollectionShardInfosupersedesetcdpb.CollectionShardInfo, a separate message that happens to share its name and holds a single truncate tick in the collection meta today. The meta swaps one for the other at the same field number, so field 1 here is that same tick and cannot be renumbered: records written before the swap decode with their remaining fields at zero, which reads as a normal, never-split shard — which is what they are.milvus.proto
DescribeCollectionResponsegains:shard_infos— index-parallel tovirtual_channel_names.shard_by— the expression producing a row's routing value,hash(<field>).routing_modulus— the one modulus the whole collection's residues are taken against.shards_numkeeps its number and changes meaning: it is the count of shards serving now, which a split moves, rather than theCreateCollectionargument. A tool that recreates a collection from it does so at the current count.ShowCollectionsResponse.shards_numfollows, andCreateCollectionRequest.shards_num's comment no longer claims the count is fixed for the collection's life — there is no API to change that argument, which is a different statement.common.proto
MsgType_SplitShard = 120— the legacy msgstream type theSplitShardWAL message maps to. Numbered with the collection DDL, not the WAL group:pkg/streaming/util/message/adaptornumbers a WAL message by the operation it represents, which is whyCreateCollectionandTruncateCollectionare 100 and 119 rather than sitting besideAlterWAL.Placement is by modulus
The legacy rule has two halves — hash the collection's routing key, then take the remainder modulo the shard count — and this PR names both.
shard_byis the first half and is immutable: changing which field is hashed changes where every existing row belongs, and no split expresses that. Before it, the second half was on the wire and the first was left to inference — a consumer had to decide from the schema whether the routing key was a namespace or a primary key, which is wrong for anyone guessing "primary key" on a namespace-sharded collection.The predicates in
shard_infosare the second half, and they move with every split. Sohash(pk)does not meanhash(pk) % shards_num, which stops holding the moment a collection is split.One modulus for the whole collection, with each shard naming its own residues, rather than a
{modulus, remainder}pair per shard. The pair form made two shards incomparable —{2,0}strictly contains{4,0}— so disjointness needed a gcd and coverage a rational sum, and the runtime normalised it away before use anyway. With one modulus the rules are set operations. The cost is size: a pair is O(1) however many splits have happened, while residues are O(modulus / share).Range routing and a
RoutingModeenum were considered and dropped. Ranges existed so a split could choose a boundary on a namespace edge and isolate one tenant; with the comparison key reduced to the routing value itself they bought nothing hash buckets do not, and relabelling whole segments is only sound when a segment holds one namespace, which is not the default. With one scheme left, an enum to select it earned nothing. Theoneofkeeps its single member so a second scheme has somewhere to arrive.What a consumer needs to get right
Spelled out in the comments; the load-bearing ones:
A collection that has never been split carries no predicate at all — not because it predates this feature, but because creation materializes none. It routes by the legacy placement,
hash(key) % len(virtual_channel_names), which is a valid modulus there and nowhere else.Test that over the whole table, not one entry. A fenced source is predicate-less by design and sits beside predicated targets for the whole of a split, so one unset
routingoneof does not mean the collection is unsplit. IgnoreShardSplittingandShardDroppedfirst.hashis not a free choice, and the mask is not part of it. A string key is CRC32-IEEE over at most its first 100 bytes, unmasked. An int64 key is murmur3_32 over the 8-byte little-endian encoding, and only then masked with0x7fffffff. Masking a string key diverges from the server.shard_bymay name a field this response's schema does not return. A namespace-sharded collection routes by$namespace_id, which the public schema projection strips; its type is pinned in the IDL instead.An empty
shard_byis not a grammar violation, it is every collection until its first split. Test for it before parsing. The legacy rule it falls back to hashes the namespace only whenenable_namespaceis set andnamespace.sharding.enabledis true — it defaults to false — andnamespace.modeispartition_key.Order two shard tables by
update_timestamp. There is no separate routing version, so it says "not older", not "the routing changed".Adding a second
routingvariant is not additive for readers built today: an unknown field number leaves the oneof reading unset, which they take as never-split. Whatever makes the two distinguishable has to arrive with the variant.Known gaps on the Milvus side
The IDL states the contract; these are where the implementation does not meet it yet, and belong to the split-implementation PR rather than this one.
CreatePartition/DropPartitionwalkvirtual_channel_namesasfor i < shards_num. Mid-split that prefix is neither the key-owning shards nor all of them. Latent on master, where no split runs and the two are equal.shards_numwas stored decodes to 0 and the server derives the value from the vchannel list, which mid-split is the wrong count. Persisting the derived value closes it, and closes theShowCollectionsdifference with it.namespace.mode=partition, and the trigger checksenable_namespacealone.etcdpb.CollectionInfocarries noshard_by.pkg/util/typeutil.Consumed by the Milvus shard-split feature (design: milvus-io/milvus#50465).