qdrant
diff --git a/‎README.md‎
Lines changed: 30 additions & 7 deletions b/‎README.md‎
Lines changed: 30 additions & 7 deletions
diff --git a/‎cmd/migrate_from_milvus.go‎
Lines changed: 24 additions & 62 deletions b/‎cmd/migrate_from_milvus.go‎
Lines changed: 24 additions & 62 deletions
@@ -2,17 +2,18 @@
 
 CLI tool for migrating data to [Qdrant](http://qdrant.tech) with support for resumable transfers in case of interruptions.
 
-> [!WARNING]  
-> This project is in beta. The API may change in future releases.
+Easily move your data to Qdrant from other vector storages. With support for resumable migration, even interrupted processes can continue smoothly.
 
-## Supported Sources
+Supported sources:
 
-* [Milvus](https://milvus.io)
-* Another [Qdrant](http://qdrant.tech) instance
+* Milvus
+* Another Qdrant instance
 
 ## Installation
 
-You can run this tool on any machine with connectivity to both the source and the Qdrant database. For best performance, use a machine with a fast network and minimal latency to both endpoints.
+The easiest way to run the qdrant-migration tool is as a container. You can run it any machine where you have connectivity to both the source and the target Qdrant databases. For optimal performance, you should run the tool on a machine with a fast network connection and minimum latency to both databases.
+
+To pull the latest image run:
 
 #### Binaries
 
@@ -23,7 +24,29 @@ Each release includes **precompiled binaries** for all major OS and CPU architec
 To get the latest Docker image run the following command.
 
 ```bash
-docker pull registry.cloud.qdrant.io/library/qdrant-migration
+$ docker run --net=host --rm -it registry.cloud.qdrant.io/library/qdrant-migration qdrant --help
+Usage: migration qdrant --source-url=STRING --source-collection=STRING --target-url=STRING --target-collection=STRING [flags]
+
+Migrate data from a Qdrant database to Qdrant.
+
+Flags:
+  -h, --help                                                      Show context-sensitive help.
+      --debug                                                     Enable debug mode.
+      --trace                                                     Enable trace mode.
+      --skip-tls-verification                                     Skip TLS verification.
+      --version                                                   Print version information and quit
+
+      --source-url=STRING                                         Source gRPC URL, e.g. https://your-qdrant-hostname:6334
+      --source-collection=STRING                                  Source collection
+      --source-api-key=STRING                                     Source API key ($SOURCE_API_KEY)
+      --target-url=STRING                                         Target gRPC URL, e.g. https://your-qdrant-hostname:6334
+      --target-collection=STRING                                  Target collection
+      --target-api-key=STRING                                     Target API key ($TARGET_API_KEY)
+  -b, --batch-size=50                                             Batch size
+  -c, --create-target-collection                                  Create the target collection if it does not exist
+      --ensure-payload-indexes                                    Ensure payload indexes are created
+      --migration-offsets-collection-name="_migration_offsets"    Collection where the current migration offset should be stored
+      --restart-migration                                         Restart the migration and do not continue from last offset
 ```
 
 ## How To Migrate?
 
@@ -4,7 +4,6 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
-	"net/url"
 	"os"
 	"os/signal"
 	"strconv"
@@ -32,27 +31,17 @@ type MigrateFromMilvusCmd struct {
 }
 
 func (r *MigrateFromMilvusCmd) Parse() error {
-	targetUrl, err := url.Parse(r.Qdrant.Url)
+	var err error
+	r.targetHost, r.targetPort, r.targetTLS, err = parseQdrantUrl(r.Qdrant.Url)
 	if err != nil {
 		return fmt.Errorf("failed to parse target URL: %w", err)
 	}
 
-	r.targetHost = targetUrl.Hostname()
-	r.targetTLS = targetUrl.Scheme == HTTPS
-	r.targetPort, err = getPort(targetUrl)
-	if err != nil {
-		return fmt.Errorf("failed to parse target port: %w", err)
-	}
-
 	return nil
 }
 
 func (r *MigrateFromMilvusCmd) Validate() error {
-	if r.Migration.BatchSize < 1 {
-		return fmt.Errorf("batch size must be greater than 0")
-	}
-
-	return nil
+	return validateBatchSize(r.Migration.BatchSize)
 }
 
 func (r *MigrateFromMilvusCmd) Run(globals *Globals) error {
@@ -91,28 +80,14 @@ func (r *MigrateFromMilvusCmd) Run(globals *Globals) error {
 		return fmt.Errorf("error preparing target collection: %w", err)
 	}
 
-	targetPointCount, err := targetClient.Count(ctx, &qdrant.CountPoints{
-		CollectionName: r.Qdrant.Collection,
-		Exact:          qdrant.PtrOf(true),
-	})
-	if err != nil {
-		return fmt.Errorf("failed to count points in target: %w", err)
-	}
-
-	pterm.DefaultSection.Println("Starting data migration")
-
-	_ = pterm.DefaultTable.WithHasHeader().WithData(pterm.TableData{
-		{"Type", "Provider", "Collection", "Points"},
-		{"Source", "milvus", r.Milvus.Collection, strconv.FormatUint(sourcePointCount, 10)},
-		{"Target", "qdrant", r.Qdrant.Collection, strconv.FormatUint(targetPointCount, 10)},
-	}).Render()
+	displayMigrationStart("milvus", r.Milvus.Collection, r.Qdrant.Collection)
 
 	err = r.migrateData(ctx, sourceClient, targetClient, sourcePointCount)
 	if err != nil {
 		return fmt.Errorf("failed to migrate data: %w", err)
 	}
 
-	targetPointCount, err = targetClient.Count(ctx, &qdrant.CountPoints{
+	targetPointCount, err := targetClient.Count(ctx, &qdrant.CountPoints{
 		CollectionName: r.Qdrant.Collection,
 		Exact:          qdrant.PtrOf(true),
 	})
@@ -154,7 +129,6 @@ func (r *MigrateFromMilvusCmd) countMilvusVectors(ctx context.Context, client *m
 	}
 
 	return count, nil
-
 }
 
 func (r *MigrateFromMilvusCmd) prepareTargetCollection(ctx context.Context, sourceClient *milvusclient.Client, targetClient *qdrant.Client) error {
@@ -168,7 +142,7 @@ func (r *MigrateFromMilvusCmd) prepareTargetCollection(ctx context.Context, sour
 	}
 
 	if targetCollectionExists {
-		pterm.Info.Printfln("Target collection '%s' already exists. Skipping creation.", r.Qdrant.Collection)
+		pterm.Info.Printfln("Target collection %q already exists. Skipping creation.", r.Qdrant.Collection)
 		return nil
 	}
 
@@ -203,36 +177,29 @@ func (r *MigrateFromMilvusCmd) prepareTargetCollection(ctx context.Context, sour
 		return fmt.Errorf("failed to create target collection: %w", err)
 	}
 
-	pterm.Success.Printfln("Created target collection '%s'", r.Qdrant.Collection)
+	pterm.Success.Printfln("Created target collection %q", r.Qdrant.Collection)
 	return nil
 }
 
 func (r *MigrateFromMilvusCmd) migrateData(ctx context.Context, sourceClient *milvusclient.Client, targetClient *qdrant.Client, sourcePointCount uint64) error {
 	startTime := time.Now()
 	batchSize := r.Migration.BatchSize
 
-	var lastID *qdrant.PointId
+	var offsetID *qdrant.PointId
 	offsetCount := uint64(0)
 	var err error
 
 	if !r.Migration.Restart {
-		offsetId, offsetStored, err := commons.GetStartOffset(ctx, r.Migration.OffsetsCollection, targetClient, r.Milvus.Collection, r.Migration.Restart)
+		id, count, err := commons.GetStartOffset(ctx, r.Migration.OffsetsCollection, targetClient, r.Milvus.Collection)
 		if err != nil {
 			return fmt.Errorf("failed to get start offset: %w", err)
 		}
-		offsetCount = offsetStored
-		lastID = offsetId
+		offsetCount = count
+		offsetID = id
 	}
 
 	bar, _ := pterm.DefaultProgressbar.WithTotal(int(sourcePointCount)).Start()
-	if offsetCount > 0 {
-		pterm.Info.Printfln("Starting from offset %d", offsetCount)
-		bar.Add(int(offsetCount))
-	} else {
-		pterm.Info.Printfln("Starting from beginning")
-	}
-
-	fmt.Print("\n")
+	displayMigrationProgress(bar, offsetCount)
 
 	schema, err := sourceClient.DescribeCollection(ctx, milvusclient.NewDescribeCollectionOption(r.Milvus.Collection))
 	if err != nil {
@@ -245,31 +212,28 @@ func (r *MigrateFromMilvusCmd) migrateData(ctx context.Context, sourceClient *mi
 
 	for {
 		filter := ""
-		if lastID != nil {
+		if offsetID != nil {
 			switch pkType {
 			case entity.FieldTypeInt64:
-				filter = fmt.Sprintf("%s > %d", pkName, lastID.GetNum())
+				filter = fmt.Sprintf("%s > %d", pkName, offsetID.GetNum())
 			case entity.FieldTypeVarChar:
-				filter = fmt.Sprintf("%s > '%s'", pkName, lastID.GetUuid())
-			default:
-				return fmt.Errorf("unsupported primary key type: %v", pkType)
+				filter = fmt.Sprintf("%s > '%s'", pkName, offsetID.GetUuid())
 			}
 		}
 
 		result, err := sourceClient.Query(ctx, milvusclient.NewQueryOption(r.Milvus.Collection).
 			WithFilter(filter).
 			WithOutputFields("*").
-			WithLimit(batchSize),
-		)
+			WithLimit(batchSize))
 		if err != nil {
-			return fmt.Errorf("query failed: %w", err)
+			return fmt.Errorf("failed to query Milvus: %w", err)
 		}
+
 		if result.ResultCount == 0 {
 			break
 		}
 
 		var targetPoints []*qdrant.PointStruct
-
 		for i := 0; i < result.ResultCount; i++ {
 			point := &qdrant.PointStruct{}
 			vectors := make(map[string]*qdrant.Vector)
@@ -286,12 +250,12 @@ func (r *MigrateFromMilvusCmd) migrateData(ctx context.Context, sourceClient *mi
 					switch col.Type() {
 					case entity.FieldTypeVarChar:
 						uuid := value.(string)
-						lastID = qdrant.NewID(uuid)
-						point.Id = lastID
+						offsetID = qdrant.NewID(uuid)
+						point.Id = offsetID
 					case entity.FieldTypeInt64:
 						num := value.(int64)
-						lastID = qdrant.NewIDNum(uint64(num))
-						point.Id = lastID
+						offsetID = qdrant.NewIDNum(uint64(num))
+						point.Id = offsetID
 					}
 					continue
 				}
@@ -318,13 +282,12 @@ func (r *MigrateFromMilvusCmd) migrateData(ctx context.Context, sourceClient *mi
 			Points:         targetPoints,
 			Wait:           qdrant.PtrOf(true),
 		})
-
 		if err != nil {
 			return fmt.Errorf("failed to insert data into target: %w", err)
 		}
 
 		offsetCount += uint64(len(targetPoints))
-		err = commons.StoreStartOffset(ctx, r.Migration.OffsetsCollection, targetClient, r.Milvus.Collection, lastID, offsetCount)
+		err = commons.StoreStartOffset(ctx, r.Migration.OffsetsCollection, targetClient, r.Milvus.Collection, offsetID, offsetCount)
 		if err != nil {
 			return fmt.Errorf("failed to store offset: %w", err)
 		}
@@ -338,12 +301,11 @@ func (r *MigrateFromMilvusCmd) migrateData(ctx context.Context, sourceClient *mi
 		// If one minute elapsed get updated sourcePointCount.
 		// Useful if any new points were added to the source during migration.
 		if time.Since(startTime) > time.Minute {
-			sourcePointCount, err := r.countMilvusVectors(ctx, sourceClient)
+			sourcePointCount, err = r.countMilvusVectors(ctx, sourceClient)
 			if err != nil {
 				return fmt.Errorf("failed to count points in source: %w", err)
 			}
 			bar.Total = int(sourcePointCount)
-			startTime = time.Now()
 		}
 	}