chore: delete offsets collection after successful migration (#190)

* Improve elastic/open search error handling

Signed-off-by: Nathan LeRoy <nleroy917@users.noreply.github.com>

* chore: Lint

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

* add helper to delete offset collection

* add offsets deletion to all methods

---------

Signed-off-by: Nathan LeRoy <nleroy917@users.noreply.github.com>
Signed-off-by: Anush008 <anushshetty90@gmail.com>
Co-authored-by: Nathan LeRoy <nleroy917@users.noreply.github.com>
Co-authored-by: Anush008 <anushshetty90@gmail.com>

Author: nleroy917

cmd/migrate_from_elasticsearch.go

@@ -100,6 +100,11 @@ func (r *MigrateFromElasticsearchCmd) Run(globals *Globals) error {
 
 	pterm.Info.Printfln("Target collection has %d points\n", targetPointCount)
 
+	err = commons.DeleteOffsetsCollection(ctx, r.Migration.OffsetsCollection, targetClient)
+	if err != nil {
+		return fmt.Errorf("failed to delete migration marker collection: %w", err)
+	}
+
 	return nil
 }
 

cmd/migrate_from_faiss.go

@@ -96,6 +96,12 @@ func (r *MigrateFromFaissCmd) Run(globals *Globals) error {
 	}
 
 	pterm.Info.Printfln("Target collection has %d points\n", targetPointCount)
+
+	err = commons.DeleteOffsetsCollection(ctx, r.Migration.OffsetsCollection, targetClient)
+	if err != nil {
+		return fmt.Errorf("failed to delete migration marker collection: %w", err)
+	}
+
 	return nil
 }
 

cmd/migrate_from_milvus.go

@@ -98,6 +98,11 @@ func (r *MigrateFromMilvusCmd) Run(globals *Globals) error {
 
 	pterm.Info.Printfln("Target collection has %d points\n", targetPointCount)
 
+	err = commons.DeleteOffsetsCollection(ctx, r.Migration.OffsetsCollection, targetClient)
+	if err != nil {
+		return fmt.Errorf("failed to delete migration marker collection: %w", err)
+	}
+
 	return nil
 }
 

cmd/migrate_from_mongo.go

@@ -99,6 +99,11 @@ func (r *MigrateFromMongoDBCmd) Run(globals *Globals) error {
 
 	pterm.Info.Printfln("Target collection has %d points\n", targetPointCount)
 
+	err = commons.DeleteOffsetsCollection(ctx, r.Migration.OffsetsCollection, targetClient)
+	if err != nil {
+		return fmt.Errorf("failed to delete migration marker collection: %w", err)
+	}
+
 	return nil
 }
 

cmd/migrate_from_opensearch.go

@@ -98,6 +98,11 @@ func (r *MigrateFromOpenSearchCmd) Run(globals *Globals) error {
 
 	pterm.Info.Printfln("Target collection has %d points\n", targetPointCount)
 
+	err = commons.DeleteOffsetsCollection(ctx, r.Migration.OffsetsCollection, targetClient)
+	if err != nil {
+		return fmt.Errorf("failed to delete migration marker collection: %w", err)
+	}
+
 	return nil
 }
 

cmd/migrate_from_pg.go

@@ -99,6 +99,11 @@ func (r *MigrateFromPGCmd) Run(globals *Globals) error {
 
 	pterm.Info.Printfln("Target collection has %d points\n", targetPointCount)
 
+	err = commons.DeleteOffsetsCollection(ctx, r.Migration.OffsetsCollection, targetClient)
+	if err != nil {
+		return fmt.Errorf("failed to delete migration marker collection: %w", err)
+	}
+
 	return nil
 }
 

cmd/migrate_from_pinecone.go

@@ -95,6 +95,11 @@ func (r *MigrateFromPineconeCmd) Run(globals *Globals) error {
 
 	pterm.Info.Printfln("Target collection has %d points\n", targetPointCount)
 
+	err = commons.DeleteOffsetsCollection(ctx, r.Migration.OffsetsCollection, targetClient)
+	if err != nil {
+		return fmt.Errorf("failed to delete migration marker collection: %w", err)
+	}
+
 	return nil
 }
 

cmd/migrate_from_qdrant.go

@@ -127,6 +127,12 @@ func (r *MigrateFromQdrantCmd) Run(globals *Globals) error {
 	}
 
 	pterm.Info.Printfln("Target collection has %d points\n", targetPointCount)
+
+	err = commons.DeleteOffsetsCollection(ctx, r.Migration.OffsetsCollection, targetClient)
+	if err != nil {
+		return fmt.Errorf("failed to delete migration marker collection: %w", err)
+	}
+
 	return nil
 }
 

cmd/migrate_from_redis.go

@@ -107,6 +107,11 @@ func (r *MigrateFromRedisCmd) Run(globals *Globals) error {
 
 	pterm.Info.Printfln("Target collection has %d points\n", targetPointCount)
 
+	err = commons.DeleteOffsetsCollection(ctx, r.Migration.OffsetsCollection, targetClient)
+	if err != nil {
+		return fmt.Errorf("failed to delete migration marker collection: %w", err)
+	}
+
 	return nil
 }
 

cmd/migrate_from_s3_vectors.go

@@ -92,6 +92,11 @@ func (r *MigrateFromS3VectorsCmd) Run(globals *Globals) error {
 
 	pterm.Info.Printfln("Target collection has %d points\n", targetPointCount)
 
+	err = commons.DeleteOffsetsCollection(ctx, r.Migration.OffsetsCollection, targetClient)
+	if err != nil {
+		return fmt.Errorf("failed to delete migration marker collection: %w", err)
+	}
+
 	return nil
 }