Add logs into the client + parallelize client upload.

Author: clems4ever

go.mod

@@ -18,7 +18,7 @@ require (
 	github.com/stretchr/testify v1.4.0
 	golang.org/x/sync v0.0.0-20201207232520-09787c993a3a
 	gopkg.in/VividCortex/ewma.v1 v1.1.1 // indirect
-	gopkg.in/cheggaaa/pb.v2 v2.0.7 // indirect
+	gopkg.in/cheggaaa/pb.v2 v2.0.7
 	gopkg.in/fatih/color.v1 v1.7.0 // indirect
 	gopkg.in/mattn/go-colorable.v0 v0.1.0 // indirect
 	gopkg.in/mattn/go-isatty.v0 v0.0.4 // indirect

internal/client/transaction.go

@@ -6,6 +6,8 @@ import (
 	"sync"
 	"time"
 
+	"gopkg.in/cheggaaa/pb.v2"
+
 	"github.com/clems4ever/go-graphkb/internal/knowledge"
 	"github.com/clems4ever/go-graphkb/internal/schema"
 )
@@ -64,34 +66,71 @@ func withRetryOnTooManyRequests(fn func() error, maxRetries int) error {
 func (cgt *Transaction) Commit() (*knowledge.Graph, error) {
 	sg := cgt.newGraph.ExtractSchema()
 
+	fmt.Println("Start uploading the schema of the graph...")
 	if err := cgt.client.UpdateSchema(sg); err != nil {
 		return nil, fmt.Errorf("Unable to update the schema of the graph")
 	}
 
+	fmt.Println("Finished uploading the schema of the graph...")
+
 	bulk := knowledge.GenerateGraphUpdatesBulk(cgt.currentGraph, cgt.newGraph)
 
-	for _, r := range bulk.GetRelationRemovals() {
-		if err := withRetryOnTooManyRequests(func() error { return cgt.client.DeleteRelation(r) }, 10); err != nil {
-			return nil, fmt.Errorf("Unable to remove the relation %v: %v", r, err)
+	fmt.Println("Start uploading the graph...")
+
+	progress := pb.New(len(bulk.GetAssetRemovals()) + len(bulk.GetAssetUpserts()) + len(bulk.GetRelationRemovals()) + len(bulk.GetRelationUpserts()))
+	progress.Start()
+	defer progress.Finish()
+
+	var wg sync.WaitGroup
+
+	wg.Add(2)
+
+	var err1, err2 error
+
+	go func() {
+		defer wg.Done()
+		for _, r := range bulk.GetRelationRemovals() {
+			if err := withRetryOnTooManyRequests(func() error { return cgt.client.DeleteRelation(r) }, 10); err != nil {
+				err1 = fmt.Errorf("Unable to remove the relation %v: %v", r, err)
+				return
+			}
+			progress.Increment()
 		}
-	}
-	for _, a := range bulk.GetAssetRemovals() {
-		if err := withRetryOnTooManyRequests(func() error { return cgt.client.DeleteAsset(a) }, 10); err != nil {
-			return nil, fmt.Errorf("Unable to remove the asset %v: %v", a, err)
+		for _, a := range bulk.GetAssetRemovals() {
+			if err := withRetryOnTooManyRequests(func() error { return cgt.client.DeleteAsset(a) }, 10); err != nil {
+				err1 = fmt.Errorf("Unable to remove the asset %v: %v", a, err)
+			}
+			progress.Increment()
 		}
-	}
+	}()
 
-	for _, a := range bulk.GetAssetUpserts() {
-		if err := withRetryOnTooManyRequests(func() error { return cgt.client.UpsertAsset(a) }, 10); err != nil {
-			return nil, fmt.Errorf("Unable to upsert the asset %v: %v", a, err)
+	go func() {
+		defer wg.Done()
+		for _, a := range bulk.GetAssetUpserts() {
+			if err := withRetryOnTooManyRequests(func() error { return cgt.client.UpsertAsset(a) }, 10); err != nil {
+				err2 = fmt.Errorf("Unable to upsert the asset %v: %v", a, err)
+			}
+			progress.Increment()
 		}
-	}
-	for _, r := range bulk.GetRelationUpserts() {
-		if err := withRetryOnTooManyRequests(func() error { return cgt.client.UpsertRelation(r) }, 10); err != nil {
-			return nil, fmt.Errorf("Unable to upsert the relation %v: %v", r, err)
+		for _, r := range bulk.GetRelationUpserts() {
+			if err := withRetryOnTooManyRequests(func() error { return cgt.client.UpsertRelation(r) }, 10); err != nil {
+				err2 = fmt.Errorf("Unable to upsert the relation %v: %v", r, err)
+			}
+			progress.Increment()
 		}
+	}()
+
+	wg.Wait()
+
+	if err1 != nil {
+		return nil, err1
+	}
+	if err2 != nil {
+		return nil, err2
 	}
 
+	fmt.Println("Finished uploading the graph...")
+
 	g := cgt.newGraph
 	cgt.newGraph = knowledge.NewGraph()
 	return g, nil // give ownership of the transaction graph so that it can be cached if needed

internal/database/mariadb.go

@@ -81,7 +81,7 @@ CREATE TABLE IF NOT EXISTS relations (
 	CONSTRAINT pk_relation PRIMARY KEY (id),
 	CONSTRAINT fk_from FOREIGN KEY (from_id) REFERENCES assets (id),
 	CONSTRAINT fk_to FOREIGN KEY (to_id) REFERENCES assets (id),
-	CONSTRAINT fk_relation_source FOREIGN KEY (source_id) REFERENCES sources (id),
+	CONSTRAINT fk_relation_source FOREIGN KEY (source_id) REFERENCES sources (id) ON DELETE CASCADE,
 
 	INDEX full_relation_type_from_to_idx (type, from_id, to_id),
     INDEX full_relation_type_to_from_idx (type, to_id, from_id),