upstream-archive/qdrant
1
0
Fork 0
mirror of https://github.com/qdrant/qdrant.git synced 2026-07-23 11:11:00 -05:00
Code Issues Packages Projects Releases Wiki Activity

Fix resharding, on queries filter shards on all shard selectors (#9882)

Browse Source 
* Fix resharding, on queries filter shards on all shard selectors

* Add failing consensus test: search during resharding with shard keys (#9880)

Reproduces a known bug: after resharding is initialized on a custom
sharded collection with a shard key, searches (with and without the
shard key selector) fail with "does not have enough active replicas",
because the new resharding shard is included in reads before it has
an active replica.

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

* Exempt explicit shard id selection from resharding read filter

Explicit shard id selection is only used by internal per-shard
operations (local shard API, internal gRPC reads), including the
resharding driver reading back migrated points from the new shard.
These must reach the resharding shard before it becomes visible to
user-facing selectors, and filtering them also made per-shard reads
return silently empty results on peers lagging on hashring commits.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* Explicitly set resharding filtering per match branch

---------

Co-authored-by: Andrey Vasnetsov <andrey@vasnetsov.com>
Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Tim Visée
2026-07-17 12:31:02 +02:00
committed by timvisee
parent 44ad62f8cd
commit 0da8d5881e
2 changed files with 149 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files

70
lib/collection/src/shards/shard_holder/mod.rs
View File

@@ -575,9 +575,10 @@ impl ShardHolder {
) -> CollectionResult<Vec<(&'a Arc<ShardReplicaSet>, Option<&'a ShardKey>)>> {
let mut res = Vec::new();
match shard_selector {
let filter_resharding = match shard_selector {
ShardSelectorInternal::Empty => {
debug_assert!(false, "Do not expect empty shard selector")
debug_assert!(false, "Do not expect empty shard selector");
false
}
ShardSelectorInternal::All => {
let is_custom_sharding = match self.sharding_method {
@@ -585,33 +586,7 @@ impl ShardHolder {
ShardingMethod::Custom => true,
};
let resharding_state = self.resharding_state.read().clone();
for (&shard_id, shard) in self.shards.iter() {
// Ignore a new resharding shard until it completed point migration
// The shard will be marked as active at the end of the migration stage
let resharding_migrating_up = resharding_state.as_ref().is_some_and(|state| {
state.direction == ReshardingDirection::Up
&& state.shard_id == shard_id
&& state.stage < ReshardingStage::ReadHashRingCommitted
});
if resharding_migrating_up {
continue;
}
// Skip shard being removed by resharding down once the write
// hash ring is committed. The shard is logically gone at this
// point; querying it on a remote peer that already applied
// `finish_resharding` would return a "shard not found" error.
let resharding_removing_down = resharding_state.as_ref().is_some_and(|state| {
state.direction == ReshardingDirection::Down
&& state.shard_id == shard_id
&& state.stage >= ReshardingStage::WriteHashRingCommitted
});
if resharding_removing_down {
continue;
}
// Technically, we could skip inactive shards regardless of sharding method,
// as we do not expect that shard id can even become inactive on all replicas.
// (if it happens, means there is a bug)
@@ -624,6 +599,8 @@ impl ShardHolder {
let shard_key = self.shard_id_to_key_mapping.get(&shard_id);
res.push((shard, shard_key));
}
true
}
ShardSelectorInternal::ShardKey(shard_key) => {
for shard_id in self.get_shard_ids_by_key(shard_key)? {
@@ -633,6 +610,8 @@ impl ShardHolder {
debug_assert!(false, "Shard id {shard_id} not found")
}
}
true
}
ShardSelectorInternal::ShardKeys(shard_keys) => {
for shard_key in shard_keys {
@@ -644,6 +623,8 @@ impl ShardHolder {
}
}
}
true
}
ShardSelectorInternal::ShardKeyWithFallback(key) => {
let (shard_ids_to_query, used_shard_key) =
@@ -658,6 +639,8 @@ impl ShardHolder {
debug_assert!(false, "Shard id {shard_id} not found")
}
}
true
}
ShardSelectorInternal::ShardId(shard_id) => {
if let Some(replica_set) = self.shards.get(shard_id) {
@@ -665,8 +648,39 @@ impl ShardHolder {
} else {
return Err(shard_not_found_error(*shard_id));
}
// Exempt from resharding filter. This selector is used by internal per-shard
// operations, such as the resharding driver reading back migrated points from the
// new shard, which must be able to reach the shard before it becomes visible to
// user-facing selectors
false
}
};
// Filter out shards that must not be queried while resharding, regardless of how they
// were selected above
if filter_resharding && let Some(state) = self.resharding_state.read().as_ref() {
res.retain(|(shard, _)| {
if state.shard_id != shard.shard_id {
return true;
}
// Ignore a new resharding shard until it completed point migration
// The shard will be marked as active at the end of the migration stage
let resharding_migrating_up = state.direction == ReshardingDirection::Up
&& state.stage < ReshardingStage::ReadHashRingCommitted;
// Skip shard being removed by resharding down once the write
// hash ring is committed. The shard is logically gone at this
// point; querying it on a remote peer that already applied
// `finish_resharding` would return a "shard not found" error.
let resharding_removing_down = state.direction == ReshardingDirection::Down
&& state.stage >= ReshardingStage::WriteHashRingCommitted;
!resharding_migrating_up && !resharding_removing_down
});
}
Ok(res)
}

107
tests/consensus_tests/test_resharding_shard_keys.py Normal file
View File

@@ -0,0 +1,107 @@
import pathlib
from .custom_sharding import create_collection_with_custom_sharding, create_shard
from .test_custom_sharding import wait_for_peer_metadata
from .test_resharding import start_resharding
from .fixtures import *
from .utils import *
N_PEERS = 3
N_REPLICAS = 2
COLLECTION_NAME = "test_collection"
SHARD_KEY = "cats"
POINTS = [
{"id": 1, "vector": [0.29, 0.81, 0.75, 0.11], "payload": {"name": "Barsik"}},
{"id": 2, "vector": [0.19, 0.11, 0.15, 0.21], "payload": {"name": "Murzik"}},
{"id": 3, "vector": [0.99, 0.81, 0.75, 0.31], "payload": {"name": "Vaska"}},
{"id": 4, "vector": [0.29, 0.01, 0.05, 0.91], "payload": {"name": "Chubais"}},
]
def test_search_during_resharding_with_shard_keys(tmp_path: pathlib.Path):
"""
Creates a custom sharded collection with a shard key and 2 replicas,
then starts resharding (initialization only, no progress) and asserts
that searches with and without shard key selector keep working.
Expected to fail due to a known bug: after resharding is started on
a collection with shard keys, searches break.
"""
assert_project_root()
# Allow resharding
env = {
"QDRANT__CLUSTER__RESHARDING_ENABLED": "true",
}
peer_api_uris, _peer_dirs, _bootstrap_uri = start_cluster(tmp_path, N_PEERS, extra_env=env)
# Wait until all peers submit their metadata to consensus
wait_for_peer_metadata(peer_api_uris[0])
create_collection_with_custom_sharding(
peer_api_uris[0],
shard_number=1,
replication_factor=N_REPLICAS,
)
wait_collection_exists_and_active_on_all_peers(
collection_name=COLLECTION_NAME,
peer_api_uris=peer_api_uris,
)
create_shard(
peer_api_uris[0],
COLLECTION_NAME,
shard_key=SHARD_KEY,
shard_number=1,
replication_factor=N_REPLICAS,
)
# Upload a minimal amount of data
r = requests.put(
f"{peer_api_uris[0]}/collections/{COLLECTION_NAME}/points?wait=true", json={
"shard_key": SHARD_KEY,
"points": POINTS,
})
assert_http_ok(r)
# Searches must work before resharding
for peer_api_uri in peer_api_uris:
assert_search_works(peer_api_uri, with_shard_key=True)
assert_search_works(peer_api_uri, with_shard_key=False)
# Start resharding, initialization only, no transfer progress needed
resp = start_resharding(peer_api_uris[0], COLLECTION_NAME, shard_key=SHARD_KEY)
assert_http_ok(resp)
wait_for_collection_resharding_operations_count(peer_api_uris[0], COLLECTION_NAME, 1)
# Searches must still work after resharding is initialized
for peer_api_uri in peer_api_uris:
assert_search_works(peer_api_uri, with_shard_key=True)
assert_search_works(peer_api_uri, with_shard_key=False)
def assert_search_works(peer_api_uri: str, with_shard_key: bool):
request = {
"vector": [0.29, 0.81, 0.75, 0.11],
"limit": 10,
"with_payload": True,
}
if with_shard_key:
request["shard_key"] = SHARD_KEY
r = requests.post(
f"{peer_api_uri}/collections/{COLLECTION_NAME}/points/search",
json=request,
)
assert_http_ok(r)
result = r.json()["result"]
selector = f"shard_key={SHARD_KEY!r}" if with_shard_key else "no shard_key"
assert len(result) == len(POINTS), \
f"search on {peer_api_uri} with {selector} returned {len(result)} points, expected {len(POINTS)}"
for point in result:
assert point["shard_key"] == SHARD_KEY