Merge branch 'main' into redis

Signed-off-by: Anush008 <anushshetty90@gmail.com>

Author: Anush008

README.md

@@ -10,6 +10,7 @@ CLI tool for migrating data to [Qdrant](http://qdrant.tech) with support for res
 * [Chroma](https://trychroma.com/)
 * [Pinecone](https://www.pinecone.io/)
 * [Milvus](https://milvus.io/)
+* [Weaviate](https://weaviate.io/)
 * [Redis](https://redis.io)
 * Another [Qdrant](http://qdrant.tech/) instance
 
@@ -102,7 +103,8 @@ Migrate data from a **Pinecone** database to **Qdrant**:
 
 ```bash
 migration pinecone \
-    --pinecone.host 'https://example-index-12345.svc.region.pinecone.io' \
+    --pinecone.index-host 'https://example-index.svc.region.pinecone.io' \
+    --pinecone.index-name 'example-index' \
     --pinecone.api-key 'optional-pinecone-api-key' \
     --qdrant.url 'https://example.cloud-region.cloud-provider.cloud.qdrant.io:6334' \
     --qdrant.api-key 'optional-qdrant-api-key' \
@@ -114,17 +116,19 @@ With Docker:
 
 ```bash
 docker run --net=host --rm -it registry.cloud.qdrant.io/library/qdrant-migration pinecone \
-    --pinecone.host 'https://example-index-12345.svc.region.pinecone.io' \
+    --pinecone.index-host 'https://example-index.svc.region.pinecone.io' \
     ...
 ```
 
 #### Pinecone Options
 
 | Flag                            | Description                                                     |
 | ------------------------------- | --------------------------------------------------------------- |
-| `--pinecone.api-key`            | Pinecone API key for authentication                             |
-| `--pinecone.host`               | Pinecone index host URL (e.g., `https://your-pinecone-url`)     |
-| `--pinecone.namespace`          | Namespace of the partition to migrate                           |
+| `--pinecone.index-name`              | Pinecone index name.                                       |
+| `--pinecone.index-host`         | Pinecone index host URL (e.g., `https://your-pinecone-url`)     |
+| `--pinecone.api-key`            | Pinecone API key for authentication.                            |
+| `--pinecone.namespace`          | Namespace of the partition to migrate. Optional.                |
+| `--pinecone.service-host`       | Pinecone service host URL. Optional.                            |
 
 #### Qdrant Options
 
@@ -195,6 +199,70 @@ docker run --net=host --rm -it registry.cloud.qdrant.io/library/qdrant-migration
 
 </details>
 
+<details>
+
+<summary><h3>From Weaviate</h3></summary>
+
+Migrate data from a **Weaviate** database to **Qdrant**:
+
+### 📥 Example
+
+> Important ⚠️:
+ > Weaviate does not expose vector dimensions and distance metric after a collection is created. [Reference](https://forum.weaviate.io/t/get-vector-dimension-of-a-collection/1769/).
+ > Therefore, you must [manually create](https://qdrant.tech/documentation/concepts/collections/#create-a-collection) a Qdrant collection before starting the migration.
+ > Ensure that the **vector dimensions in Qdrant exactly match** those used in Weaviate.
+
+```bash
+migration weaviate \
+    --weaviate.host 'example.c0.asia-southeast1.gcp.weaviate.cloud' \
+    --weaviate.scheme 'https' \
+    --weaviate.auth-type 'apiKey' \
+    --weaviate.api-key 'optional-api-key' \
+    --weaviate.class-name 'ExampleClass' \
+    --qdrant.url 'http://localhost:6334' \
+    --qdrant.collection 'target-collection' \
+    --migration.batch-size 64
+```
+
+With Docker:
+
+```bash
+docker run --net=host --rm -it registry.cloud.qdrant.io/library/qdrant-migration weaviate \
+    --weaviate.host 'example.c0.asia-southeast1.gcp.weaviate.cloud' \
+    ...
+```
+
+#### Weaviate Options
+
+| Flag                       | Description                                                                                      |
+| -------------------------- | ------------------------------------------------------------------------------------------------ |
+| `--weaviate.host`          | Host of the Weaviate instance (e.g. `localhost:8080`) **(required)**                             |
+| `--weaviate.scheme`        | Scheme of the Weaviate instance (e.g. `http` or `https`) _(default: http)_                       |
+| `--weaviate.class-name`    | Name of the Weaviate class to migrate **(required)**                                             |
+| `--weaviate.auth-type`     | Authentication type _(default: none)_. Options: `none`, `apiKey`, `password`, `client`, `bearer` |
+| `--weaviate.api-key`       | API key for authentication (when `auth-type` is `apiKey`)                                        |
+| `--weaviate.username`      | Username for authentication (when `auth-type` is `password`)                                     |
+| `--weaviate.password`      | Password for authentication (when `auth-type` is `password`)                                     |
+| `--weaviate.scopes`        | Scopes for authentication (when `auth-type` is `password` or `client`)                           |
+| `--weaviate.client-secret` | Client secret for authentication (when `auth-type` is `client`)                                  |
+| `--weaviate.token`         | Bearer token for authentication (when `auth-type` is `bearer`)                                   |
+| `--weaviate.refresh-token` | Refresh token for authentication (when `auth-type` is `bearer`)                                  |
+| `--weaviate.expires-in`    | Access token expiration time in seconds (when `auth-type` is `bearer`)                           |
+| `--weaviate.tenant`        | Objects belonging to which tenant to migrate                                                     |
+
+#### Qdrant Options
+
+| Flag                    | Description                                                                                                      |
+| ----------------------- | ---------------------------------------------------------------------------------------------------------------- |
+| `--qdrant.url`          | Qdrant gRPC URL. Default: `"http://localhost:6334"`                                                              |
+| `--qdrant.collection`   | Target collection name                                                                                           |
+| `--qdrant.api-key`      | Qdrant API key                                                                                                   |
+
+* See [Shared Migration Options](#shared-migration-options) for common migration parameters.
+
+</details>
+
+
 <details>
 
 <summary><h3>From Redis</h3></summary>

cmd/migrate_from_chroma.go

@@ -9,7 +9,6 @@ import (
 	"os"
 	"os/signal"
 	"syscall"
-	"time"
 
 	chroma "github.com/amikos-tech/chroma-go/pkg/api/v2"
 	"github.com/pterm/pterm"
@@ -195,12 +194,11 @@ func (r *MigrateFromChromaCmd) prepareTargetCollection(ctx context.Context, coll
 		return fmt.Errorf("failed to create target collection: %w", err)
 	}
 
-	pterm.Success.Printfln("Created target collection '%s' with dimension", r.Qdrant.Collection)
+	pterm.Success.Printfln("Created target collection '%s'", r.Qdrant.Collection)
 	return nil
 }
 
 func (r *MigrateFromChromaCmd) migrateData(ctx context.Context, collection chroma.Collection, targetClient *qdrant.Client, sourcePointCount uint64) error {
-	startTime := time.Now()
 	batchSize := r.Migration.BatchSize
 
 	var currentOffset uint64 = 0
@@ -295,16 +293,6 @@ func (r *MigrateFromChromaCmd) migrateData(ctx context.Context, collection chrom
 		}
 
 		bar.Add(count)
-
-		// 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.countChromaVectors(ctx, collection)
-			if err != nil {
-				return fmt.Errorf("failed to count vectors in Chroma: %w", err)
-			}
-			bar.Total = int(sourcePointCount)
-		}
 	}
 
 	pterm.Success.Printfln("Data migration finished successfully")

cmd/migrate_from_milvus.go

@@ -8,7 +8,6 @@ import (
 	"os/signal"
 	"strconv"
 	"syscall"
-	"time"
 
 	"github.com/milvus-io/milvus/client/v2/column"
 	"github.com/milvus-io/milvus/client/v2/entity"
@@ -182,7 +181,6 @@ func (r *MigrateFromMilvusCmd) prepareTargetCollection(ctx context.Context, sour
 }
 
 func (r *MigrateFromMilvusCmd) migrateData(ctx context.Context, sourceClient *milvusclient.Client, targetClient *qdrant.Client, sourcePointCount uint64) error {
-	startTime := time.Now()
 	batchSize := r.Migration.BatchSize
 
 	var offsetID *qdrant.PointId
@@ -299,15 +297,6 @@ func (r *MigrateFromMilvusCmd) migrateData(ctx context.Context, sourceClient *mi
 			break
 		}
 
-		// 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)
-			if err != nil {
-				return fmt.Errorf("failed to count vectors in Milvus: %w", err)
-			}
-			bar.Total = int(sourcePointCount)
-		}
 	}
 
 	pterm.Success.Printfln("Data migration finished successfully")

cmd/migrate_from_pinecone.go

@@ -5,9 +5,7 @@ import (
 	"fmt"
 	"os"
 	"os/signal"
-	"strings"
 	"syscall"
-	"time"
 
 	"github.com/pinecone-io/go-pinecone/v3/pinecone"
 	"github.com/pterm/pterm"
@@ -30,22 +28,13 @@ type MigrateFromPineconeCmd struct {
 	targetTLS  bool
 }
 
-func stripScheme(input string) string {
-	if idx := strings.Index(input, "://"); idx != -1 {
-		return input[idx+3:]
-	}
-	return input
-}
-
 func (r *MigrateFromPineconeCmd) Parse() error {
 	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.Pinecone.Host = stripScheme(r.Pinecone.Host)
-
 	return nil
 }
 
@@ -89,7 +78,7 @@ func (r *MigrateFromPineconeCmd) Run(globals *Globals) error {
 		return fmt.Errorf("error preparing target collection: %w", err)
 	}
 
-	displayMigrationStart("pinecone", r.Pinecone.Host, r.Qdrant.Collection)
+	displayMigrationStart("pinecone", r.Pinecone.IndexHost, r.Qdrant.Collection)
 
 	err = r.migrateData(ctx, sourceIndexConn, targetClient, sourcePointCount)
 	if err != nil {
@@ -111,14 +100,15 @@ func (r *MigrateFromPineconeCmd) Run(globals *Globals) error {
 
 func (r *MigrateFromPineconeCmd) connectToPinecone() (*pinecone.Client, *pinecone.IndexConnection, error) {
 	client, err := pinecone.NewClient(pinecone.NewClientParams{
+		Host:   r.Pinecone.ServiceHost,
 		ApiKey: r.Pinecone.APIKey,
 	})
 	if err != nil {
 		return nil, nil, fmt.Errorf("failed to create Pinecone client: %w", err)
 	}
 
 	indexConn, err := client.Index(pinecone.NewIndexConnParams{
-		Host:      r.Pinecone.Host,
+		Host:      r.Pinecone.IndexHost,
 		Namespace: r.Pinecone.Namespace,
 	})
 	if err != nil {
@@ -159,14 +149,14 @@ func (r *MigrateFromPineconeCmd) prepareTargetCollection(ctx context.Context, so
 
 	var foundIndex *pinecone.Index
 	for i := range indexes {
-		if indexes[i].Host == r.Pinecone.Host {
+		if indexes[i].Name == r.Pinecone.IndexName {
 			foundIndex = indexes[i]
 			break
 		}
 	}
 
 	if foundIndex == nil {
-		return fmt.Errorf("index %q not found in Pinecone", r.Pinecone.Host)
+		return fmt.Errorf("index %q not found in Pinecone", r.Pinecone.IndexName)
 	}
 
 	distanceMapping := map[pinecone.IndexMetric]qdrant.Distance{
@@ -208,14 +198,13 @@ func (r *MigrateFromPineconeCmd) prepareTargetCollection(ctx context.Context, so
 }
 
 func (r *MigrateFromPineconeCmd) migrateData(ctx context.Context, sourceIndexConn *pinecone.IndexConnection, targetClient *qdrant.Client, sourcePointCount uint64) error {
-	startTime := time.Now()
 	batchSize := r.Migration.BatchSize
 
 	var offsetId *qdrant.PointId
 	offsetCount := uint64(0)
 
 	if !r.Migration.Restart {
-		id, offsetStored, err := commons.GetStartOffset(ctx, r.Migration.OffsetsCollection, targetClient, r.Pinecone.Host)
+		id, offsetStored, err := commons.GetStartOffset(ctx, r.Migration.OffsetsCollection, targetClient, r.Pinecone.IndexHost)
 		if err != nil {
 			return fmt.Errorf("failed to get start offset: %w", err)
 		}
@@ -241,7 +230,7 @@ func (r *MigrateFromPineconeCmd) migrateData(ctx context.Context, sourceIndexCon
 		}
 
 		if len(listRes.VectorIds) < 1 {
-			return fmt.Errorf("pinecone.ListVectors returned no IDs")
+			break
 		}
 
 		ids := make([]string, 0, len(listRes.VectorIds))
@@ -300,7 +289,7 @@ func (r *MigrateFromPineconeCmd) migrateData(ctx context.Context, sourceIndexCon
 		if listRes.NextPaginationToken != nil {
 			offsetCount += uint64(len(targetPoints))
 			offsetId = qdrant.NewID(*listRes.NextPaginationToken)
-			err = commons.StoreStartOffset(ctx, r.Migration.OffsetsCollection, targetClient, r.Pinecone.Host, offsetId, offsetCount)
+			err = commons.StoreStartOffset(ctx, r.Migration.OffsetsCollection, targetClient, r.Pinecone.IndexHost, offsetId, offsetCount)
 			if err != nil {
 				return fmt.Errorf("failed to store offset: %w", err)
 			}
@@ -312,15 +301,6 @@ func (r *MigrateFromPineconeCmd) migrateData(ctx context.Context, sourceIndexCon
 			break
 		}
 
-		// 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.countPineconeVectors(ctx, sourceIndexConn)
-			if err != nil {
-				return fmt.Errorf("failed to count vectors in Pinecone: %w", err)
-			}
-			bar.Total = int(sourcePointCount)
-		}
 	}
 
 	pterm.Success.Printfln("Data migration finished successfully")

cmd/migrate_from_qdrant.go

@@ -6,7 +6,6 @@ import (
 	"os"
 	"os/signal"
 	"syscall"
-	"time"
 
 	"github.com/pterm/pterm"
 
@@ -213,7 +212,6 @@ func getFieldType(dataType qdrant.PayloadSchemaType) *qdrant.FieldType {
 }
 
 func (r *MigrateFromQdrantCmd) migrateData(ctx context.Context, sourceClient *qdrant.Client, sourceCollection string, targetClient *qdrant.Client, targetCollection string, sourcePointCount uint64) error {
-	startTime := time.Now()
 	limit := uint32(r.Migration.BatchSize)
 
 	var offsetId *qdrant.PointId
@@ -323,18 +321,6 @@ func (r *MigrateFromQdrantCmd) migrateData(ctx context.Context, sourceClient *qd
 			break
 		}
 
-		// 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 := sourceClient.Count(ctx, &qdrant.CountPoints{
-				CollectionName: sourceCollection,
-				Exact:          qdrant.PtrOf(true),
-			})
-			if err != nil {
-				return fmt.Errorf("failed to count points in source: %w", err)
-			}
-			bar.Total = int(sourcePointCount)
-		}
 	}
 
 	pterm.Success.Printfln("Data migration finished successfully")