qdrant · generall · Jan 22, 2023 · Jan 12, 2023 · Jan 17, 2023 · Jan 17, 2023
diff --git a/docs/grpc/docs.md b/docs/grpc/docs.md
@@ -333,6 +333,7 @@
 | vectors_config | [VectorsConfig](#qdrant-VectorsConfig) | optional | Configuration for vectors |
 | replication_factor | [uint32](#uint32) | optional | Number of replicas of each shard that network tries to maintain, default = 1 |
 | write_consistency_factor | [uint32](#uint32) | optional | How many replicas should apply the operation for us to consider it successful, default = 1 |
+| init_from_collection | [string](#string) | optional | Specify name of the other collection to copy data from |
 
 
 

diff --git a/docs/redoc/master/openapi.json b/docs/redoc/master/openapi.json
@@ -4261,6 +4261,18 @@
                 "nullable": true
               }
             ]
+          },
+          "init_from": {
+            "description": "Specify other collection to copy data from.",
+            "default": null,
+            "anyOf": [
+              {
+                "$ref": "#/components/schemas/InitFrom"
+              },
+              {
+                "nullable": true
+              }
+            ]
           }
         }
       },
@@ -4391,6 +4403,18 @@
           }
         }
       },
+      "InitFrom": {
+        "description": "Operation for creating new collection and (optionally) specify index params",
+        "type": "object",
+        "required": [
+          "collection"
+        ],
+        "properties": {
+          "collection": {
+            "type": "string"
+          }
+        }
+      },
       "UpdateCollection": {
         "description": "Operation for updating parameters of the existing collection",
         "type": "object",

diff --git a/lib/api/src/grpc/proto/collections.proto b/lib/api/src/grpc/proto/collections.proto
@@ -167,6 +167,7 @@ message CreateCollection {
   optional VectorsConfig vectors_config = 10; // Configuration for vectors
   optional uint32 replication_factor = 11; // Number of replicas of each shard that network tries to maintain, default = 1
   optional uint32 write_consistency_factor = 12; // How many replicas should apply the operation for us to consider it successful, default = 1
+  optional string init_from_collection = 13; // Specify name of the other collection to copy data from
 }
 
 message UpdateCollection {

diff --git a/lib/api/src/grpc/qdrant.rs b/lib/api/src/grpc/qdrant.rs
@@ -202,6 +202,9 @@ pub struct CreateCollection {
     /// How many replicas should apply the operation for us to consider it successful, default = 1
     #[prost(uint32, optional, tag = "12")]
     pub write_consistency_factor: ::core::option::Option<u32>,
+    /// Specify name of the other collection to copy data from
+    #[prost(string, optional, tag = "13")]
+    pub init_from_collection: ::core::option::Option<::prost::alloc::string::String>,
 }
 #[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]

diff --git a/lib/collection/src/collection.rs b/lib/collection/src/collection.rs
@@ -18,9 +18,10 @@ use semver::Version;
 use tar::Builder as TarBuilder;
 use tokio::fs::{copy, create_dir_all, remove_dir_all, remove_file, rename};
 use tokio::runtime::Handle;
-use tokio::sync::{Mutex, RwLock};
+use tokio::sync::{Mutex, RwLock, RwLockWriteGuard};
 
 use crate::collection_state::{ShardInfo, State};
+use crate::common::is_ready::IsReady;
 use crate::config::CollectionConfig;
 use crate::hash_ring::HashRing;
 use crate::operations::config_diff::{CollectionParamsDiff, DiffConfig, OptimizersConfigDiff};
@@ -81,6 +82,13 @@ pub struct Collection {
     #[allow(dead_code)] //Might be useful in case of repartition implementation
     notify_peer_failure_cb: OnPeerFailure,
     init_time: Duration,
+    // One-way boolean flag that is set to true when the collection is fully initialized
+    // i.e. all shards are activated for the first time.
+    is_initialized: Arc<IsReady>,
+    // Lock to temporary block collection update operations while the collection is being migrated.
+    // Lock is acquired for read on update operation and can be acquired for write externally,
+    // which will block all update operations until the lock is released.
+    updates_lock: RwLock<()>,
 }
 
 impl Collection {
@@ -151,6 +159,8 @@ impl Collection {
             request_shard_transfer_cb: request_shard_transfer.clone(),
             notify_peer_failure_cb: on_replica_failure.clone(),
             init_time: start_time.elapsed(),
+            is_initialized: Arc::new(Default::default()),
+            updates_lock: RwLock::new(()),
         })
     }
 
@@ -250,6 +260,8 @@ impl Collection {
             request_shard_transfer_cb: request_shard_transfer.clone(),
             notify_peer_failure_cb: on_replica_failure,
             init_time: start_time.elapsed(),
+            is_initialized: Arc::new(Default::default()),
+            updates_lock: RwLock::new(()),
         }
     }
 
@@ -328,6 +340,25 @@ impl Collection {
             }
         }
 
+        if !self.is_initialized.check_ready() {
+            // If not initialized yet, we need to check if it was initialized by this call
+            let state = self.state().await;
+            let mut is_fully_active = true;
+            for (_shard_id, shard_info) in state.shards {
+                if shard_info
+                    .replicas
+                    .into_iter()
+                    .any(|(_peer_id, state)| state != ReplicaState::Active)
+                {
+                    is_fully_active = false;
+                    break;
+                }
+            }
+            if is_fully_active {
+                self.is_initialized.make_ready();
+            }
+        }
+
         // Try to request shard transfer if replicas on the current peer are dead
         if state == ReplicaState::Dead && self.this_peer_id == peer_id {
             let transfer_from = replica_set
@@ -624,6 +655,7 @@ impl Collection {
         shard_selection: ShardId,
         wait: bool,
     ) -> CollectionResult<UpdateResult> {
+        let _update_lock = self.updates_lock.read().await;
         let shard_holder_guard = self.shards_holder.read().await;
 
         let res = match shard_holder_guard.get_shard(&shard_selection) {
@@ -647,6 +679,7 @@ impl Collection {
         wait: bool,
     ) -> CollectionResult<UpdateResult> {
         operation.validate()?;
+        let _update_lock = self.updates_lock.read().await;
 
         let mut results = {
             let shards_holder = self.shards_holder.read().await;
@@ -1418,6 +1451,14 @@ impl Collection {
 
         Ok(())
     }
+
+    pub fn wait_collection_initiated(&self, timeout: Duration) -> bool {
+        self.is_initialized.await_ready_for_timeout(timeout)
+    }
+
+    pub async fn lock_updates(&self) -> RwLockWriteGuard<()> {
+        self.updates_lock.write().await
+    }
 }
 
 impl Drop for Collection {

diff --git a/...src/content_manager/consensus/is_ready.rs → lib/collection/src/common/is_ready.rs b/...src/content_manager/consensus/is_ready.rs → lib/collection/src/common/is_ready.rs
diff --git a/lib/collection/src/common/mod.rs b/lib/collection/src/common/mod.rs
@@ -1,2 +1,3 @@
+pub mod is_ready;
 pub mod stoppable_task;
 pub mod stoppable_task_async;
diff --git a/lib/collection/src/lib.rs b/lib/collection/src/lib.rs
@@ -3,7 +3,7 @@
 pub mod collection;
 pub mod collection_manager;
 pub mod collection_state;
-mod common;
+pub mod common;
 pub mod config;
 pub mod hash_ring;
 pub mod operations;

diff --git a/lib/storage/src/content_manager/collection_meta_ops.rs b/lib/storage/src/content_manager/collection_meta_ops.rs
@@ -85,6 +85,13 @@ impl From<RenameAlias> for AliasOperations {
     }
 }
 
+/// Operation for creating new collection and (optionally) specify index params
+#[derive(Debug, Deserialize, Serialize, JsonSchema, PartialEq, Eq, Hash, Clone)]
+#[serde(rename_all = "snake_case")]
+pub struct InitFrom {
+    pub collection: CollectionId,
+}
+
 /// Operation for creating new collection and (optionally) specify index params
 #[derive(Debug, Deserialize, Serialize, JsonSchema, PartialEq, Eq, Hash, Clone)]
 #[serde(rename_all = "snake_case")]
@@ -120,6 +127,9 @@ pub struct CreateCollection {
     pub wal_config: Option<WalConfigDiff>,
     /// Custom params for Optimizers.  If none - values from service configuration file are used.
     pub optimizers_config: Option<OptimizersConfigDiff>,
+    /// Specify other collection to copy data from.
+    #[serde(default)]
+    pub init_from: Option<InitFrom>,
 }
 
 /// Operation for creating new collection and (optionally) specify index params

diff --git a/lib/storage/src/content_manager/collections_ops.rs b/lib/storage/src/content_manager/collections_ops.rs
@@ -2,10 +2,11 @@ use std::collections::HashMap;
 
 use async_trait::async_trait;
 use collection::collection::Collection;
+use collection::shards::CollectionId;
 
 use crate::content_manager::errors::StorageError;
 
-pub type Collections = HashMap<String, Collection>;
+pub type Collections = HashMap<CollectionId, Collection>;
 
 #[async_trait]
 pub trait Checker {

diff --git a/lib/storage/src/content_manager/consensus/mod.rs b/lib/storage/src/content_manager/consensus/mod.rs
@@ -1,5 +1,4 @@
 pub mod consensus_wal;
 pub mod entry_queue;
-pub mod is_ready;
 pub mod operation_sender;
 pub mod persistent;
diff --git a/lib/storage/src/content_manager/consensus_manager.rs b/lib/storage/src/content_manager/consensus_manager.rs
@@ -9,6 +9,7 @@ use std::time::Duration;
 use anyhow::{anyhow, Context};
 use chrono::Utc;
 use collection::collection_state;
+use collection::common::is_ready::IsReady;
 use collection::shards::shard::PeerId;
 use collection::shards::CollectionId;
 use futures::future::join_all;
@@ -27,7 +28,6 @@ use super::errors::StorageError;
 use super::CollectionContainer;
 use crate::content_manager::consensus::consensus_wal::ConsensusOpWal;
 use crate::content_manager::consensus::entry_queue::EntryId;
-use crate::content_manager::consensus::is_ready::IsReady;
 use crate::content_manager::consensus::operation_sender::OperationSender;
 use crate::content_manager::consensus::persistent::Persistent;
 use crate::types::{

diff --git a/lib/storage/src/content_manager/conversions.rs b/lib/storage/src/content_manager/conversions.rs
@@ -6,7 +6,7 @@ use tonic::Status;
 use crate::content_manager::collection_meta_ops::{
     AliasOperations, ChangeAliasesOperation, CollectionMetaOperations, CreateAlias,
     CreateAliasOperation, CreateCollection, CreateCollectionOperation, DeleteAlias,
-    DeleteAliasOperation, DeleteCollectionOperation, RenameAlias, RenameAliasOperation,
+    DeleteAliasOperation, DeleteCollectionOperation, InitFrom, RenameAlias, RenameAliasOperation,
     UpdateCollection, UpdateCollectionOperation,
 };
 use crate::content_manager::errors::StorageError;
@@ -56,6 +56,9 @@ impl TryFrom<api::grpc::qdrant::CreateCollection> for CollectionMetaOperations {
                 on_disk_payload: value.on_disk_payload,
                 replication_factor: value.replication_factor,
                 write_consistency_factor: value.write_consistency_factor,
+                init_from: value
+                    .init_from_collection
+                    .map(|v| InitFrom { collection: v }),
             },
         )))
     }