Repository: cloudspannerecosystem/harbourbridge
Pull request #188 (2021-09-30 07:50:50)

Changed progress reporting for foreign keys to fractions
Changed progress reporting to also support fractional updates. Also changed the maxworkers during foreign key creation to 20.


Patch:
diff --git a/conversion/conversion.go b/conversion/conversion.go
index 0efc092..7493517 100644
--- a/conversion/conversion.go
+++ b/conversion/conversion.go
@@ -83,7 +83,10 @@ const (
 
 var (
 	// Set the maximum number of concurrent workers during foreign key creation.
-	MaxWorkers = 10
+	// This number should not be too high so as to not hit the AdminQuota limit.
+	// AdminQuota limits are mentioned here: https://cloud.google.com/spanner/quotas#administrative_limits
+	// If facing a quota limit error, consider reducing this value.
+	MaxWorkers = 20
 )
 
 func SchemaConv(driver string, targetDb string, ioHelper *IOStreams, schemaSampleSize int64) (*internal.Conv, error) {
@@ -201,7 +204,7 @@ func dataFromSQL(driver string, config spanner.BatchWriterConfig, client *sp.Cli
 		return nil, err
 	}
 	totalRows := conv.Rows()
-	p := internal.NewProgress(totalRows, "Writing data to Spanner", internal.Verbose())
+	p := internal.NewProgress(totalRows, "Writing data to Spanner", internal.Verbose(), false)
 	rows := int64(0)
 	config.Write = func(m []*sp.Mutation) error {
 		_, err := client.Apply(context.Background(), m)
@@ -251,7 +254,7 @@ func dataFromDynamoDB(config spanner.BatchWriterConfig, client *sp.Client, conv
 	dydbClient := dydb.New(mySession, getDynamoDBClientConfig())
 	dynamodb.SetRowStats(conv, dydbClient)
 	totalRows := conv.Rows()
-	p := internal.NewProgress(totalRows, "Writing data to Spanner", internal.Verbose())
+	p := internal.NewProgress(totalRows, "Writing data to Spanner", internal.Verbose(), false)
 
 	rows := int64(0)
 	config.Write = func(m []*sp.Mutation) error {
@@ -310,7 +313,7 @@ func schemaFromDump(driver string, targetDb string, ioHelper *IOStreams) (*inter
 	ioHelper.BytesRead = n
 	conv := internal.MakeConv()
 	conv.TargetDb = targetDb
-	p := internal.NewProgress(n, "Generating schema", internal.Verbose())
+	p := internal.NewProgress(n, "Generating schema", internal.Verbose(), false)
 	r := internal.NewReader(bufio.NewReader(f), p)
 	conv.SetSchemaMode() // Build schema and ignore data in dump.
 	conv.SetDataSink(nil)
@@ -345,7 +348,7 @@ func dataFromDump(driver string, config spanner.BatchWriterConfig, ioHelper *IOS
 	}
 	totalRows := conv.Rows()
 
-	p := internal.NewProgress(totalRows, "Writing data to Spanner", internal.Verbose())
+	p := internal.NewProgress(totalRows, "Writing data to Spanner", internal.Verbose(), false)
 	r := internal.NewReader(bufio.NewReader(ioHelper.SeekableIn), nil)
 	rows := int64(0)
 	config.Write = func(m []*sp.Mutation) error {
@@ -581,8 +584,18 @@ func UpdateDDLForeignKeys(ctx context.Context, adminClient *database.DatabaseAdm
 	if len(fkStmts) == 0 {
 		return nil
 	}
+	if len(fkStmts) > 50 {
+		fmt.Println(`
+Warning: Large number of foreign keys detected. Spanner can take a long amount of 
+time to create foreign keys (over 5 mins per batch of Foreign Keys even with no data). 
+Harbourbridge does not have control over a single foreign key creation time. The number 
+of concurrent Foreign Key Creation Requests sent to spanner can be increased by 
+tweaking the MaxWorkers variable (https://github.com/cloudspannerecosystem/harbourbridge/blob/master/conversion/conversion.go#L89).
+However, setting it to a very high value might lead to exceeding the admin quota limit.
+Recommended value is between 20-30.`)
+	}
 	msg := fmt.Sprintf("Updating schema of database %s with foreign key constraints ...", dbURI)
-	p := internal.NewProgress(int64(len(fkStmts)), msg, internal.Verbose())
+	p := internal.NewProgress(int64(len(fkStmts)), msg, internal.Verbose(), true)
 
 	workers := make(chan int, MaxWorkers)
 	for i := 1; i <= MaxWorkers; i++ {
@@ -612,12 +625,12 @@ func UpdateDDLForeignKeys(ctx context.Context, adminClient *database.DatabaseAdm
 				Statements: []string{fkStmt},
 			})
 			if err != nil {
-				fmt.Printf("Cannot submit request for create foreign key with statement %s: %s\n", fkStmt, err)
+				fmt.Printf("Cannot submit request for create foreign key with statement: %s\n due to error: %s. Skipping this foreign key...\n", fkStmt, err)
 				conv.Unexpected(fmt.Sprintf("Can't add foreign key with statement %s: %s", fkStmt, err))
 				return
 			}
 			if err := op.Wait(ctx); err != nil {
-				fmt.Printf("Can't add foreign key with statement %s: %s\n", fkStmt, err)
+				fmt.Printf("Can't add foreign key with statement: %s\n due to error: %s. Skipping this foreign key...\n", fkStmt, err)
 				conv.Unexpected(fmt.Sprintf("Can't add foreign key with statement %s: %s", fkStmt, err))
 				return
 			}
diff --git a/internal/progress.go b/internal/progress.go
index 8902f8d..627150f 100644
--- a/internal/progress.go
+++ b/internal/progress.go
@@ -16,26 +16,35 @@
 // HarbourBridge.
 package internal
 
-import "fmt"
+import (
+	"fmt"
+	"strconv"
+	"strings"
+)
 
 // Progress provides console progress functionality. i.e. it reports what
 // percentage of a task is complete to the console, overwriting previous
 // progress percentage with new progress.
 type Progress struct {
-	total    int64  // How much we have to do.
-	progress int64  // How much we have done so far.
-	pct      int    // Percentage done i.e. progress/total * 100
-	message  string // Name of task being monitored.
-	verbose  bool   // If true, print detailed info about each progress step.
+	total      int64  // How much we have to do.
+	progress   int64  // How much we have done so far.
+	pct        int    // Percentage done i.e. progress/total * 100
+	message    string // Name of task being monitored.
+	verbose    bool   // If true, print detailed info about each progress step.
+	fractional bool   // If true, report progress in fractions instead of percentages.
 }
 
 // NewProgress creates and returns a Progress instance.
-func NewProgress(total int64, message string, verbose bool) *Progress {
-	p := &Progress{total, 0, 0, message, verbose}
+func NewProgress(total int64, message string, verbose, fractional bool) *Progress {
+	p := &Progress{total, 0, 0, message, verbose, fractional}
 	if total == 0 {
 		p.pct = 100
 	}
-	p.report(true)
+	if p.fractional {
+		p.reportFraction(true)
+	} else {
+		p.reportPct(true)
+	}
 	return p
 }
 
@@ -46,6 +55,10 @@ func NewProgress(total int64, message string, verbose bool) *Progress {
 func (p *Progress) MaybeReport(progress int64) {
 	if progress > p.progress {
 		p.progress = progress
+		if p.fractional {
+			p.reportFraction(false)
+			return
+		}
 		var pct int
 		if p.total > 0 {
 			pct = int((progress * 100) / p.total)
@@ -57,7 +70,7 @@ func (p *Progress) MaybeReport(progress int64) {
 		}
 		if pct > p.pct {
 			p.pct = pct
-			p.report(false)
+			p.reportPct(false)
 		}
 
 	}
@@ -69,7 +82,7 @@ func (p *Progress) Done() {
 	p.MaybeReport(p.total)
 }
 
-func (p *Progress) report(firstCall bool) {
+func (p *Progress) reportPct(firstCall bool) {
 	if p.verbose {
 		fmt.Printf("%s: %2d%%\n", p.message, p.pct)
 		return
@@ -83,3 +96,21 @@ func (p *Progress) report(firstCall bool) {
 		fmt.Printf("\n")
 	}
 }
+
+func (p *Progress) reportFraction(firstCall bool) {
+	if p.verbose {
+		fmt.Printf("%s: %d/%d\n", p.message, p.progress, p.total)
+		return
+	}
+	if firstCall {
+		fmt.Printf("%s: %d/%d", p.message, p.progress, p.total)
+	} else {
+		// We delete the previous outputted fraction hence, adding backspace that many number of times.
+		// The length of the previous outputted fraction would be len(p.progress-1) +len(total) and 1 for the "/".
+		delStr := strings.Repeat("\b", len(strconv.Itoa(int(p.progress-1)))+1+len(strconv.Itoa(int(p.total))))
+		fmt.Printf(delStr+"%d/%d", p.progress, p.total)
+	}
+	if p.progress == p.total {
+		fmt.Printf("\n")
+	}
+}


Test patch:
diff --git a/internal/progress_test.go b/internal/progress_test.go
index fc41f33..231d6d2 100644
--- a/internal/progress_test.go
+++ b/internal/progress_test.go
@@ -23,7 +23,7 @@ import (
 
 func TestProgress(t *testing.T) {
 	total := int64(4321)
-	p := NewProgress(total, "Progress", false)
+	p := NewProgress(total, "Progress", false, false)
 	assert.Equal(t, 0, p.pct)
 	time.Sleep(500 * time.Millisecond)
 	for _, v := range []int64{1000, 2000, 3000} {
@@ -37,12 +37,12 @@ func TestProgress(t *testing.T) {
 	p.MaybeReport(5000)
 	assert.Equal(t, 100, p.pct) // Never exceed 100%.
 	// Test corner case where total is 0.
-	p = NewProgress(0, "Progress", false)
+	p = NewProgress(0, "Progress", false, false)
 	assert.Equal(t, 100, p.pct)
 }
 
 func TestDone(t *testing.T) {
-	p := NewProgress(567, "Progress", false)
+	p := NewProgress(567, "Progress", false, false)
 	assert.Equal(t, 0, p.pct)
 	p.Done()
 	assert.Equal(t, 100, p.pct)