Make reads and point counts consistent during resharding (#4617)

* Apply resharding filter only to existing shards, include shard selection

* Add test for stable exact point count during resharding

* Improve filtering, only two separate requests, filtered and non-filtered

* Make scrolling stable while resharding

* Add test for stable scroll during resharding

* Make search stable while resharding

* Add test for stable search during resharding

* Remove resharding post filter in retrieve

* Also assert cardinality point count in resharding test

* Fix typos and some tweaks

* Only clone filter if resharding is active

* Also minimize cloning with resharding filter on count request

* Add test for stable exact point count during resharding with indexing

* Update lib/collection/src/shards/shard_holder/mod.rs

Co-authored-by: Roman Titov <ffuugoo@users.noreply.github.com>

* Update lib/collection/src/collection/point_ops.rs

Co-authored-by: Roman Titov <ffuugoo@users.noreply.github.com>

* Also apply resharding filter to retrieve

* Restructure resharding filter usage in `Collection::retrieve`

---------

Co-authored-by: Roman Titov <ffuugoo@users.noreply.github.com>

lib/collection/src/collection/point_ops.rs

@@ -207,15 +207,10 @@ impl Collection {
 
     pub async fn scroll_by(
         &self,
-        mut request: ScrollRequestInternal,
+        request: ScrollRequestInternal,
         read_consistency: Option<ReadConsistency>,
         shard_selection: &ShardSelectorInternal,
     ) -> CollectionResult<ScrollResult> {
-        merge_filters(
-            &mut request.filter,
-            self.shards_holder.read().await.resharding_filter(),
-        );
-
         let default_request = ScrollRequestInternal::default();
 
         let id_offset = request.offset;
@@ -252,7 +247,27 @@ impl Collection {
         let retrieved_points: Vec<_> = {
             let shards_holder = self.shards_holder.read().await;
             let target_shards = shards_holder.select_shards(shard_selection)?;
+
+            // Resharding filter to apply when resharding is active
+            let resharding_filter = shards_holder.resharding_filter();
+            let reshard_shard_id = shards_holder
+                .resharding_state
+                .read()
+                .as_ref()
+                .map(|state| state.shard_id);
+
+            // Create a normal and resharding filter
+            // Resharding filter must be used on existing shards if resharding is active
+            let (normal_filter, reshard_filter) =
+                normal_and_resharding_filter(request.filter, resharding_filter);
+
             let scroll_futures = target_shards.into_iter().map(|(shard, shard_key)| {
+                // Take resharding filter if available on existing shards, otherwise take normal filter
+                let filter = reshard_filter
+                    .as_ref()
+                    .filter(|_| Some(shard.shard_id) != reshard_shard_id)
+                    .or(normal_filter.as_ref());
+
                 let shard_key = shard_key.cloned();
                 shard
                     .scroll_by(
@@ -260,7 +275,7 @@ impl Collection {
                         limit,
                         &with_payload_interface,
                         &with_vector,
-                        request.filter.as_ref(),
+                        filter,
                         read_consistency,
                         local_only,
                         order_by.as_ref(),
@@ -348,28 +363,38 @@ impl Collection {
 
     pub async fn count(
         &self,
-        mut request: CountRequestInternal,
+        request: CountRequestInternal,
         read_consistency: Option<ReadConsistency>,
         shard_selection: &ShardSelectorInternal,
     ) -> CollectionResult<CountResult> {
-        merge_filters(
-            &mut request.filter,
-            self.shards_holder.read().await.resharding_filter(),
-        );
-
         let shards_holder = self.shards_holder.read().await;
         let shards = shards_holder.select_shards(shard_selection)?;
 
-        let request = Arc::new(request);
+        // Resharding filter to apply when resharding is active
+        let resharding_filter = shards_holder.resharding_filter();
+        let reshard_shard_id = shards_holder
+            .resharding_state
+            .read()
+            .as_ref()
+            .map(|state| state.shard_id);
+
+        // Create a request with resharding filtering a normal and resharding filter
+        // Should be used on all shards, except the new resharding shard
+        let (normal_request, reshard_request) =
+            normal_and_resharding_count_request(request, resharding_filter);
+
         let mut requests: futures::stream::FuturesUnordered<_> = shards
             .into_iter()
             // `count` requests received through internal gRPC *always* have `shard_selection`
             .map(|(shard, _shard_key)| {
-                shard.count(
-                    request.clone(),
-                    read_consistency,
-                    shard_selection.is_shard_id(),
-                )
+                // Take resharding request if available on existing shards, otherwise take normal request
+                let request = reshard_request
+                    .as_ref()
+                    .filter(|_| Some(shard.shard_id) != reshard_shard_id)
+                    .unwrap_or(&normal_request)
+                    .clone();
+
+                shard.count(request, read_consistency, shard_selection.is_shard_id())
             })
             .collect();
 
@@ -395,35 +420,55 @@ impl Collection {
         let with_payload = WithPayload::from(with_payload_interface);
         let request = Arc::new(request);
 
-        #[allow(unused_assignments)]
-        let mut resharding_filter = None;
-
         let all_shard_collection_results = {
             let shard_holder = self.shards_holder.read().await;
-
-            // Get resharding filter, while we hold the lock to shard holder
-            resharding_filter = shard_holder.resharding_filter_impl();
-
             let target_shards = shard_holder.select_shards(shard_selection)?;
+
+            // Resharding filter to apply when resharding is active
+            let resharding_filter = shard_holder.resharding_filter_impl();
+            let reshard_shard_id = shard_holder
+                .resharding_state
+                .read()
+                .as_ref()
+                .map(|state| state.shard_id);
+
             let retrieve_futures = target_shards.into_iter().map(|(shard, shard_key)| {
-                let shard_key = shard_key.cloned();
-                shard
-                    .retrieve(
-                        request.clone(),
-                        &with_payload,
-                        &request.with_vector,
-                        read_consistency,
-                        shard_selection.is_shard_id(),
-                    )
-                    .and_then(move |mut records| async move {
-                        if shard_key.is_none() {
-                            return Ok(records);
-                        }
-                        for point in &mut records {
-                            point.shard_key.clone_from(&shard_key);
-                        }
-                        Ok(records)
-                    })
+                // Explicitly borrow `request` and `with_payload`, so we can use them in `async move`
+                // block below without unnecessarily cloning anything
+                let request = &request;
+                let with_payload = &with_payload;
+
+                // If resharding, prepare filter to exclude migrated points from *old* shards
+                let resharding_filter = resharding_filter
+                    .as_ref()
+                    .filter(|_| Some(shard.shard_id) != reshard_shard_id);
+
+                async move {
+                    let mut records = shard
+                        .retrieve(
+                            request.clone(),
+                            with_payload,
+                            &request.with_vector,
+                            read_consistency,
+                            shard_selection.is_shard_id(),
+                        )
+                        .await?;
+
+                    // If resharding, exclude migrated points from *old* shards
+                    if let Some(filter) = resharding_filter {
+                        records.retain(|record| !filter.check(record.id));
+                    }
+
+                    if shard_key.is_none() {
+                        return Ok(records);
+                    }
+
+                    for point in &mut records {
+                        point.shard_key.clone_from(&shard_key.cloned());
+                    }
+
+                    CollectionResult::Ok(records)
+                }
             });
 
             future::try_join_all(retrieve_futures).await?
@@ -433,11 +478,6 @@ impl Collection {
         let points = all_shard_collection_results
             .into_iter()
             .flatten()
-            // If resharding is in progress, and *read* hash-ring is committed, filter-out "resharded" points
-            .filter(|point| match &resharding_filter {
-                Some(filter) => filter.check(point.id),
-                None => true,
-            })
             // Add each point only once, deduplicate point IDs
             .filter(|point| covered_point_ids.insert(point.id))
             .collect();
@@ -446,6 +486,7 @@ impl Collection {
     }
 }
 
+#[inline]
 fn merge_filters(filter: &mut Option<Filter>, resharding_filter: Option<Filter>) {
     if let Some(resharding_filter) = resharding_filter {
         *filter = Some(match filter.take() {
@@ -454,3 +495,48 @@ fn merge_filters(filter: &mut Option<Filter>, resharding_filter: Option<Filter>)
         });
     }
 }
+
+/// Merge a regular and resharding filter
+///
+/// The first element is always the given `filter`.
+///
+/// The second element is the `filter` with `resharding_filter` merged into it. It's None if no
+/// resharding filter was given.
+///
+/// This function minimizes cloning of the filter to when it's strictly necessary.
+#[inline]
+fn normal_and_resharding_count_request(
+    mut request: CountRequestInternal,
+    resharding_filter: Option<Filter>,
+) -> (Arc<CountRequestInternal>, Option<Arc<CountRequestInternal>>) {
+    match resharding_filter {
+        None => (Arc::new(request), None),
+        Some(resharding_filter) => (Arc::new(request.clone()), {
+            merge_filters(&mut request.filter, Some(resharding_filter));
+            Some(Arc::new(request))
+        }),
+    }
+}
+
+/// Merge a regular and resharding filter
+///
+/// The first element is always the given `filter`.
+///
+/// The second element is the `filter` with `resharding_filter` merged into it. It's None if no
+/// resharding filter was given.
+///
+/// This function minimizes cloning of the filter to when it's strictly necessary.
+#[inline]
+fn normal_and_resharding_filter(
+    filter: Option<Filter>,
+    resharding_filter: Option<Filter>,
+) -> (Option<Filter>, Option<Filter>) {
+    match (filter, resharding_filter) {
+        (filter, None) => (filter, None),
+        (Some(filter), Some(resharding_filter)) => (
+            Some(filter.clone()),
+            Some(filter.merge_owned(resharding_filter)),
+        ),
+        (None, Some(resharding_filter)) => (None, Some(resharding_filter)),
+    }
+}

lib/collection/src/collection/search.rs

@@ -125,13 +125,28 @@ impl Collection {
 
     async fn do_core_search_batch(
         &self,
-        mut request: CoreSearchRequestBatch,
+        request: CoreSearchRequestBatch,
         read_consistency: Option<ReadConsistency>,
         shard_selection: &ShardSelectorInternal,
         timeout: Option<Duration>,
     ) -> CollectionResult<Vec<Vec<ScoredPoint>>> {
-        if let Some(resharding_filter) = self.shards_holder.read().await.resharding_filter() {
-            for search in &mut request.searches {
+        // Resharding filter to apply when resharding is active
+        let (resharding_filter, reshard_shard_id) = {
+            let shards_holder = self.shards_holder.read().await;
+            let resharding_filter = shards_holder.resharding_filter();
+            let reshard_shard_id = shards_holder
+                .resharding_state
+                .read()
+                .as_ref()
+                .map(|state| state.shard_id);
+            (resharding_filter, reshard_shard_id)
+        };
+
+        // Create filtered request, which has resharding filter applied
+        // Should be used on all shards, except the new resharding shard
+        let mut filtered_request = request.clone();
+        if let Some(resharding_filter) = resharding_filter {
+            for search in &mut filtered_request.searches {
                 match &mut search.filter {
                     Some(filter) => {
                         *filter = filter.merge(&resharding_filter);
@@ -143,8 +158,7 @@ impl Collection {
                 }
             }
         }
-
-        let request = Arc::new(request);
+        let filtered_request = Arc::new(filtered_request);
 
         let instant = Instant::now();
 
@@ -153,10 +167,17 @@ impl Collection {
             let shard_holder = self.shards_holder.read().await;
             let target_shards = shard_holder.select_shards(shard_selection)?;
             let all_searches = target_shards.iter().map(|(shard, shard_key)| {
+                // Take the filtered request, or the original request for the resharding shard
+                let request = if Some(shard.shard_id) == reshard_shard_id {
+                    Arc::new(request.clone())
+                } else {
+                    filtered_request.clone()
+                };
+
                 let shard_key = shard_key.cloned();
                 shard
                     .core_search(
-                        Arc::clone(&request),
+                        request,
                         read_consistency,
                         shard_selection.is_shard_id(),
                         timeout,
@@ -179,7 +200,7 @@ impl Collection {
         let result = self
             .merge_from_shards(
                 all_searches_res,
-                Arc::clone(&request),
+                Arc::clone(&filtered_request),
                 !shard_selection.is_shard_id(),
             )
             .await;

lib/collection/src/shards/resharding/mod.rs

@@ -62,6 +62,11 @@ impl ReshardState {
     }
 }
 
+/// Reshard stages
+///
+/// # Warning
+///
+/// This enum is ordered!
 #[derive(Copy, Clone, Debug, Default, Eq, PartialEq, Ord, PartialOrd, Deserialize, Serialize)]
 #[serde(rename_all = "snake_case")]
 pub enum ReshardStage {

lib/collection/src/shards/shard_holder/mod.rs

@@ -15,7 +15,7 @@ use tokio::sync::{broadcast, RwLock};
 
 use super::replica_set::AbortShardTransfer;
 use super::resharding::tasks_pool::ReshardTasksPool;
-use super::resharding::ReshardState;
+use super::resharding::{ReshardStage, ReshardState};
 use super::transfer::transfer_tasks_pool::TransferTasksPool;
 use crate::collection::payload_index_schema::PayloadIndexSchema;
 use crate::common::validate_snapshot_archive::validate_open_snapshot_archive;
@@ -452,15 +452,14 @@ impl ShardHolder {
             }
             ShardSelectorInternal::All => {
                 for (&shard_id, shard) in self.shards.iter() {
-                    // TODO(resharding): Handle resharded shard!?
-
-                    let is_resharding = self
-                        .resharding_state
-                        .read()
-                        .clone()
-                        .map_or(false, |state| state.shard_id == shard_id);
-
-                    if is_resharding {
+                    // 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 =
+                        self.resharding_state.read().clone().map_or(false, |state| {
+                            state.shard_id == shard_id
+                                && state.stage < ReshardStage::ReadHashRingCommitted
+                        });
+                    if resharding_migrating {
                         continue;
                     }
 

lib/collection/src/shards/shard_holder/resharding.rs

@@ -270,12 +270,15 @@ impl ShardHolder {
     }
 
     /// A filter that excludes points migrated to a different shard, as part of resharding.
+    ///
+    /// `None` if resharding is not active or if the read hash ring is not committed yet.
     pub fn resharding_filter(&self) -> Option<Filter> {
         let filter = self.resharding_filter_impl()?;
         let filter = Filter::new_must_not(Condition::Resharding(Arc::new(filter)));
         Some(filter)
     }
 
+    #[inline]
     pub fn resharding_filter_impl(&self) -> Option<hash_ring::Filter> {
         let state = self.resharding_state.read();
 

tests/consensus_tests/test_resharding.py

@@ -3,7 +3,7 @@ import pathlib
 import random
 from time import sleep
 
-from .fixtures import upsert_random_points, create_collection
+from .fixtures import upsert_random_points, create_collection, random_dense_vector
 from .utils import *
 
 COLLECTION_NAME = "test_collection"
@@ -269,6 +269,363 @@ def test_resharding_concurrent_updates(tmp_path: pathlib.Path):
     check_data_consistency(data)
 
 
+# Test point count during resharding.
+#
+# On a static collection, this performs resharding a few times and asserts the
+# exact point count remains stable on all peers during the whole process.
+def test_resharding_stable_point_count(tmp_path: pathlib.Path):
+    assert_project_root()
+
+    num_points = 1000
+
+    # Prevent optimizers messing with point counts
+    env={
+        "QDRANT__STORAGE__OPTIMIZERS__INDEXING_THRESHOLD_KB": "0",
+    }
+
+    peer_api_uris, _peer_dirs, _bootstrap_uri = start_cluster(tmp_path, 3, None, extra_env=env)
+    first_peer_id = get_cluster_info(peer_api_uris[0])['peer_id']
+
+    # Create collection, insert points
+    create_collection(peer_api_uris[0], shard_number=1, replication_factor=3)
+    wait_collection_exists_and_active_on_all_peers(
+        collection_name=COLLECTION_NAME,
+        peer_api_uris=peer_api_uris,
+    )
+    upsert_random_points(peer_api_uris[0], num_points)
+
+    sleep(1)
+
+    # Assert node shard and point sum count
+    for uri in peer_api_uris:
+        assert check_collection_local_shards_count(uri, COLLECTION_NAME, 1)
+        assert check_collection_local_shards_point_count(uri, COLLECTION_NAME, num_points)
+
+    # Reshard 3 times in sequence
+    for shard_count in range(2, 5):
+        # Start resharding
+        r = requests.post(
+            f"{peer_api_uris[0]}/collections/{COLLECTION_NAME}/cluster", json={
+                "start_resharding": {
+                    "peer_id": first_peer_id,
+                    "shard_key": None,
+                }
+            })
+        assert_http_ok(r)
+
+        # Wait for resharding to start
+        wait_for_collection_resharding_operations_count(peer_api_uris[0], COLLECTION_NAME, 1)
+
+        # Continuously assert point count on all peers, must be stable
+        # Stop once all peers reported completed resharding
+        while True:
+            for uri in peer_api_uris:
+                assert get_collection_point_count(uri, COLLECTION_NAME, exact=True) == num_points
+                cardinality_count = get_collection_point_count(uri, COLLECTION_NAME, exact=False)
+                assert cardinality_count >= num_points / 2 and cardinality_count < num_points * 2
+
+            all_completed = True
+            for uri in peer_api_uris:
+                if not check_collection_resharding_operations_count(uri, COLLECTION_NAME, 0):
+                    all_completed = False
+                    break
+            if all_completed:
+                break
+
+        # Assert node shard count
+        for uri in peer_api_uris:
+            assert check_collection_local_shards_count(uri, COLLECTION_NAME, shard_count)
+
+    sleep(1)
+
+    # Match all points on all nodes exactly
+    data = []
+    for uri in peer_api_uris:
+        r = requests.post(
+            f"{uri}/collections/{COLLECTION_NAME}/points/scroll", json={
+                "limit": 999999999,
+                "with_vectors": True,
+                "with_payload": True,
+            }
+        )
+        assert_http_ok(r)
+        data.append(r.json()["result"])
+    check_data_consistency(data)
+
+
+# Test point count during resharding and indexing.
+#
+# On a static collection, this performs resharding and indexing a few times and
+# asserts the exact point count remains stable on all peers during the whole
+# process.
+def test_resharding_indexing_stable_point_count(tmp_path: pathlib.Path):
+    assert_project_root()
+
+    num_points = 1000
+
+    # Configure optimizers to index right away with a low vector count
+    env={
+        "QDRANT__STORAGE__OPTIMIZERS__INDEXING_THRESHOLD_KB": "1",
+    }
+
+    peer_api_uris, _peer_dirs, _bootstrap_uri = start_cluster(tmp_path, 3, None, extra_env=env)
+    first_peer_id = get_cluster_info(peer_api_uris[0])['peer_id']
+
+    # Create collection, insert points
+    create_collection(peer_api_uris[0], shard_number=1, replication_factor=3)
+    wait_collection_exists_and_active_on_all_peers(
+        collection_name=COLLECTION_NAME,
+        peer_api_uris=peer_api_uris,
+    )
+    upsert_random_points(peer_api_uris[0], num_points)
+
+    sleep(1)
+
+    # Assert node shard and exact point count
+    for uri in peer_api_uris:
+        assert check_collection_local_shards_count(uri, COLLECTION_NAME, 1)
+        assert get_collection_point_count(uri, COLLECTION_NAME, exact=True) == num_points
+
+    # Reshard 3 times in sequence
+    for shard_count in range(2, 5):
+        # Start resharding
+        r = requests.post(
+            f"{peer_api_uris[0]}/collections/{COLLECTION_NAME}/cluster", json={
+                "start_resharding": {
+                    "peer_id": first_peer_id,
+                    "shard_key": None,
+                }
+            })
+        assert_http_ok(r)
+
+        # Wait for resharding operation to start and stop
+        wait_for_collection_resharding_operations_count(peer_api_uris[0], COLLECTION_NAME, 1)
+
+        # Continuously assert exact point count on all peers, must be stable
+        # Stop once all peers reported completed resharding
+        while True:
+            for uri in peer_api_uris:
+                assert get_collection_point_count(uri, COLLECTION_NAME, exact=True) == num_points
+
+            all_completed = True
+            for uri in peer_api_uris:
+                if not check_collection_resharding_operations_count(uri, COLLECTION_NAME, 0):
+                    all_completed = False
+                    break
+            if all_completed:
+                break
+
+        # Assert node shard count
+        for uri in peer_api_uris:
+            assert check_collection_local_shards_count(uri, COLLECTION_NAME, shard_count)
+
+    # Wait for optimizations to complete
+    for uri in peer_api_uris:
+        wait_collection_green(uri, COLLECTION_NAME)
+
+    # Assert exact point count one more time
+    for uri in peer_api_uris:
+        assert get_collection_point_count(uri, COLLECTION_NAME, exact=True) == num_points
+
+    # Match all points on all nodes exactly
+    data = []
+    for uri in peer_api_uris:
+        r = requests.post(
+            f"{uri}/collections/{COLLECTION_NAME}/points/scroll", json={
+                "limit": 999999999,
+                "with_vectors": True,
+                "with_payload": True,
+            }
+        )
+        assert_http_ok(r)
+        data.append(r.json()["result"])
+    check_data_consistency(data)
+
+
+# Test point scroll stability during resharding.
+#
+# On a static collection, this performs resharding a few times and asserts
+# scrolling remains stable on all peers during the whole process.
+def test_resharding_stable_scroll(tmp_path: pathlib.Path):
+    assert_project_root()
+
+    num_points = 1000
+    scroll_limit = 25
+
+    # Prevent optimizers messing with point counts
+    env={
+        "QDRANT__STORAGE__OPTIMIZERS__INDEXING_THRESHOLD_KB": "0",
+    }
+
+    peer_api_uris, _peer_dirs, _bootstrap_uri = start_cluster(tmp_path, 3, None, extra_env=env)
+    first_peer_id = get_cluster_info(peer_api_uris[0])['peer_id']
+
+    # Create collection, insert points
+    create_collection(peer_api_uris[0], shard_number=1, replication_factor=3)
+    wait_collection_exists_and_active_on_all_peers(
+        collection_name=COLLECTION_NAME,
+        peer_api_uris=peer_api_uris,
+    )
+    upsert_random_points(peer_api_uris[0], num_points)
+
+    sleep(1)
+
+    # Assert node shard and point sum count
+    for uri in peer_api_uris:
+        assert check_collection_local_shards_count(uri, COLLECTION_NAME, 1)
+        assert check_collection_local_shards_point_count(uri, COLLECTION_NAME, num_points)
+
+    # Match scroll sample of points on all nodes exactly
+    data = []
+    for uri in peer_api_uris:
+        r = requests.post(
+            f"{uri}/collections/{COLLECTION_NAME}/points/scroll", json={
+                "limit": scroll_limit,
+                "with_vectors": True,
+                "with_payload": False,
+            }
+        )
+        assert_http_ok(r)
+        data.append(r.json()["result"])
+    check_data_consistency(data)
+
+    # Reshard 3 times in sequence
+    for shard_count in range(2, 5):
+        # Start resharding
+        r = requests.post(
+            f"{peer_api_uris[0]}/collections/{COLLECTION_NAME}/cluster", json={
+                "start_resharding": {
+                    "peer_id": first_peer_id,
+                    "shard_key": None,
+                }
+            })
+        assert_http_ok(r)
+
+        # Wait for resharding to start
+        wait_for_collection_resharding_operations_count(peer_api_uris[0], COLLECTION_NAME, 1)
+
+        # Continuously assert point scroll samples on all peers, must be stable
+        # Stop once all peers reported completed resharding
+        while True:
+            # Match scroll sample of points on all nodes exactly
+            data = []
+            for uri in peer_api_uris:
+                r = requests.post(
+                    f"{uri}/collections/{COLLECTION_NAME}/points/scroll", json={
+                        "limit": scroll_limit,
+                        "with_vectors": True,
+                        "with_payload": False,
+                    }
+                )
+                assert_http_ok(r)
+                data.append(r.json()["result"])
+            check_data_consistency(data)
+
+            all_completed = True
+            for uri in peer_api_uris:
+                if not check_collection_resharding_operations_count(uri, COLLECTION_NAME, 0):
+                    all_completed = False
+                    break
+            if all_completed:
+                break
+
+        # Assert node shard count
+        for uri in peer_api_uris:
+            assert check_collection_local_shards_count(uri, COLLECTION_NAME, shard_count)
+
+
+# Test point query stability during resharding.
+#
+# On a static collection, this performs resharding a few times and asserts
+# query remains stable on all peers during the whole process.
+def test_resharding_stable_query(tmp_path: pathlib.Path):
+    assert_project_root()
+
+    num_points = 1000
+    query_limit = 10
+
+    # Prevent optimizers messing with point counts
+    env={
+        "QDRANT__STORAGE__OPTIMIZERS__INDEXING_THRESHOLD_KB": "0",
+    }
+
+    peer_api_uris, _peer_dirs, _bootstrap_uri = start_cluster(tmp_path, 3, None, extra_env=env)
+    first_peer_id = get_cluster_info(peer_api_uris[0])['peer_id']
+
+    # Create collection, insert points
+    create_collection(peer_api_uris[0], shard_number=1, replication_factor=3)
+    wait_collection_exists_and_active_on_all_peers(
+        collection_name=COLLECTION_NAME,
+        peer_api_uris=peer_api_uris,
+    )
+    upsert_random_points(peer_api_uris[0], num_points)
+
+    sleep(1)
+
+    # Assert node shard and point sum count
+    for uri in peer_api_uris:
+        assert check_collection_local_shards_count(uri, COLLECTION_NAME, 1)
+        assert check_collection_local_shards_point_count(uri, COLLECTION_NAME, num_points)
+
+    # Match search sample of points on all nodes exactly
+    data = []
+    search_vector = random_dense_vector()
+    for uri in peer_api_uris:
+        r = requests.post(
+            f"{uri}/collections/{COLLECTION_NAME}/points/query", json={
+                "vector": search_vector,
+                "limit": query_limit,
+            }
+        )
+        assert_http_ok(r)
+        data.append(r.json()["result"])
+    check_query_consistency(data)
+
+    # Reshard 3 times in sequence
+    for shard_count in range(2, 5):
+        # Start resharding
+        r = requests.post(
+            f"{peer_api_uris[0]}/collections/{COLLECTION_NAME}/cluster", json={
+                "start_resharding": {
+                    "peer_id": first_peer_id,
+                    "shard_key": None,
+                }
+            })
+        assert_http_ok(r)
+
+        # Wait for resharding to start
+        wait_for_collection_resharding_operations_count(peer_api_uris[0], COLLECTION_NAME, 1)
+
+        # Continuously assert point search samples on all peers, must be stable
+        # Stop once all peers reported completed resharding
+        while True:
+            # Match search sample of points on all nodes exactly
+            data = []
+            search_vector = random_dense_vector()
+            for uri in peer_api_uris:
+                r = requests.post(
+                    f"{uri}/collections/{COLLECTION_NAME}/points/query", json={
+                        "vector": search_vector,
+                        "limit": query_limit,
+                    }
+                )
+                assert_http_ok(r)
+                data.append(r.json()["result"])
+            check_query_consistency(data)
+
+            all_completed = True
+            for uri in peer_api_uris:
+                if not check_collection_resharding_operations_count(uri, COLLECTION_NAME, 0):
+                    all_completed = False
+                    break
+            if all_completed:
+                break
+
+        # Assert node shard count
+        for uri in peer_api_uris:
+            assert check_collection_local_shards_count(uri, COLLECTION_NAME, shard_count)
+
+
 def run_in_background(run, *args, **kwargs):
     p = multiprocessing.Process(target=run, args=args, kwargs=kwargs)
     p.start()
@@ -316,12 +673,12 @@ def check_data_consistency(data):
     for i in range(len(data) - 1):
         j = i + 1
 
-        data_i = data[i]
-        data_j = data[j]
+        data_i = data[i]["points"]
+        data_j = data[j]["points"]
 
         if data_i != data_j:
-            ids_i = set(x.id for x in data_i["points"])
-            ids_j = set(x.id for x in data_j["points"])
+            ids_i = set(x.id for x in data_i)
+            ids_j = set(x.id for x in data_j)
 
             diff = ids_i - ids_j
 
@@ -331,3 +688,33 @@ def check_data_consistency(data):
                 print(f"Diff len between {i} and {j}: {len(diff)}")
 
             assert False, "Data on all nodes should be consistent"
+
+
+def check_query_consistency(data):
+    assert(len(data) > 1)
+
+    for i in range(len(data) - 1):
+        j = i + 1
+
+        data_i = data[i]["points"]
+        data_j = data[j]["points"]
+
+        for item in data_i:
+            if "version" in item:
+                del item["version"]
+        for item in data_j:
+            if "version" in item:
+                del item["version"]
+
+        if data_i != data_j:
+            ids_i = set(x["id"] for x in data_i)
+            ids_j = set(x["id"] for x in data_j)
+
+            diff = ids_i - ids_j
+
+            if len(diff) < 100:
+                print(f"Diff between {i} and {j}: {diff}")
+            else:
+                print(f"Diff len between {i} and {j}: {len(diff)}")
+
+            assert False, "Query results on all nodes should be consistent"

tests/consensus_tests/utils.py

@@ -469,6 +469,14 @@ def wait_peer_added(peer_api_uri: str, expected_size: int = 1, headers={}) -> st
     return get_leader(peer_api_uri, headers=headers)
 
 
+def wait_collection_green(peer_api_uri: str, collection_name: str):
+    try:
+        wait_for(check_collection_green, peer_api_uri, collection_name)
+    except Exception as e:
+        print_clusters_info([peer_api_uri])
+        raise e
+
+
 def wait_for_some_replicas_not_active(peer_api_uri: str, collection_name: str):
     try:
         wait_for(check_some_replicas_not_active, peer_api_uri, collection_name)
@@ -590,6 +598,11 @@ def wait_for_peer_online(peer_api_uri: str, path="/readyz"):
         raise e
 
 
+def check_collection_green(peer_api_uri: str, collection_name: str, expected_status: str = "green") -> bool:
+    collection_cluster_info = get_collection_info(peer_api_uri, collection_name)
+    return collection_cluster_info['status'] == expected_status
+
+
 def check_collection_points_count(peer_api_uri: str, collection_name: str, expected_size: int) -> bool:
     collection_cluster_info = get_collection_info(peer_api_uri, collection_name)
     return collection_cluster_info['points_count'] == expected_size