feat(spanner): Add support for MYSQLDUMP in import command: (GoogleCloudPlatform#1089)

1. This enables importing data and schema from mysql dump.
2. It updates the provided database with the schema in the Mysql Dump.
3. Unit test covering the entire flow of Import command is added.

Author: pratickchokhani

accessors/spanner/mocks.go

@@ -15,6 +15,7 @@ package spanneraccessor
 
 import (
 	"context"
+	spannerclient "github.com/GoogleCloudPlatform/spanner-migration-tool/accessors/clients/spanner/client"
 
 	"github.com/GoogleCloudPlatform/spanner-migration-tool/internal"
 )
@@ -38,6 +39,10 @@ type SpannerAccessorMock struct {
 	DropDatabaseMock                func(ctx context.Context, dbURI string) error
 	ValidateDMLMock                 func(ctx context.Context, query string) (bool, error)
 	TableExistsMock                 func(ctx context.Context, tableName string) (bool, error)
+	GetDatabaseNameMock             func() string
+	RefreshMock                     func(ctx context.Context, dbURI string)
+	SetSpannerClientMock            func(spannerClient spannerclient.SpannerClient)
+	GetSpannerClientMock            func() spannerclient.SpannerClient
 }
 
 func (sam *SpannerAccessorMock) GetDatabaseDialect(ctx context.Context, dbURI string) (string, error) {
@@ -99,3 +104,18 @@ func (sam *SpannerAccessorMock) ValidateDML(ctx context.Context, query string) (
 func (sam *SpannerAccessorMock) TableExists(ctx context.Context, tableName string) (bool, error) {
 	return sam.TableExistsMock(ctx, tableName)
 }
+func (sam *SpannerAccessorMock) GetDatabaseName() string {
+	return sam.GetDatabaseNameMock()
+}
+
+func (sam *SpannerAccessorMock) Refresh(ctx context.Context, dbURI string) {
+	sam.RefreshMock(ctx, dbURI)
+}
+
+func (sam *SpannerAccessorMock) SetSpannerClient(spannerClient spannerclient.SpannerClient) {
+	sam.SetSpannerClientMock(spannerClient)
+}
+
+func (sam *SpannerAccessorMock) GetSpannerClient() spannerclient.SpannerClient {
+	return sam.GetSpannerClientMock()
+}

accessors/spanner/spanner_accessor.go

@@ -80,6 +80,14 @@ type SpannerAccessor interface {
 	ValidateDML(ctx context.Context, query string) (bool, error)
 
 	TableExists(ctx context.Context, tableName string) (bool, error)
+
+	GetDatabaseName() string
+
+	Refresh(ctx context.Context, dbURI string)
+
+	SetSpannerClient(spannerClient spannerclient.SpannerClient)
+
+	GetSpannerClient() spannerclient.SpannerClient
 }
 
 // This implements the SpannerAccessor interface. This is the primary implementation that should be used in all places other than tests.
@@ -508,6 +516,18 @@ func (sp *SpannerAccessorImpl) ValidateDML(ctx context.Context, query string) (b
 	}
 }
 
+func (sp *SpannerAccessorImpl) GetDatabaseName() string {
+	return sp.SpannerClient.DatabaseName()
+}
+
 func (sp *SpannerAccessorImpl) Refresh(ctx context.Context, dbURI string) {
 	sp.SpannerClient.Refresh(ctx, dbURI)
 }
+
+func (sp *SpannerAccessorImpl) SetSpannerClient(spannerClient spannerclient.SpannerClient) {
+	sp.SpannerClient = spannerClient
+}
+
+func (sp *SpannerAccessorImpl) GetSpannerClient() spannerclient.SpannerClient {
+	return sp.SpannerClient
+}

cmd/import_data.go cmd/import.gocmd/import_data.go renamed to cmd/import.go

@@ -18,13 +18,13 @@ import (
 	"context"
 	"flag"
 	"fmt"
+	"github.com/GoogleCloudPlatform/spanner-migration-tool/import_file"
+	"github.com/GoogleCloudPlatform/spanner-migration-tool/sources/spanner"
 	"time"
 
 	spanneraccessor "github.com/GoogleCloudPlatform/spanner-migration-tool/accessors/spanner"
 	"github.com/GoogleCloudPlatform/spanner-migration-tool/common/constants"
-	"github.com/GoogleCloudPlatform/spanner-migration-tool/import_data"
 	"github.com/GoogleCloudPlatform/spanner-migration-tool/logger"
-	"github.com/GoogleCloudPlatform/spanner-migration-tool/sources/spanner"
 	"github.com/google/subcommands"
 	"go.uber.org/zap"
 )
@@ -46,7 +46,7 @@ func (cmd *ImportDataCmd) SetFlags(set *flag.FlagSet) {
 	set.StringVar(&cmd.databaseName, "database-name", "", "Spanner database name")
 	set.StringVar(&cmd.tableName, "table-name", "", "Spanner table name")
 	set.StringVar(&cmd.sourceUri, "source-uri", "", "URI of the file to import")
-	set.StringVar(&cmd.sourceFormat, "source-format", "", "Format of the file to import. Valid values {csv}")
+	set.StringVar(&cmd.sourceFormat, "source-format", "", "Format of the file to import. Valid values {csv, mysqldump}")
 	set.StringVar(&cmd.schemaUri, "schema-uri", "", "URI of the file with schema for the csv to import. Only used for csv format.")
 	set.StringVar(&cmd.csvLineDelimiter, "csv-line-delimiter", "", "Token to be used as line delimiter for csv format. Defaults to '\\n'. Only used for csv format.")
 	set.StringVar(&cmd.csvFieldDelimiter, "csv-field-delimiter", "", "Token to be used as field delimiter for csv format. Defaults to ','. Only used for csv format.")
@@ -57,60 +57,97 @@ func (cmd *ImportDataCmd) Execute(ctx context.Context, f *flag.FlagSet, args ...
 	logger.Log.Debug(fmt.Sprintf("instanceId %s, dbName %s, schemaUri %s\n", cmd.instanceId, cmd.databaseName, cmd.schemaUri))
 
 	dbURI := fmt.Sprintf("projects/%s/instances/%s/databases/%s", cmd.project, cmd.instanceId, cmd.databaseName)
-	infoSchema, err := spanner.NewInfoSchemaImplWithSpannerClient(ctx, dbURI, constants.DIALECT_GOOGLESQL)
-	if err != nil {
-		logger.Log.Error(fmt.Sprintf("Unable to read Spanner schema %v", err))
-		return subcommands.ExitFailure
-	}
 
 	switch cmd.sourceFormat {
 	case constants.CSV:
-		err := cmd.handleCsv(ctx, infoSchema)
+		//TODO: handle POSTGRESQL
+		dialect := constants.DIALECT_GOOGLESQL
+		err := cmd.handleCsv(ctx, dbURI, dialect)
 		if err != nil {
 			logger.Log.Error(fmt.Sprintf("Unable to handle Csv %v", err))
 			return subcommands.ExitFailure
 		}
 		return subcommands.ExitSuccess
+	case constants.MYSQLDUMP:
+		err := cmd.handleDatabaseDumpFile(ctx, dbURI, constants.MYSQLDUMP, constants.DIALECT_GOOGLESQL)
+		if err != nil {
+			logger.Log.Error(fmt.Sprintf("Unable to handle MYSQL Dump %v. Please reachout to the support team.", err))
+			return subcommands.ExitFailure
+		}
+		return subcommands.ExitSuccess
 	default:
 		logger.Log.Warn(fmt.Sprintf("format %s not supported yet", cmd.sourceFormat))
 	}
 	return subcommands.ExitFailure
 }
 
-func (cmd *ImportDataCmd) handleCsv(ctx context.Context, infoSchema *spanner.InfoSchemaImpl) error {
-	//TODO: handle POSTGRESQL
-	dialect := constants.DIALECT_GOOGLESQL
-
-	dbURI := fmt.Sprintf("projects/%s/instances/%s/databases/%s", cmd.project, cmd.instanceId, cmd.databaseName)
+func (cmd *ImportDataCmd) handleCsv(ctx context.Context, dbURI string, dialect string) error {
+	// TODO: move this to dependent classes
 	sp, err := spanneraccessor.NewSpannerAccessorClientImplWithSpannerClient(ctx, dbURI)
 	if err != nil {
-		logger.Log.Error(fmt.Sprintf("Unable to instantiate spanner client %v", err))
+		logger.Log.Error(fmt.Sprintf("Unable to read Spanner schema %v", err))
+		return fmt.Errorf("unable to read Spanner schema %v", err)
+	}
+	infoSchema, err := spanner.NewInfoSchemaImplWithSpannerClient(ctx, dbURI, dialect)
+	if err != nil {
+		logger.Log.Error(fmt.Sprintf("Unable to read Spanner schema %v", err))
 		return err
 	}
 
 	startTime := time.Now()
-	csvSchema := import_data.CsvSchemaImpl{ProjectId: cmd.project, InstanceId: cmd.instanceId,
+	csvSchema := import_file.CsvSchemaImpl{ProjectId: cmd.project, InstanceId: cmd.instanceId,
 		TableName: cmd.tableName, DbName: cmd.databaseName, SchemaUri: cmd.schemaUri, CsvFieldDelimiter: cmd.csvFieldDelimiter}
 	err = csvSchema.CreateSchema(ctx, dialect, sp)
 
 	endTime1 := time.Now()
 	elapsedTime := endTime1.Sub(startTime)
-	fmt.Println("Schema creation took ", elapsedTime.Seconds(), "  secs")
+	logger.Log.Info(fmt.Sprintf("Schema creation took %f secs", elapsedTime.Seconds()))
 	if err != nil {
 		return err
 	}
 
-	csvData := import_data.CsvDataImpl{ProjectId: cmd.project, InstanceId: cmd.instanceId,
+	csvData := import_file.CsvDataImpl{ProjectId: cmd.project, InstanceId: cmd.instanceId,
 		TableName: cmd.tableName, DbName: cmd.databaseName, SourceUri: cmd.sourceUri, CsvFieldDelimiter: cmd.csvFieldDelimiter}
 	err = csvData.ImportData(ctx, infoSchema, dialect)
 
 	endTime2 := time.Now()
 	elapsedTime = endTime2.Sub(endTime1)
-	fmt.Println("Data import took ", elapsedTime.Seconds(), "  secs")
+	logger.Log.Info(fmt.Sprintf("Data import took %f secs", elapsedTime.Seconds()))
 	return err
 
 }
 
+func (cmd *ImportDataCmd) handleDatabaseDumpFile(ctx context.Context, dbUri, sourceFormat string, dialect string) error {
+
+	importDump, err := import_file.NewImportFromDump(ctx, cmd.project, cmd.instanceId, cmd.databaseName, cmd.sourceUri, sourceFormat, dbUri)
+	if err != nil {
+		return fmt.Errorf("can't open dump file or create spanner client: %v", err)
+	}
+
+	defer importDump.Close()
+
+	schemaStartTime := time.Now()
+	conv, err := importDump.CreateSchema(ctx, dialect)
+	if err != nil {
+		return fmt.Errorf("can't create schema: %v", err)
+	}
+
+	schemaEndTime := time.Now()
+	elapsedTime := schemaEndTime.Sub(schemaStartTime)
+	logger.Log.Info(fmt.Sprintf("Schema creation took %f secs", elapsedTime.Seconds()))
+
+	err = importDump.ImportData(ctx, conv)
+
+	dataEndTime := time.Now()
+	elapsedTime = dataEndTime.Sub(schemaEndTime)
+	logger.Log.Info(fmt.Sprintf("Data import took %f secs", elapsedTime.Seconds()))
+	if err != nil {
+		return fmt.Errorf("can't import data: %v", err)
+	}
+	return nil
+
+}
+
 func init() {
 	logger.Log, _ = zap.NewDevelopment()
 }