Repository: cloudspannerecosystem/harbourbridge
Pull request #213 (2021-11-03 09:31:26)

Use source profile connection params for direct connect in subcommand…
Refactor the **driverConfig** variable to **sqlConnectionStr** which is a more suitable name and avoids clashes.

For schema, data and eval in subcommand mode, we can pass the sqlConnectionStr throughout as a parameter. 

If `sqlConnectionStr == ""`,  it means user is running HB in legacy mode. We follow the original flow then where the params are fetched from env variables.

In subcommand mode i.e. `sqlConnectionStr != ""`, the expectation is user will provide all connection params in the source-profile. If host/user/dbname are missing, we throw an error asking to specify those. If the user does not provide any flag (to support backward compatibility), we read the params from env variables. This is only temporary and will be removed once we fully switch to subcommand mode.

For DynamoDB, the code path is kept same as before because the SDK itself only supports reading from the env variables. We cannot specify credentials in the go client.


Patch:
diff --git a/README.md b/README.md
index 91a6070..ecbd275 100644
--- a/README.md
+++ b/README.md
@@ -315,6 +315,11 @@ Spanner DDL files.
 `-session` Specifies a session file that contains all schema and data 
 conversion state endcoded as JSON.
 
+`-dump-file` Specifies the path where the dump file is stored. 
+If the location is Google Cloud Storage (GCS), you can use the following format:
+`-dump-file=gs://{bucket_name}/{path/to/dump_file}`. Please ensure you have
+read pemissions to the GCS bucket you would like to use. 
+
 ## Example Usage
 
 Details on HarbourBridge example usage can be found here: 
diff --git a/cmd/cmd.go b/cmd/cmd.go
index 5fe8729..84bba51 100644
--- a/cmd/cmd.go
+++ b/cmd/cmd.go
@@ -41,7 +41,9 @@ func CommandLine(ctx context.Context, driver, targetDb, dbURI string, dataOnly,
 	var conv *internal.Conv
 	var err error
 	if !dataOnly {
-		conv, err = conversion.SchemaConv(driver, targetDb, ioHelper, schemaSampleSize)
+		// We pass an empty string to the sqlConnectionStr parameter as this is the legacy codepath,
+		// which reads the environment variables and constructs the string later on.
+		conv, err = conversion.SchemaConv(driver, "", targetDb, ioHelper, schemaSampleSize)
 		if err != nil {
 			return err
 		}
@@ -77,7 +79,9 @@ func CommandLine(ctx context.Context, driver, targetDb, dbURI string, dataOnly,
 		return fmt.Errorf("can't create client for db %s: %v", dbURI, err)
 	}
 
-	bw, err := conversion.DataConv(driver, ioHelper, client, conv, dataOnly)
+	// We pass an empty string to the sqlConnectionStr parameter as this is the legacy codepath,
+	// which reads the environment variables and constructs the string later on.
+	bw, err := conversion.DataConv(driver, "", ioHelper, client, conv, dataOnly)
 	if err != nil {
 		return fmt.Errorf("can't finish data conversion for db %s: %v", dbURI, err)
 	}
diff --git a/cmd/common.go b/cmd/common.go
index 827ea26..8e0d9f3 100644
--- a/cmd/common.go
+++ b/cmd/common.go
@@ -77,3 +77,35 @@ func getResourceIds(ctx context.Context, targetProfile TargetProfile, now time.T
 	}
 	return project, instance, dbName, err
 }
+
+func getSQLConnectionStr(sourceProfile SourceProfile) string {
+	sqlConnectionStr := ""
+	if sourceProfile.ty == SourceProfileTypeConnection {
+		switch sourceProfile.conn.ty {
+		case SourceProfileConnectionTypeMySQL:
+			connParams := sourceProfile.conn.mysql
+			return conversion.GetMYSQLConnectionStr(connParams.host, connParams.port, connParams.user, connParams.pwd, connParams.db)
+		case SourceProfileConnectionTypePostgreSQL:
+			connParams := sourceProfile.conn.pg
+			return conversion.GetPGSQLConnectionStr(connParams.host, connParams.port, connParams.user, connParams.pwd, connParams.db)
+		case SourceProfileConnectionTypeDynamoDB:
+			// For DynamoDB, client provided by aws-sdk reads connection credentials from env variables only.
+			// Thus, there is no need to create sqlConnectionStr for the same. We instead set the env variables
+			// programmatically if not set.
+			return ""
+		}
+	}
+	return sqlConnectionStr
+}
+
+func getSchemaSampleSize(sourceProfile SourceProfile) int64 {
+	schemaSampleSize := int64(100000)
+	if sourceProfile.ty == SourceProfileTypeConnection {
+		if sourceProfile.conn.ty == SourceProfileConnectionTypeDynamoDB {
+			if sourceProfile.conn.dydb.schemaSampleSize != 0 {
+				schemaSampleSize = sourceProfile.conn.dydb.schemaSampleSize
+			}
+		}
+	}
+	return schemaSampleSize
+}
diff --git a/cmd/data.go b/cmd/data.go
index 463e5c7..7fb1d7b 100644
--- a/cmd/data.go
+++ b/cmd/data.go
@@ -129,7 +129,7 @@ func (cmd *DataCmd) Execute(ctx context.Context, f *flag.FlagSet, _ ...interface
 		return subcommands.ExitFailure
 	}
 
-	bw, err := conversion.DataConv(driverName, &ioHelper, client, conv, true)
+	bw, err := conversion.DataConv(driverName, getSQLConnectionStr(sourceProfile), &ioHelper, client, conv, true)
 	if err != nil {
 		err = fmt.Errorf("can't finish data conversion for db %s: %v", dbURI, err)
 		return subcommands.ExitFailure
diff --git a/cmd/eval.go b/cmd/eval.go
index 05d6f36..95726ce 100644
--- a/cmd/eval.go
+++ b/cmd/eval.go
@@ -97,16 +97,10 @@ func (cmd *EvalCmd) Execute(ctx context.Context, f *flag.FlagSet, _ ...interface
 		cmd.filePrefix = dbName + "."
 	}
 
-	schemaSampleSize := int64(100000)
-	if sourceProfile.ty == SourceProfileTypeConnection {
-		if sourceProfile.conn.ty == SourceProfileConnectionTypeDynamoDB {
-			if sourceProfile.conn.dydb.schemaSampleSize != 0 {
-				schemaSampleSize = sourceProfile.conn.dydb.schemaSampleSize
-			}
-		}
-	}
+	sqlConnectionStr := getSQLConnectionStr(sourceProfile)
+
 	var conv *internal.Conv
-	conv, err = conversion.SchemaConv(driverName, targetDb, &ioHelper, schemaSampleSize)
+	conv, err = conversion.SchemaConv(driverName, sqlConnectionStr, targetDb, &ioHelper, getSchemaSampleSize(sourceProfile))
 	if err != nil {
 		panic(err)
 	}
@@ -140,7 +134,7 @@ func (cmd *EvalCmd) Execute(ctx context.Context, f *flag.FlagSet, _ ...interface
 		return subcommands.ExitFailure
 	}
 
-	bw, err := conversion.DataConv(driverName, &ioHelper, client, conv, true)
+	bw, err := conversion.DataConv(driverName, sqlConnectionStr, &ioHelper, client, conv, true)
 	if err != nil {
 		err = fmt.Errorf("can't finish data conversion for db %s: %v", dbURI, err)
 		return subcommands.ExitFailure
diff --git a/cmd/schema.go b/cmd/schema.go
index 3da365f..219a9a2 100644
--- a/cmd/schema.go
+++ b/cmd/schema.go
@@ -94,16 +94,8 @@ func (cmd *SchemaCmd) Execute(ctx context.Context, f *flag.FlagSet, _ ...interfa
 		cmd.filePrefix = dbName + "."
 	}
 
-	schemaSampleSize := int64(100000)
-	if sourceProfile.ty == SourceProfileTypeConnection {
-		if sourceProfile.conn.ty == SourceProfileConnectionTypeDynamoDB {
-			if sourceProfile.conn.dydb.schemaSampleSize != 0 {
-				schemaSampleSize = sourceProfile.conn.dydb.schemaSampleSize
-			}
-		}
-	}
 	var conv *internal.Conv
-	conv, err = conversion.SchemaConv(driverName, targetDb, &ioHelper, schemaSampleSize)
+	conv, err = conversion.SchemaConv(driverName, getSQLConnectionStr(sourceProfile), targetDb, &ioHelper, getSchemaSampleSize(sourceProfile))
 	if err != nil {
 		return subcommands.ExitFailure
 	}
diff --git a/cmd/source_profile.go b/cmd/source_profile.go
index 01bac85..92312fa 100644
--- a/cmd/source_profile.go
+++ b/cmd/source_profile.go
@@ -7,6 +7,7 @@ import (
 	"strings"
 
 	"github.com/cloudspannerecosystem/harbourbridge/common/constants"
+	"github.com/cloudspannerecosystem/harbourbridge/conversion"
 )
 
 type SourceProfileType int
@@ -55,27 +56,55 @@ type SourceProfileConnectionMySQL struct {
 	pwd  string // Same as MYSQLPWD environment variable
 }
 
-func NewSourceProfileConnectionMySQL(params map[string]string) SourceProfileConnectionMySQL {
-	// TODO: Move parsing of environment variables in this function.
+func NewSourceProfileConnectionMySQL(params map[string]string) (SourceProfileConnectionMySQL, error) {
 	mysql := SourceProfileConnectionMySQL{}
-	if host, ok := params["host"]; ok {
-		mysql.host = host
-	}
-	if port, ok := params["port"]; ok {
-		mysql.port = port
-	} else { // Set default port for mysql, which rarely changes.
-		mysql.port = "3306"
+	host, hostOk := params["host"]
+	user, userOk := params["user"]
+	db, dbOk := params["db_name"]
+	port, portOk := params["port"]
+	pwd, pwdOk := params["password"]
+	// We don't users to mix and match params from source-profile and environment variables.
+	// We either try to get all params from the source-profile and if none are set, we read from the env variables.
+	if !(hostOk || userOk || dbOk || portOk || pwdOk) {
+		// No connection params provided through source-profile. Fetching from env variables.
+		fmt.Printf("Connection parameters not specified in source-profile. Reading from " +
+			"environment variables MYSQLHOST, MYSQLUSER, MYSQLDATABASE, MYSQLPORT, MYSQLPWD...\n")
+		mysql.host = os.Getenv("MYSQLHOST")
+		mysql.user = os.Getenv("MYSQLUSER")
+		mysql.db = os.Getenv("MYSQLDATABASE")
+		mysql.port = os.Getenv("MYSQLPORT")
+		mysql.pwd = os.Getenv("MYSQLPWD")
+		// Throw error if the input entered is empty.
+		if mysql.host == "" || mysql.user == "" || mysql.db == "" {
+			return mysql, fmt.Errorf("found empty string for MYSQLHOST/MYSQLUSER/MYSQLDATABASE. Please specify these environment variables with correct values")
+		}
+	} else if hostOk && userOk && dbOk {
+		// If atleast host, username and dbname are provided through source-profile,
+		// go ahead and use source-profile. Port and password handled later even if they are empty.
+		mysql.host, mysql.user, mysql.db, mysql.port, mysql.pwd = host, user, db, port, pwd
+		// Throw error if the input entered is empty.
+		if mysql.host == "" || mysql.user == "" || mysql.db == "" {
+			return mysql, fmt.Errorf("found empty string for host/user/db_name. Please specify host, port, user and db_name in the source-profile")
+		}
+	} else {
+		// Partial params provided through source-profile. Ask user to provide all through the source-profile.
+		return mysql, fmt.Errorf("please specify host, port, user and db_name in the source-profile")
 	}
-	if user, ok := params["user"]; ok {
-		mysql.user = user
+
+	// Throw same error if the input entered is empty.
+	if mysql.host == "" || mysql.user == "" || mysql.db == "" {
+		return mysql, fmt.Errorf("found empty string for host/user/db. please specify host, port, user and db_name in the source-profile")
 	}
-	if db, ok := params["db_name"]; ok {
-		mysql.db = db
+
+	if mysql.port == "" {
+		// Set default port for mysql, which rarely changes.
+		mysql.port = "3306"
 	}
-	if pwd, ok := params["password"]; ok {
-		mysql.pwd = pwd
+	if mysql.pwd == "" {
+		mysql.pwd = conversion.GetPassword()
 	}
-	return mysql
+
+	return mysql, nil
 }
 
 type SourceProfileConnectionPostgreSQL struct {
@@ -86,29 +115,55 @@ type SourceProfileConnectionPostgreSQL struct {
 	pwd  string // Same as PGPASSWORD environment variable
 }
 
-func NewSourceProfileConnectionPostgreSQL(params map[string]string) SourceProfileConnectionPostgreSQL {
+func NewSourceProfileConnectionPostgreSQL(params map[string]string) (SourceProfileConnectionPostgreSQL, error) {
 	pg := SourceProfileConnectionPostgreSQL{}
-	if host, ok := params["host"]; ok {
-		pg.host = host
+	host, hostOk := params["host"]
+	user, userOk := params["user"]
+	db, dbOk := params["db_name"]
+	port, portOk := params["port"]
+	pwd, pwdOk := params["password"]
+	// We don't users to mix and match params from source-profile and environment variables.
+	// We either try to get all params from the source-profile and if none are set, we read from the env variables.
+	if !(hostOk || userOk || dbOk || portOk || pwdOk) {
+		// No connection params provided through source-profile. Fetching from env variables.
+		fmt.Printf("Connection parameters not specified in source-profile. Reading from " +
+			"environment variables PGHOST, PGUSER, PGDATABASE, PGPORT, PGPASSWORD...\n")
+		pg.host = os.Getenv("PGHOST")
+		pg.user = os.Getenv("PGUSER")
+		pg.db = os.Getenv("PGDATABASE")
+		pg.port = os.Getenv("PGPORT")
+		pg.pwd = os.Getenv("PGPASSWORD")
+		// Throw error if the input entered is empty.
+		if pg.host == "" || pg.user == "" || pg.db == "" {
+			return pg, fmt.Errorf("found empty string for PGHOST/PGUSER/PGDATABASE. Please specify these environment variables with correct values")
+		}
+	} else if hostOk && userOk && dbOk {
+		// All connection params provided through source-profile. Port and password handled later.
+		pg.host, pg.user, pg.db, pg.port, pg.pwd = host, user, db, port, pwd
+		// Throw error if the input entered is empty.
+		if pg.host == "" || pg.user == "" || pg.db == "" {
+			return pg, fmt.Errorf("found empty string for host/user/db_name. Please specify host, port, user and db_name in the source-profile")
+		}
+	} else {
+		// Partial params provided through source-profile. Ask user to provide all through the source-profile.
+		return pg, fmt.Errorf("please specify host, port, user and db_name in the source-profile")
 	}
-	if port, ok := params["port"]; ok {
-		pg.port = port
-	} else { // Set default port for postgresql, which rarely changes.
+
+	if pg.port == "" {
+		// Set default port for postgresql, which rarely changes.
 		pg.port = "5432"
 	}
-	if user, ok := params["user"]; ok {
-		pg.user = user
-	}
-	if db, ok := params["db_name"]; ok {
-		pg.db = db
-	}
-	if pwd, ok := params["password"]; ok {
-		pg.pwd = pwd
+	if pg.pwd == "" {
+		pg.pwd = conversion.GetPassword()
 	}
-	return pg
+
+	return pg, nil
 }
 
 type SourceProfileConnectionDynamoDB struct {
+	// These connection params are not used currently because the SDK reads directly from the env variables.
+	// These are still kept around as reference when we refactor passing
+	// SourceProfile instead of sqlConnectionStr around.
 	awsAccessKeyID     string // Same as AWS_ACCESS_KEY_ID environment variable
 	awsSecretAccessKey string // Same as AWS_SECRET_ACCESS_KEY environment variable
 	awsRegion          string // Same as AWS_REGION environment variable
@@ -117,20 +172,7 @@ type SourceProfileConnectionDynamoDB struct {
 }
 
 func NewSourceProfileConnectionDynamoDB(params map[string]string) (SourceProfileConnectionDynamoDB, error) {
-	// TODO: Populate environment variables for DynamoDB if passed as flags.
 	dydb := SourceProfileConnectionDynamoDB{}
-	if awsAccessKeyID, ok := params["awsAccessKeyID"]; ok {
-		dydb.awsAccessKeyID = awsAccessKeyID
-	}
-	if awsSecretAccessKey, ok := params["awsSecretAccessKey"]; ok {
-		dydb.awsSecretAccessKey = awsSecretAccessKey
-	}
-	if awsRegion, ok := params["awsRegion"]; ok {
-		dydb.awsRegion = awsRegion
-	}
-	if dydbEndpoint, ok := params["dydbEndpoint"]; ok {
-		dydb.dydbEndpoint = dydbEndpoint
-	}
 	if schemaSampleSize, ok := params["schemaSampleSize"]; ok {
 		schemaSampleSizeInt, err := strconv.Atoi(schemaSampleSize)
 		if err != nil {
@@ -138,6 +180,22 @@ func NewSourceProfileConnectionDynamoDB(params map[string]string) (SourceProfile
 		}
 		dydb.schemaSampleSize = int64(schemaSampleSizeInt)
 	}
+	// For DynamoDB, the preferred way to provide connection params is through env variables.
+	// Unlike postgres and mysql, there may not be deprecation of env variables, hence it
+	// is better to override env variables optionally via source profile params.
+	var ok bool
+	if dydb.awsAccessKeyID, ok = params["awsAccessKeyID"]; ok {
+		os.Setenv("AWS_ACCESS_KEY_ID", dydb.awsAccessKeyID)
+	}
+	if dydb.awsSecretAccessKey, ok = params["awsSecretAccessKey"]; ok {
+		os.Setenv("AWS_SECRET_ACCESS_KEY", dydb.awsAccessKeyID)
+	}
+	if dydb.awsRegion, ok = params["awsRegion"]; ok {
+		os.Setenv("AWS_REGION", dydb.awsAccessKeyID)
+	}
+	if dydb.dydbEndpoint, ok = params["dydbEndpoint"]; ok {
+		os.Setenv("DYNAMODB_ENDPOINT_OVERRIDE", dydb.awsAccessKeyID)
+	}
 	return dydb, nil
 }
 
@@ -150,22 +208,28 @@ type SourceProfileConnection struct {
 
 func NewSourceProfileConnection(source string, params map[string]string) (SourceProfileConnection, error) {
 	conn := SourceProfileConnection{}
+	var err error
 	switch strings.ToLower(source) {
 	case "mysql":
 		{
 			conn.ty = SourceProfileConnectionTypeMySQL
-			conn.mysql = NewSourceProfileConnectionMySQL(params)
+			conn.mysql, err = NewSourceProfileConnectionMySQL(params)
+			if err != nil {
+				return conn, err
+			}
 		}
 	case "postgresql", "postgres", "pg":
 		{
 			conn.ty = SourceProfileConnectionTypePostgreSQL
-			conn.pg = NewSourceProfileConnectionPostgreSQL(params)
+			conn.pg, err = NewSourceProfileConnectionPostgreSQL(params)
+			if err != nil {
+				return conn, err
+			}
 		}
 	case "dynamodb":
 		{
 			conn.ty = SourceProfileConnectionTypeDynamoDB
-			dydb, err := NewSourceProfileConnectionDynamoDB(params)
-			conn.dydb = dydb
+			conn.dydb, err = NewSourceProfileConnectionDynamoDB(params)
 			if err != nil {
 				return conn, err
 			}
@@ -191,7 +255,7 @@ type SourceProfile struct {
 	config SourceProfileConfig
 }
 
-// ToLegacyDriver converts source profile to equivalent legacy global flags
+// ToLegacyDriver converts source-profile to equivalent legacy global flags
 // e.g., -driver, -dump-file etc since the rest of the codebase still uses the
 // same. TODO: Deprecate this function and pass around SourceProfile across the
 // codebase wherever information about source connection is required.
@@ -230,8 +294,8 @@ func (src SourceProfile) ToLegacyDriver(source string) (string, error) {
 	}
 }
 
-// Source profile is passed as a list of key value pairs on the command line.
-// Following 3 formats are supported as valid source profiles.
+// Flag source-profile is passed as a list of key value pairs on the command line.
+// Following 3 formats are supported as a valid source-profile.
 //
 // Format 1. Specify file path and file format.
 // File path can be a local file path or a gcs file path. Support for more file
@@ -253,7 +317,7 @@ func NewSourceProfile(s string, source string) (SourceProfile, error) {
 	}
 	params, err := parseProfile(s)
 	if err != nil {
-		return SourceProfile{}, fmt.Errorf("could not parse source profile, error = %v", err)
+		return SourceProfile{}, fmt.Errorf("could not parse source-profile, error = %v", err)
 	}
 
 	if _, ok := params["file"]; ok || filePipedToStdin() {
diff --git a/cmd/target_profile.go b/cmd/target_profile.go
index 961b45e..c0c2645 100644
--- a/cmd/target_profile.go
+++ b/cmd/target_profile.go
@@ -39,7 +39,7 @@ type TargetProfile struct {
 	conn TargetProfileConnection
 }
 
-// ToLegacyTargetDb converts source profile to equivalent legacy global flag
+// ToLegacyTargetDb converts source-profile to equivalent legacy global flag
 // -target-db etc since the rest of the codebase still uses the same.
 // TODO: Deprecate this function and pass around TargetProfile across the
 // codebase wherever information about target connection is required.
diff --git a/conversion/conversion.go b/conversion/conversion.go
index 330c0d3..d92008c 100644
--- a/conversion/conversion.go
+++ b/conversion/conversion.go
@@ -33,6 +33,7 @@ import (
 	"io"
 	"io/ioutil"
 	"log"
+	"net/url"
 	"os"
 	"os/exec"
 	"strings"
@@ -55,6 +56,8 @@ import (
 	adminpb "google.golang.org/genproto/googleapis/spanner/admin/database/v1"
 	instancepb "google.golang.org/genproto/googleapis/spanner/admin/instance/v1"
 
+	"cloud.google.com/go/storage"
+
 	"github.com/cloudspannerecosystem/harbourbridge/common/constants"
 	"github.com/cloudspannerecosystem/harbourbridge/internal"
 	"github.com/cloudspannerecosystem/harbourbridge/sources/common"
@@ -73,10 +76,15 @@ var (
 	MaxWorkers = 20
 )
 
-func SchemaConv(driver string, targetDb string, ioHelper *IOStreams, schemaSampleSize int64) (*internal.Conv, error) {
+// The sqlConnectionStr param provides the connection details to use the go SQL library.
+// It is empty in the following cases:
+//  - Driver is DynamoDB or a dump file mode.
+//  - This function is called as part of the legacy global CLI flag mode. (This string is constructed from env variables later on)
+// When using source-profile, the sqlConnectionStr is constructed from the input params.
+func SchemaConv(driver, sqlConnectionStr, targetDb string, ioHelper *IOStreams, schemaSampleSize int64) (*internal.Conv, error) {
 	switch driver {
 	case constants.POSTGRES, constants.MYSQL:
-		return schemaFromSQL(driver, targetDb)
+		return schemaFromSQL(driver, sqlConnectionStr, targetDb)
 	case constants.PGDUMP, constants.MYSQLDUMP:
 		return schemaFromDump(driver, targetDb, ioHelper)
 	case constants.DYNAMODB:
@@ -86,7 +94,12 @@ func SchemaConv(driver string, targetDb string, ioHelper *IOStreams, schemaSampl
 	}
 }
 
-func DataConv(driver string, ioHelper *IOStreams, client *sp.Client, conv *internal.Conv, dataOnly bool) (*spanner.BatchWriter, error) {
+// The sqlConnectionStr param provides the connection details to use the go SQL library.
+// It is empty in the following cases:
+//  - Driver is DynamoDB or a dump file mode.
+//  - This function is called as part of the legacy global CLI flag mode. (This string is constructed from env variables later on)
+// When using source-profile, the sqlConnectionStr is constructed from the input params.
+func DataConv(driver, sqlConnectionStr string, ioHelper *IOStreams, client *sp.Client, conv *internal.Conv, dataOnly bool) (*spanner.BatchWriter, error) {
 	config := spanner.BatchWriterConfig{
 		BytesLimit: 100 * 1000 * 1000,
 		WriteLimit: 40,
@@ -95,7 +108,7 @@ func DataConv(driver string, ioHelper *IOStreams, client *sp.Client, conv *inter
 	}
 	switch driver {
 	case constants.POSTGRES, constants.MYSQL:
-		return dataFromSQL(driver, config, client, conv)
+		return dataFromSQL(driver, sqlConnectionStr, config, client, conv)
 	case constants.PGDUMP, constants.MYSQLDUMP:
 		if conv.SpSchema.CheckInterleaved() {
 			return nil, fmt.Errorf("HarbourBridge does not currently support data conversion from dump files\nif the schema contains interleaved tables. Suggest using direct access to source database\ni.e. using drivers postgres and mysql.")
@@ -108,60 +121,88 @@ func DataConv(driver string, ioHelper *IOStreams, client *sp.Client, conv *inter
 	}
 }
 
-func driverConfig(driver string) (string, error) {
+func generateSQLConnectionStr(driver string) (string, error) {
 	switch driver {
 	case constants.POSTGRES:
-		return pgDriverConfig()
+		return generatePGSQLConnectionStr()
 	case constants.MYSQL:
-		return mysqlDriverConfig()
+		return generateMYSQLConnectionStr()
 	default:
-		return "", fmt.Errorf("Driver %s not supported", driver)
+		return "", fmt.Errorf("driver %s not supported", driver)
 	}
 }
 
-func pgDriverConfig() (string, error) {
+func generatePGSQLConnectionStr() (string, error) {
 	server := os.Getenv("PGHOST")
 	port := os.Getenv("PGPORT")
 	user := os.Getenv("PGUSER")
 	dbname := os.Getenv("PGDATABASE")
 	if server == "" || port == "" || user == "" || dbname == "" {
 		fmt.Printf("Please specify host, port, user and database using PGHOST, PGPORT, PGUSER and PGDATABASE environment variables\n")
-		return "", fmt.Errorf("Could not connect to source database")
+		return "", fmt.Errorf("could not connect to source database")
 	}
 	password := os.Getenv("PGPASSWORD")
 	if password == "" {
-		password = getPassword()
+		password = GetPassword()
 	}
-	return fmt.Sprintf("host=%s port=%s user=%s password=%s dbname=%s sslmode=disable", server, port, user, password, dbname), nil
+	return GetPGSQLConnectionStr(server, port, user, password, dbname), nil
+}
+
+func GetPGSQLConnectionStr(server, port, user, password, dbname string) string {
+	return fmt.Sprintf("host=%s port=%s user=%s password=%s dbname=%s sslmode=disable", server, port, user, password, dbname)
 }
 
-func mysqlDriverConfig() (string, error) {
+func generateMYSQLConnectionStr() (string, error) {
 	server := os.Getenv("MYSQLHOST")
 	port := os.Getenv("MYSQLPORT")
 	user := os.Getenv("MYSQLUSER")
 	dbname := os.Getenv("MYSQLDATABASE")
 	if server == "" || port == "" || user == "" || dbname == "" {
 		fmt.Printf("Please specify host, port, user and database using MYSQLHOST, MYSQLPORT, MYSQLUSER and MYSQLDATABASE environment variables\n")
-		return "", fmt.Errorf("Could not connect to source database")
+		return "", fmt.Errorf("could not connect to source database")
 	}
 	password := os.Getenv("MYSQLPWD")
 	if password == "" {
-		password = getPassword()
+		password = GetPassword()
 	}
-	return fmt.Sprintf("%s:%s@tcp(%s:%s)/%s", user, password, server, port, dbname), nil
+	return GetMYSQLConnectionStr(server, port, user, password, dbname), nil
 }
 
-func schemaFromSQL(driver string, targetDb string) (*internal.Conv, error) {
-	driverConfig, err := driverConfig(driver)
-	if err != nil {
-		return nil, err
+func GetMYSQLConnectionStr(server, port, user, password, dbname string) string {
+	return fmt.Sprintf("%s:%s@tcp(%s:%s)/%s", user, password, server, port, dbname)
+}
+
+func getDbNameFromSQLConnectionStr(driver, sqlConnectionStr string) string {
+	switch driver {
+	case constants.POSTGRES:
+		dbParam := strings.Split(sqlConnectionStr, " ")[4]
+		return strings.Split(dbParam, "=")[1]
+	case constants.MYSQL:
+		return strings.Split(sqlConnectionStr, ")/")[1]
+	}
+	return ""
+}
+
+// TODO: Pass around cmd.SourceProfile instead of sqlConnectionStr.
+// Doing that requires refactoring since that would introduce a circular dependency between
+// conversion.go and cmd/source_profile.go.
+func schemaFromSQL(driver, sqlConnectionStr, targetDb string) (*internal.Conv, error) {
+	// If empty, this is called as part of the legacy mode witih global CLI flags.
+	// When using source-profile mode is used, the sqlConnectionStr is already populated.
+	if sqlConnectionStr == "" {
+		var err error
+		sqlConnectionStr, err = generateSQLConnectionStr(driver)
+		if err != nil {
+			return nil, err
+		}
 	}
-	sourceDB, err := sql.Open(driver, driverConfig)
+	sourceDB, err := sql.Open(driver, sqlConnectionStr)
 	if err != nil {
 		return nil, err
 	}
 	conv := internal.MakeConv()
 	conv.TargetDb = targetDb
+	conv.SrcDbName = getDbNameFromSQLConnectionStr(driver, sqlConnectionStr)
 	err = ProcessInfoSchema(driver, conv, sourceDB)
 	if err != nil {
 		return nil, err
@@ -169,17 +210,21 @@ func schemaFromSQL(driver string, targetDb string) (*internal.Conv, error) {
 	return conv, nil
 }
 
-func dataFromSQL(driver string, config spanner.BatchWriterConfig, client *sp.Client, conv *internal.Conv) (*spanner.BatchWriter, error) {
-	// TODO: Refactor to avoid redundant calls to driverConfig and
+func dataFromSQL(driver, sqlConnectionStr string, config spanner.BatchWriterConfig, client *sp.Client, conv *internal.Conv) (*spanner.BatchWriter, error) {
+	// TODO: Refactor to avoid redundant calls to sqlConnectionStr and
 	// Open in schemaFromSQL and dataFromSQL. Also refactor to
 	// share code with dataFromPgDump. Use single transaction for
 	// reading schema and data from source db to get consistent
 	// dump.
-	driverConfig, err := driverConfig(driver)
-	if err != nil {
-		return nil, err
+	if sqlConnectionStr == "" {
+		// If empty, this is called as part of the old command line workflow.
+		var err error
+		sqlConnectionStr, err = generateSQLConnectionStr(driver)
+		if err != nil {
+			return nil, err
+		}
 	}
-	sourceDB, err := sql.Open(driver, driverConfig)
+	sourceDB, err := sql.Open(driver, sqlConnectionStr)
 	if err != nil {
 		return nil, err
 	}
@@ -270,14 +315,81 @@ type IOStreams struct {
 	BytesRead           int64
 }
 
+// downloadFromGCS returns the dump file that is downloaded from GCS
+func downloadFromGCS(bucketName string, filePath string) (*os.File, error) {
+	ctx := context.Background()
+
+	client, err := storage.NewClient(ctx)
+	if err != nil {
+		fmt.Printf("Failed to create GCS client for bucket %q", bucketName)
+		log.Fatal(err)
+	}
+	defer client.Close()
+
+	bucket := client.Bucket(bucketName)
+	rc, err := bucket.Object(filePath).NewReader(ctx)
+	if err != nil {
+		fmt.Printf("readFile: unable to open file from bucket %q, file %q: %v", bucketName, filePath, err)
+		log.Fatal(err)
+		return nil, err
+	}
+	defer rc.Close()
+	r := bufio.NewReader(rc)
+
+	tmpfile, err := ioutil.TempFile("", "harbourbridge.gcs.data")
+	if err != nil {
+		fmt.Printf("saveFile: unable to open temporary file to save dump file from GCS bucket %v", err)
+		log.Fatal(err)
+		return nil, err
+	}
+	syscall.Unlink(tmpfile.Name()) // File will be deleted when this process exits.
+
+	fmt.Printf("\nDownloading dump file from GCS bucket %s, path %s\n", bucketName, filePath)
+	buffer := make([]byte, 1024)
+	for {
+		// read a chunk
+		n, err := r.Read(buffer[:cap(buffer)])
+
+		if err != nil && err != io.EOF {
+			fmt.Printf("readFile: unable to read entire dump file from bucket %s, file %s: %v", bucketName, filePath, err)
+			log.Fatal(err)
+			return nil, err
+		}
+		if n == 0 && err == io.EOF {
+			break
+		}
+
+		// write a chunk
+		if _, err = tmpfile.Write(buffer[:n]); err != nil {
+			fmt.Printf("saveFile: unable to save read data from bucket %s, file %s: %v", bucketName, filePath, err)
+			log.Fatal(err)
+		}
+	}
+
+	return tmpfile, nil
+}
+
 // NewIOStreams returns a new IOStreams struct such that input stream is set
 // to open file descriptor for dumpFile if driver is PGDUMP or MYSQLDUMP.
 // Input stream defaults to stdin. Output stream is always set to stdout.
 func NewIOStreams(driver string, dumpFile string) IOStreams {
 	io := IOStreams{In: os.Stdin, Out: os.Stdout}
+	u, err := url.Parse(dumpFile)
+	if err != nil {
+		fmt.Printf("parseFilePath: unable parse file path for dumpfile %s", dumpFile)
+		log.Fatal(err)
+	}
 	if (driver == constants.PGDUMP || driver == constants.MYSQLDUMP) && dumpFile != "" {
 		fmt.Printf("\nLoading dump file from path: %s\n", dumpFile)
-		f, err := os.Open(dumpFile)
+		var f *os.File
+		var err error
+		if u.Scheme == "gs" {
+			bucketName := u.Host
+			filePath := u.Path[1:] // removes "/" from beginning of path
+			f, err = downloadFromGCS(bucketName, filePath)
+		} else {
+			f, err = os.Open(dumpFile)
+		}
 		if err != nil {
 			fmt.Printf("\nError reading dump file: %v err:%v\n", dumpFile, err)
 			log.Fatal(err)
@@ -497,7 +609,7 @@ func CreateDatabase(ctx context.Context, adminClient *database.DatabaseAdminClie
 		// a) backticks around the database name, and
 		// b) DDL statements as part of a CreateDatabase operation (so schema
 		// must be set using a separate UpdateDatabase operation).
-		req.CreateStatement = "CREATE DATABASE " + dbName
+		req.CreateStatement = "CREATE DATABASE \"" + dbName + "\""
 		req.DatabaseDialect = adminpb.DatabaseDialect_POSTGRESQL
 	} else {
 		req.CreateStatement = "CREATE DATABASE `" + dbName + "`"
@@ -876,7 +988,7 @@ func GetDatabaseName(driver string, now time.Time) (string, error) {
 	return generateName(fmt.Sprintf("%s_%s", driver, now.Format("2006-01-02")))
 }
 
-func getPassword() string {
+func GetPassword() string {
 	fmt.Print("Enter Password: ")
 	bytePassword, err := terminal.ReadPassword(int(syscall.Stdin))
 	if err != nil {
@@ -1050,7 +1162,7 @@ func ProcessDump(driver string, conv *internal.Conv, r *internal.Reader) error {
 func ProcessInfoSchema(driver string, conv *internal.Conv, db *sql.DB) error {
 	switch driver {
 	case constants.MYSQL:
-		return common.ProcessInfoSchema(conv, db, mysql.InfoSchemaImpl{os.Getenv("MYSQLDATABASE")})
+		return common.ProcessInfoSchema(conv, db, mysql.InfoSchemaImpl{conv.SrcDbName})
 	case constants.POSTGRES:
 		return common.ProcessInfoSchema(conv, db, postgres.InfoSchemaImpl{})
 	default:
@@ -1062,11 +1174,11 @@ func ProcessInfoSchema(driver string, conv *internal.Conv, db *sql.DB) error {
 func SetRowStats(driver string, conv *internal.Conv, db *sql.DB) error {
 	switch driver {
 	case constants.MYSQL:
-		common.SetRowStats(conv, db, mysql.InfoSchemaImpl{os.Getenv("MYSQLDATABASE")})
+		common.SetRowStats(conv, db, mysql.InfoSchemaImpl{conv.SrcDbName})
 	case constants.POSTGRES:
 		common.SetRowStats(conv, db, postgres.InfoSchemaImpl{})
 	default:
-		return fmt.Errorf("Could not set rows stats for '%s' driver", driver)
+		return fmt.Errorf("could not set rows stats for '%s' driver", driver)
 	}
 	return nil
 }
@@ -1076,11 +1188,11 @@ func ProcessSQLData(driver string, conv *internal.Conv, db *sql.DB) error {
 	switch driver {
 	//TODO - move this logic into a factory within the sources dir
 	case constants.MYSQL:
-		common.ProcessSQLData(conv, db, mysql.InfoSchemaImpl{os.Getenv("MYSQLDATABASE")})
+		common.ProcessSQLData(conv, db, mysql.InfoSchemaImpl{conv.SrcDbName})
 	case constants.POSTGRES:
 		common.ProcessSQLData(conv, db, postgres.InfoSchemaImpl{})
 	default:
-		return fmt.Errorf("Data conversion for driver %s is not supported", driver)
+		return fmt.Errorf("data conversion for driver %s is not supported", driver)
 	}
 	return nil
 }
diff --git a/go.mod b/go.mod
index 41b24cf..68e5515 100644
--- a/go.mod
+++ b/go.mod
@@ -3,8 +3,9 @@ module github.com/cloudspannerecosystem/harbourbridge
 go 1.13
 
 require (
-	cloud.google.com/go v0.93.3
+	cloud.google.com/go v0.94.1
 	cloud.google.com/go/spanner v1.10.0
+	cloud.google.com/go/storage v1.17.0
 	github.com/DATA-DOG/go-sqlmock v1.4.1
 	github.com/aws/aws-sdk-go v1.34.5
 	github.com/go-sql-driver/mysql v1.5.0
@@ -21,8 +22,8 @@ require (
 	github.com/stretchr/testify v1.6.1
 	golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9
 	golang.org/x/tools v0.1.7 // indirect
-	google.golang.org/api v0.54.0
-	google.golang.org/genproto v0.0.0-20210827211047-25e5f791fe06
+	google.golang.org/api v0.57.0
+	google.golang.org/genproto v0.0.0-20210921142501-181ce0d877f6
 )
 
 // cloud.google.com/go will upgrade grpc to v1.40.0
diff --git a/go.sum b/go.sum
index cb6b253..d6644a2 100644
--- a/go.sum
+++ b/go.sum
@@ -12,6 +12,8 @@ cloud.google.com/go/storage v1.5.0/go.mod h1:tpKbwo567HUNpVclU5sGELwQWBDZ8gh0Zeo
 cloud.google.com/go/storage v1.6.0/go.mod h1:N7U0C8pVQ/+NIKOBQyamJIeKQKkZ+mxpohlUTyfDhBk=
 cloud.google.com/go/storage v1.8.0/go.mod h1:Wv1Oy7z6Yz3DshWRJFhqM/UCfaWIRTdp0RXyy7KQOVs=
 cloud.google.com/go/storage v1.10.0/go.mod h1:FLPqc6j+Ki4BU591ie1oL6qBQGu2Bl/tZ9ullr3+Kg0=
+cloud.google.com/go/storage v1.17.0 h1:CDpe3jS3EiD5nGlbtvyA4EUfkF6k9GMrxLR8+hLmoec=
+cloud.google.com/go/storage v1.17.0/go.mod h1:0wRtHSM3Npk/QJYdwcpRNVRVJlH2OxyWF9Dws3J+MtE=
 dmitri.shuralyov.com/gpu/mtl v0.0.0-20190408044501-666a987793e9/go.mod h1:H6x//7gZCb22OMCxBHrMx7a5I7Hp++hsVxbQ4BYO7hU=
 github.com/BurntSushi/toml v0.3.1 h1:WXkYYl6Yr3qBf1K79EBnL4mak0OimBfB0XUf9Vl28OQ=
 github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
@@ -187,6 +189,7 @@ github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/
 github.com/google/go-cmp v0.5.6 h1:BKbKCqvP6I+rmFHt06ZmyQtvB8xAkWdhFyr0ZUNZcxQ=
 github.com/google/go-cmp v0.5.6/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
+github.com/google/martian/v3 v3.2.1 h1:d8MncMlErDFTwQGBK1xhv026j9kqhvw1Qv9IbWT1VLQ=
 github.com/google/martian/v3 v3.2.1/go.mod h1:oBOf6HBosgwRXnUGWUB05QECsc6uvmMiJ3+6W4l/CUk=
 github.com/google/pprof v0.0.0-20190930153522-6ce02741cba3 h1:3CYI9xg87xNAD+es02gZxbX/ky4KQeoFBsNOzuoAQZg=
 github.com/google/pprof v0.0.0-20190930153522-6ce02741cba3/go.mod h1:zfwlbNMJ+OItoe0UupaVj+oy1omPYYDuagoSzA8v9mc=
@@ -197,8 +200,9 @@ github.com/google/subcommands v1.2.0/go.mod h1:ZjhPrFU+Olkh9WazFPsl27BQ4UPiG37m3
 github.com/google/uuid v1.0.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/google/uuid v1.1.1 h1:Gkbcsh/GbpXz7lPftLA3P6TYMwjCLYm83jiFQZF/3gY=
 github.com/google/uuid v1.1.1/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
-github.com/googleapis/gax-go/v2 v2.0.5 h1:sjZBwGj9Jlw33ImPtvFviGYvseOtDM7hkSKB7+Tv3SM=
 github.com/googleapis/gax-go/v2 v2.0.5/go.mod h1:DWXyrwAJ9X0FpwwEdw+IPEYBICEFu5mhpdKc/us6bOk=
+github.com/googleapis/gax-go/v2 v2.1.0 h1:6DWmvNpomjL1+3liNSZbVns3zsYzzCjm6pRBO1tLeso=
+github.com/googleapis/gax-go/v2 v2.1.0/go.mod h1:Q3nei7sK6ybPYH7twZdmQpAd1MKb7pfu6SK+H1/DsU0=
 github.com/gorilla/handlers v1.5.0 h1:4wjo3sf9azi99c8hTmyaxp9y5S+pFszsy3pP0rAw/lw=
 github.com/gorilla/handlers v1.5.0/go.mod h1:t8XrUpc4KVXb7HGyJ4/cEnwQiaxrX/hz1Zv/4g96P1Q=
 github.com/gorilla/mux v1.7.3 h1:gnP5JzjVOuiZD07fKKToCAOjS0yOpj/qPETTXCCS6hw=
@@ -558,8 +562,9 @@ golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAG
 golang.org/x/oauth2 v0.0.0-20190604053449-0f29369cfe45/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw=
 golang.org/x/oauth2 v0.0.0-20191202225959-858c2ad4c8b6/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw=
 golang.org/x/oauth2 v0.0.0-20200107190931-bf48bf16ab8d/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw=
-golang.org/x/oauth2 v0.0.0-20210805134026-6f1e6394065a h1:4Kd8OPUx1xgUwrHDaviWZO8MsgoZTZYC3g+8m16RBww=
 golang.org/x/oauth2 v0.0.0-20210805134026-6f1e6394065a/go.mod h1:KelEdhl1UZF7XfJ4dDtk6s++YSgaE7mD/BuKKDLBl4A=
+golang.org/x/oauth2 v0.0.0-20210819190943-2bc19b11175f h1:Qmd2pbz05z7z6lm0DrgQVVPuBm92jqujBKMHMOlOQEw=
+golang.org/x/oauth2 v0.0.0-20210819190943-2bc19b11175f/go.mod h1:KelEdhl1UZF7XfJ4dDtk6s++YSgaE7mD/BuKKDLBl4A=
 golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20181108010431-42b317875d0f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20181221193216-37e7f081c4d4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
@@ -603,8 +608,9 @@ golang.org/x/sys v0.0.0-20210330210617-4fbd30eecc44/go.mod h1:h1NjWce9XRLGQEsW7w
 golang.org/x/sys v0.0.0-20210423082822-04245dca01da/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210510120138-977fb7262007/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20210806184541-e5e7981a1069/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.0.0-20210809222454-d867a43fc93e h1:WUoyKPm6nCo1BnNUvPGnFG3T5DUVem42yDJZZ4CNxMA=
 golang.org/x/sys v0.0.0-20210809222454-d867a43fc93e/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.0.0-20210908233432-aa78b53d3365 h1:6wSTsvPddg9gc/mVEEyk9oOAoxn+bT4Z9q1zx+4RwA4=
+golang.org/x/sys v0.0.0-20210908233432-aa78b53d3365/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.1-0.20180807135948-17ff2d5776d2/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
@@ -674,8 +680,9 @@ google.golang.org/api v0.20.0/go.mod h1:BwFmGc8tA3vsd7r/7kR8DY7iEEGSU04BFxCo5jP/
 google.golang.org/api v0.22.0/go.mod h1:BwFmGc8tA3vsd7r/7kR8DY7iEEGSU04BFxCo5jP/sfE=
 google.golang.org/api v0.24.0/go.mod h1:lIXQywCXRcnZPGlsd8NbLnOjtAoL6em04bJ9+z0MncE=
 google.golang.org/api v0.28.0/go.mod h1:lIXQywCXRcnZPGlsd8NbLnOjtAoL6em04bJ9+z0MncE=
-google.golang.org/api v0.54.0 h1:ECJUVngj71QI6XEm7b1sAf8BljU5inEhMbKPR8Lxhhk=
 google.golang.org/api v0.54.0/go.mod h1:7C4bFFOvVDGXjfDTAsgGwDgAxRDeQ4X8NvUedIt6z3k=
+google.golang.org/api v0.57.0 h1:4t9zuDlHLcIx0ZEhmXEeFVCRsiOgpgn2QOH9N0MNjPI=
+google.golang.org/api v0.57.0/go.mod h1:dVPlbZyBo2/OjBpmvNdpn2GRm6rPy75jyU7bmhdrMgI=
 google.golang.org/appengine v1.4.0/go.mod h1:xpcJRLb0r/rnEns0DIKYYv+WjYCduHsrkT7/EB5XEv4=
 google.golang.org/appengine v1.5.0/go.mod h1:xpcJRLb0r/rnEns0DIKYYv+WjYCduHsrkT7/EB5XEv4=
 google.golang.org/appengine v1.6.5/go.mod h1:8WjMMxjGQR8xUklV/ARdw2HLXBOI7O7uCIDZVag1xfc=
diff --git a/internal/convert.go b/internal/convert.go
index 91ba61d..356ca4b 100644
--- a/internal/convert.go
+++ b/internal/convert.go
@@ -28,6 +28,7 @@ type Conv struct {
 	SpSchema       ddl.Schema                          // Maps Spanner table name to Spanner schema.
 	SyntheticPKeys map[string]SyntheticPKey            // Maps Spanner table name to synthetic primary key (if needed).
 	SrcSchema      map[string]schema.Table             // Maps source-DB table name to schema information.
+	SrcDbName      string                              // Name of the source database. Currently only used for mysql infoschema.
 	Issues         map[string]map[string][]SchemaIssue // Maps source-DB table/col to list of schema conversion issues.
 	ToSpanner      map[string]NameAndCols              // Maps from source-DB table name to Spanner name and column mapping.
 	ToSource       map[string]NameAndCols              // Maps from Spanner table name to source-DB table name and column mapping.
@@ -36,8 +37,9 @@ type Conv struct {
 	Location       *time.Location // Timezone (for timestamp conversion).
 	sampleBadRows  rowSamples     // Rows that generated errors during conversion.
 	Stats          stats
-	TimezoneOffset string // Timezone offset for timestamp conversion.
-	TargetDb       string // The target database to which HarbourBridge is writing.
+	TimezoneOffset string              // Timezone offset for timestamp conversion.
+	TargetDb       string              // The target database to which HarbourBridge is writing.
+	UniquePKey     map[string][]string // Maps Spanner table name to unique column name being used as primary key (if needed).
 }
 
 type mode int
@@ -140,6 +142,7 @@ func MakeConv() *Conv {
 			Unexpected: make(map[string]int64),
 		},
 		TimezoneOffset: "+00:00", // By default, use +00:00 offset which is equal to UTC timezone
+		UniquePKey:     make(map[string][]string),
 	}
 }
 
@@ -256,12 +259,30 @@ func (conv *Conv) SampleBadRows(n int) []string {
 func (conv *Conv) AddPrimaryKeys() {
 	for t, ct := range conv.SpSchema {
 		if len(ct.Pks) == 0 {
-			k := conv.buildPrimaryKey(t)
-			ct.ColNames = append(ct.ColNames, k)
-			ct.ColDefs[k] = ddl.ColumnDef{Name: k, T: ddl.Type{Name: ddl.Int64}}
-			ct.Pks = []ddl.IndexKey{{Col: k}}
+			primaryKeyPopulated := false
+			// Populating column with unique constraint as primary key in case
+			// table doesn't have primary key and removing the unique index.
+			if len(ct.Indexes) != 0 {
+				for i, index := range ct.Indexes {
+					if index.Unique {
+						for _, indexKey := range index.Keys {
+							ct.Pks = append(ct.Pks, ddl.IndexKey{Col: indexKey.Col, Desc: indexKey.Desc})
+							conv.UniquePKey[t] = append(conv.UniquePKey[t], indexKey.Col)
+						}
+						primaryKeyPopulated = true
+						ct.Indexes = append(ct.Indexes[:i], ct.Indexes[i+1:]...)
+						break
+					}
+				}
+			}
+			if !primaryKeyPopulated {
+				k := conv.buildPrimaryKey(t)
+				ct.ColNames = append(ct.ColNames, k)
+				ct.ColDefs[k] = ddl.ColumnDef{Name: k, T: ddl.Type{Name: ddl.Int64}}
+				ct.Pks = []ddl.IndexKey{{Col: k}}
+				conv.SyntheticPKeys[t] = SyntheticPKey{k, 0}
+			}
 			conv.SpSchema[t] = ct
-			conv.SyntheticPKeys[t] = SyntheticPKey{k, 0}
 		}
 	}
 }
diff --git a/internal/report.go b/internal/report.go
index 44b96cb..5658186 100644
--- a/internal/report.go
+++ b/internal/report.go
@@ -127,9 +127,11 @@ func buildTableReport(conv *Conv, srcTable string, badWrites map[string]int64) t
 	tr.Warnings = warnings
 	if pk, ok := conv.SyntheticPKeys[spTable]; ok {
 		tr.SyntheticPKey = pk.Col
-		tr.Body = buildTableReportBody(conv, srcTable, issues, spSchema, srcSchema, &pk.Col)
+		tr.Body = buildTableReportBody(conv, srcTable, issues, spSchema, srcSchema, &pk.Col, nil)
+	} else if pk, ok := conv.UniquePKey[spTable]; ok {
+		tr.Body = buildTableReportBody(conv, srcTable, issues, spSchema, srcSchema, nil, pk)
 	} else {
-		tr.Body = buildTableReportBody(conv, srcTable, issues, spSchema, srcSchema, nil)
+		tr.Body = buildTableReportBody(conv, srcTable, issues, spSchema, srcSchema, nil, nil)
 	}
 	if !conv.SchemaMode() {
 		fillRowStats(conv, srcTable, badWrites, &tr)
@@ -137,7 +139,7 @@ func buildTableReport(conv *Conv, srcTable string, badWrites map[string]int64) t
 	return tr
 }
 
-func buildTableReportBody(conv *Conv, srcTable string, issues map[string][]SchemaIssue, spSchema ddl.CreateTable, srcSchema schema.Table, syntheticPK *string) []tableReportBody {
+func buildTableReportBody(conv *Conv, srcTable string, issues map[string][]SchemaIssue, spSchema ddl.CreateTable, srcSchema schema.Table, syntheticPK *string, uniquePK []string) []tableReportBody {
 	var body []tableReportBody
 	for _, p := range []struct {
 		heading  string
@@ -161,6 +163,13 @@ func buildTableReportBody(conv *Conv, srcTable string, issues map[string][]Schem
 				l = append(l, fmt.Sprintf("Column '%s' was added because this table didn't have a primary key. Spanner requires a primary key for every table", *syntheticPK))
 			}
 		}
+		if uniquePK != nil {
+			// Warning about using a column with unique constraint as primary key
+			// in case primary key is absent.
+			if p.severity == warning {
+				l = append(l, fmt.Sprintf("UNIQUE constraint on column(s) '%s' replaced with primary key since this table didn't have one. Spanner requires a primary key for every table", strings.Join(uniquePK, ", ")))
+			}
+		}
 		issueBatcher := make(map[SchemaIssue]bool)
 		for _, srcCol := range cols {
 			for _, i := range issues[srcCol] {
diff --git a/main.go b/main.go
index cfd2204..64a7162 100644
--- a/main.go
+++ b/main.go
@@ -111,6 +111,9 @@ func main() {
 		flag.Parse()
 		os.Exit(int(subcommands.Execute(ctx)))
 	}
+	fmt.Printf("\nWarning: Found usage of deprecated flags. Support for these " +
+		"flags will be discontinued soon.\nIt is recommended to use Harbourbridge " +
+		"using connection profiles. Checkout usage here: https://github.com/cloudspannerecosystem/harbourbridge/tree/master/cmd#command-line-flags\n\n")
 	// Running HB CLI in global command line mode.
 	setupGlobalFlags()
 	flag.Usage = usage
diff --git a/sources/dynamodb/README.md b/sources/dynamodb/README.md
index b6f1af5..5d076e9 100644
--- a/sources/dynamodb/README.md
+++ b/sources/dynamodb/README.md
@@ -48,7 +48,7 @@ harbourbridge -driver=dynamodb -schema-sample-size=500000
 
 The HarbourBridge tool maps DynamoDB types to Spanner types as follows:
 
-| PostgreSQL Type    | Spanner Type               | Notes                                     |
+| DynamoDB Type    | Spanner Type               | Notes                                     |
 | ------------------ | -------------------------- | ----------------------------------------- |
 | `Number`           | `NUMERIC` or `STRING`      | defaults to NUMERIC, otherwise, STRING    |
 | `String`           | `STRING`                   |                                           |
diff --git a/web/web.go b/web/web.go
index 663910d..13324c8 100644
--- a/web/web.go
+++ b/web/web.go
@@ -168,7 +168,7 @@ func convertSchemaDump(w http.ResponseWriter, r *http.Request) {
 		http.Error(w, fmt.Sprintf("failed to open dump file %v : %v", dc.FilePath, err), http.StatusNotFound)
 		return
 	}
-	conv, err := conversion.SchemaConv(dc.Driver, constants.TARGET_SPANNER, &conversion.IOStreams{In: f, Out: os.Stdout}, 0)
+	conv, err := conversion.SchemaConv(dc.Driver, "", constants.TARGET_SPANNER, &conversion.IOStreams{In: f, Out: os.Stdout}, 0)
 	if err != nil {
 		http.Error(w, fmt.Sprintf("Schema Conversion Error : %v", err), http.StatusNotFound)
 		return


Test patch:
diff --git a/cmd/source_profile_test.go b/cmd/source_profile_test.go
index 1b0e5dc..b7a3a76 100644
--- a/cmd/source_profile_test.go
+++ b/cmd/source_profile_test.go
@@ -53,3 +53,85 @@ func TestNewSourceProfileFile(t *testing.T) {
 		assert.Equal(t, profile, tc.want, tc.name)
 	}
 }
+
+func TestNewSourceProfileConnectionSQL(t *testing.T) {
+	// Avoid getting/settinng env variables in the unit tests.
+	testCases := []struct {
+		name          string
+		params        map[string]string
+		errorExpected bool
+	}{
+		{
+			name:          "mandatory params provided",
+			params:        map[string]string{"host": "a", "user": "b", "db_name": "c", "password": "e"},
+			errorExpected: false,
+		},
+		{
+			name:          "partial mandatory params provided",
+			params:        map[string]string{"user": "b", "db_name": "c"},
+			errorExpected: true,
+		},
+		{
+			name:          "no mandatory params but optional provided",
+			params:        map[string]string{"port": "b"},
+			errorExpected: true,
+		},
+		{
+			name:          "partial mandatory params and optional provided",
+			params:        map[string]string{"host": "a", "port": "b"},
+			errorExpected: true,
+		},
+		{
+			name:          "all params provided",
+			params:        map[string]string{"host": "a", "user": "b", "db_name": "c", "port": "d", "password": "e"},
+			errorExpected: false,
+		},
+		{
+			name:          "empty mandatory param",
+			params:        map[string]string{"host": "", "user": "b", "db_name": "c"},
+			errorExpected: true,
+		},
+		{
+			name:          "empty port",
+			params:        map[string]string{"host": "a", "user": "b", "db_name": "c", "password": "e"},
+			errorExpected: false,
+		},
+	}
+
+	for _, tc := range testCases {
+		_, pgErr := NewSourceProfileConnectionPostgreSQL(tc.params)
+		_, mysqlErr := NewSourceProfileConnectionMySQL(tc.params)
+		assert.Equal(t, tc.errorExpected, pgErr != nil)
+		assert.Equal(t, tc.errorExpected, mysqlErr != nil)
+	}
+}
+
+func TestNewSourceProfileConnectionDynamoDB(t *testing.T) {
+	// Avoid getting/settinng env variables in the unit tests.
+	testCases := []struct {
+		name          string
+		params        map[string]string
+		errorExpected bool
+	}{
+		{
+			name:          "no params",
+			params:        map[string]string{},
+			errorExpected: false,
+		},
+		{
+			name:          "valid schema sample size",
+			params:        map[string]string{"schemaSampleSize": "15"},
+			errorExpected: false,
+		},
+		{
+			name:          "invalid schema sample size",
+			params:        map[string]string{"schemaSampleSize": "a"},
+			errorExpected: true,
+		},
+	}
+
+	for _, tc := range testCases {
+		_, err := NewSourceProfileConnectionDynamoDB(tc.params)
+		assert.Equal(t, tc.errorExpected, err != nil)
+	}
+}
diff --git a/internal/convert_test.go b/internal/convert_test.go
index aa972c5..82832ab 100644
--- a/internal/convert_test.go
+++ b/internal/convert_test.go
@@ -84,25 +84,187 @@ func TestGetBadRows(t *testing.T) {
 }
 
 func TestAddPrimaryKeys(t *testing.T) {
-	conv := MakeConv()
-	conv.SpSchema["table"] = ddl.CreateTable{
-		Name:     "table",
-		ColNames: []string{"a", "b"},
-		ColDefs: map[string]ddl.ColumnDef{
-			"a": {Name: "a", T: ddl.Type{Name: ddl.Int64}},
-			"b": {Name: "b", T: ddl.Type{Name: ddl.Float64}},
+	addPrimaryKeyTests := []struct {
+		name           string
+		inputSchema    map[string]ddl.CreateTable
+		expectedSchema map[string]ddl.CreateTable
+		syntheticKey   string
+		uniqueKey      []string
+	}{
+		{
+			name: "primary key already exists",
+			inputSchema: map[string]ddl.CreateTable{
+				"table": {
+					Name:     "table",
+					ColNames: []string{"a", "b"},
+					ColDefs: map[string]ddl.ColumnDef{
+						"a": {Name: "a", T: ddl.Type{Name: ddl.Int64}},
+						"b": {Name: "b", T: ddl.Type{Name: ddl.Float64}},
+					},
+					Pks: []ddl.IndexKey{{Col: "a"}}}},
+
+			expectedSchema: map[string]ddl.CreateTable{
+				"table": {
+					Name:     "table",
+					ColNames: []string{"a", "b"},
+					ColDefs: map[string]ddl.ColumnDef{
+						"a": {Name: "a", T: ddl.Type{Name: ddl.Int64}},
+						"b": {Name: "b", T: ddl.Type{Name: ddl.Float64}},
+					},
+					Pks: []ddl.IndexKey{{Col: "a"}},
+				},
+			},
+		},
+		{
+			name: "single unique key as primary key",
+			inputSchema: map[string]ddl.CreateTable{
+				"table": {
+					Name:     "table",
+					ColNames: []string{"a", "b"},
+					ColDefs: map[string]ddl.ColumnDef{
+						"a": {Name: "a", T: ddl.Type{Name: ddl.Int64}},
+						"b": {Name: "b", T: ddl.Type{Name: ddl.Float64}},
+					},
+					Pks:     []ddl.IndexKey{},
+					Indexes: []ddl.CreateIndex{{Name: "", Table: "", Unique: true, Keys: []ddl.IndexKey{{Col: "b"}}}},
+				},
+			},
+			expectedSchema: map[string]ddl.CreateTable{
+				"table": {
+					Name:     "table",
+					ColNames: []string{"a", "b"},
+					ColDefs: map[string]ddl.ColumnDef{
+						"a": {Name: "a", T: ddl.Type{Name: ddl.Int64}},
+						"b": {Name: "b", T: ddl.Type{Name: ddl.Float64}},
+					},
+					Pks:     []ddl.IndexKey{{Col: "b"}},
+					Indexes: []ddl.CreateIndex{}},
+			},
+			uniqueKey: []string{"b"},
+		},
+		{
+			name: "multiple unique key as primary key",
+			inputSchema: map[string]ddl.CreateTable{
+				"table": {
+					Name:     "table",
+					ColNames: []string{"a", "b"},
+					ColDefs: map[string]ddl.ColumnDef{
+						"a": {Name: "a", T: ddl.Type{Name: ddl.Int64}},
+						"b": {Name: "b", T: ddl.Type{Name: ddl.Float64}},
+					},
+					Pks:     []ddl.IndexKey{},
+					Indexes: []ddl.CreateIndex{{Name: "", Table: "", Unique: true, Keys: []ddl.IndexKey{{Col: "a"}, {Col: "b"}}}},
+				},
+			},
+			expectedSchema: map[string]ddl.CreateTable{
+				"table": {
+					Name:     "table",
+					ColNames: []string{"a", "b"},
+					ColDefs: map[string]ddl.ColumnDef{
+						"a": {Name: "a", T: ddl.Type{Name: ddl.Int64}},
+						"b": {Name: "b", T: ddl.Type{Name: ddl.Float64}},
+					},
+					Pks:     []ddl.IndexKey{{Col: "a"}, {Col: "b"}},
+					Indexes: []ddl.CreateIndex{}},
+			},
+			uniqueKey: []string{"a", "b"},
+		},
+		{
+			name: "in case two unique keys are present first one is taken as primary key",
+			inputSchema: map[string]ddl.CreateTable{
+				"table": {
+					Name:     "table",
+					ColNames: []string{"a", "b"},
+					ColDefs: map[string]ddl.ColumnDef{
+						"a": {Name: "a", T: ddl.Type{Name: ddl.Int64}},
+						"b": {Name: "b", T: ddl.Type{Name: ddl.Float64}},
+					},
+					Pks:     []ddl.IndexKey{},
+					Indexes: []ddl.CreateIndex{{Name: "", Table: "", Unique: true, Keys: []ddl.IndexKey{{Col: "a"}}}, {Name: "", Table: "", Unique: true, Keys: []ddl.IndexKey{{Col: "b"}}}},
+				},
+			},
+			expectedSchema: map[string]ddl.CreateTable{
+				"table": {
+					Name:     "table",
+					ColNames: []string{"a", "b"},
+					ColDefs: map[string]ddl.ColumnDef{
+						"a": {Name: "a", T: ddl.Type{Name: ddl.Int64}},
+						"b": {Name: "b", T: ddl.Type{Name: ddl.Float64}},
+					},
+					Pks:     []ddl.IndexKey{{Col: "a"}},
+					Indexes: []ddl.CreateIndex{{Name: "", Table: "", Unique: true, Keys: []ddl.IndexKey{{Col: "b"}}}},
+				},
+			},
+			uniqueKey: []string{"a"},
+		},
+		{
+			name: "unique index doesn't exist so synthetic primary key created",
+			inputSchema: map[string]ddl.CreateTable{
+				"table": {
+					Name:     "table",
+					ColNames: []string{"a", "b"},
+					ColDefs: map[string]ddl.ColumnDef{
+						"a": {Name: "a", T: ddl.Type{Name: ddl.Int64}},
+						"b": {Name: "b", T: ddl.Type{Name: ddl.Float64}},
+					},
+					Pks:     []ddl.IndexKey{},
+					Indexes: []ddl.CreateIndex{{Name: "", Table: "", Unique: false, Keys: []ddl.IndexKey{{Col: "b"}}}},
+				},
+			},
+			expectedSchema: map[string]ddl.CreateTable{
+				"table": {
+					Name:     "table",
+					ColNames: []string{"a", "b", "synth_id"},
+					ColDefs: map[string]ddl.ColumnDef{
+						"a":        {Name: "a", T: ddl.Type{Name: ddl.Int64}},
+						"b":        {Name: "b", T: ddl.Type{Name: ddl.Float64}},
+						"synth_id": {Name: "synth_id", T: ddl.Type{Name: ddl.Int64}},
+					},
+					Pks:     []ddl.IndexKey{{Col: "synth_id"}},
+					Indexes: []ddl.CreateIndex{{Name: "", Table: "", Unique: false, Keys: []ddl.IndexKey{{Col: "b"}}}},
+				},
+			},
+			syntheticKey: "synth_id",
 		},
-		Pks: []ddl.IndexKey{}}
-	conv.AddPrimaryKeys()
-	e := ddl.CreateTable{
-		Name:     "table",
-		ColNames: []string{"a", "b", "synth_id"},
-		ColDefs: map[string]ddl.ColumnDef{
-			"a":        {Name: "a", T: ddl.Type{Name: ddl.Int64}},
-			"b":        {Name: "b", T: ddl.Type{Name: ddl.Float64}},
-			"synth_id": {Name: "synth_id", T: ddl.Type{Name: ddl.Int64}},
+		{
+			name: "synthetic key as primary key",
+			inputSchema: map[string]ddl.CreateTable{
+				"table": {
+					Name:     "table",
+					ColNames: []string{"a", "b"},
+					ColDefs: map[string]ddl.ColumnDef{
+						"a": {Name: "a", T: ddl.Type{Name: ddl.Int64}},
+						"b": {Name: "b", T: ddl.Type{Name: ddl.Float64}},
+					},
+					Pks: []ddl.IndexKey{},
+				},
+			},
+			expectedSchema: map[string]ddl.CreateTable{
+				"table": {
+					Name:     "table",
+					ColNames: []string{"a", "b", "synth_id"},
+					ColDefs: map[string]ddl.ColumnDef{
+						"a":        {Name: "a", T: ddl.Type{Name: ddl.Int64}},
+						"b":        {Name: "b", T: ddl.Type{Name: ddl.Float64}},
+						"synth_id": {Name: "synth_id", T: ddl.Type{Name: ddl.Int64}},
+					},
+					Pks: []ddl.IndexKey{{Col: "synth_id"}}},
+			},
+			syntheticKey: "synth_id",
 		},
-		Pks: []ddl.IndexKey{{Col: "synth_id"}}}
-	assert.Equal(t, e, conv.SpSchema["table"])
-	assert.Equal(t, SyntheticPKey{Col: "synth_id", Sequence: 0}, conv.SyntheticPKeys["table"])
+	}
+	for _, tc := range addPrimaryKeyTests {
+		conv := MakeConv()
+		conv.SpSchema = tc.inputSchema
+		conv.AddPrimaryKeys()
+		if tc.expectedSchema != nil {
+			assert.Equal(t, tc.expectedSchema["table"], conv.SpSchema["table"])
+		}
+		if tc.uniqueKey != nil {
+			assert.Equal(t, tc.uniqueKey, conv.UniquePKey["table"])
+		}
+		if tc.syntheticKey != "" {
+			assert.Equal(t, tc.syntheticKey, conv.SyntheticPKeys["table"].Col)
+		}
+	}
 }
diff --git a/sources/mysql/mysqldump_test.go b/sources/mysql/mysqldump_test.go
index f1f0a7d..1111e81 100644
--- a/sources/mysql/mysqldump_test.go
+++ b/sources/mysql/mysqldump_test.go
@@ -359,15 +359,14 @@ func TestProcessMySQLDump_MultiCol(t *testing.T) {
 			expectedSchema: map[string]ddl.CreateTable{
 				"test": ddl.CreateTable{
 					Name:     "test",
-					ColNames: []string{"a", "b", "c", "synth_id"},
+					ColNames: []string{"a", "b", "c"},
 					ColDefs: map[string]ddl.ColumnDef{
-						"a":        ddl.ColumnDef{Name: "a", T: ddl.Type{Name: ddl.Int64}},
-						"b":        ddl.ColumnDef{Name: "b", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
-						"c":        ddl.ColumnDef{Name: "c", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
-						"synth_id": ddl.ColumnDef{Name: "synth_id", T: ddl.Type{Name: ddl.Int64}},
+						"a": ddl.ColumnDef{Name: "a", T: ddl.Type{Name: ddl.Int64}},
+						"b": ddl.ColumnDef{Name: "b", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
+						"c": ddl.ColumnDef{Name: "c", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
 					},
-					Pks:     []ddl.IndexKey{ddl.IndexKey{Col: "synth_id"}},
-					Indexes: []ddl.CreateIndex{ddl.CreateIndex{Name: "custom_index", Table: "test", Unique: true, Keys: []ddl.IndexKey{ddl.IndexKey{Col: "b", Desc: false}, ddl.IndexKey{Col: "c", Desc: false}}}}}},
+					Pks:     []ddl.IndexKey{ddl.IndexKey{Col: "b"}, ddl.IndexKey{Col: "c"}},
+					Indexes: []ddl.CreateIndex{}}},
 		},
 		{
 			name: "Create table with multiple index keys with different order",
@@ -381,16 +380,14 @@ func TestProcessMySQLDump_MultiCol(t *testing.T) {
 			expectedSchema: map[string]ddl.CreateTable{
 				"test": ddl.CreateTable{
 					Name:     "test",
-					ColNames: []string{"a", "b", "c", "synth_id"},
+					ColNames: []string{"a", "b", "c"},
 					ColDefs: map[string]ddl.ColumnDef{
-						"a":        ddl.ColumnDef{Name: "a", T: ddl.Type{Name: ddl.Int64}},
-						"b":        ddl.ColumnDef{Name: "b", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
-						"c":        ddl.ColumnDef{Name: "c", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
-						"synth_id": ddl.ColumnDef{Name: "synth_id", T: ddl.Type{Name: ddl.Int64}},
+						"a": ddl.ColumnDef{Name: "a", T: ddl.Type{Name: ddl.Int64}},
+						"b": ddl.ColumnDef{Name: "b", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
+						"c": ddl.ColumnDef{Name: "c", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
 					},
-					Pks: []ddl.IndexKey{ddl.IndexKey{Col: "synth_id"}},
-					Indexes: []ddl.CreateIndex{ddl.CreateIndex{Name: "custom_index", Table: "test", Unique: true, Keys: []ddl.IndexKey{ddl.IndexKey{Col: "b", Desc: false}, ddl.IndexKey{Col: "c", Desc: false}}},
-						ddl.CreateIndex{Name: "custom_index2", Table: "test", Unique: false, Keys: []ddl.IndexKey{ddl.IndexKey{Col: "c", Desc: false}, ddl.IndexKey{Col: "a", Desc: false}}}}}},
+					Pks:     []ddl.IndexKey{ddl.IndexKey{Col: "b"}, ddl.IndexKey{Col: "c"}},
+					Indexes: []ddl.CreateIndex{ddl.CreateIndex{Name: "custom_index2", Table: "test", Unique: false, Keys: []ddl.IndexKey{ddl.IndexKey{Col: "c", Desc: false}, ddl.IndexKey{Col: "a", Desc: false}}}}}},
 		},
 		{
 			name: "Alter table add index keys",
@@ -424,15 +421,15 @@ func TestProcessMySQLDump_MultiCol(t *testing.T) {
 			expectedSchema: map[string]ddl.CreateTable{
 				"test": ddl.CreateTable{
 					Name:     "test",
-					ColNames: []string{"a", "b", "c", "synth_id"},
+					ColNames: []string{"a", "b", "c"},
 					ColDefs: map[string]ddl.ColumnDef{
-						"a":        ddl.ColumnDef{Name: "a", T: ddl.Type{Name: ddl.Int64}},
-						"b":        ddl.ColumnDef{Name: "b", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
-						"c":        ddl.ColumnDef{Name: "c", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
-						"synth_id": ddl.ColumnDef{Name: "synth_id", T: ddl.Type{Name: ddl.Int64}},
+						"a": ddl.ColumnDef{Name: "a", T: ddl.Type{Name: ddl.Int64}},
+						"b": ddl.ColumnDef{Name: "b", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
+						"c": ddl.ColumnDef{Name: "c", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
 					},
-					Pks:     []ddl.IndexKey{ddl.IndexKey{Col: "synth_id"}},
-					Indexes: []ddl.CreateIndex{ddl.CreateIndex{Name: "custom_index", Table: "test", Unique: true, Keys: []ddl.IndexKey{ddl.IndexKey{Col: "b", Desc: false}, ddl.IndexKey{Col: "c", Desc: false}}}}}},
+					Pks:     []ddl.IndexKey{ddl.IndexKey{Col: "b"}, ddl.IndexKey{Col: "c"}},
+					Indexes: []ddl.CreateIndex{},
+				}},
 		},
 		{
 			name: "Alter table add index keys",
@@ -445,15 +442,15 @@ func TestProcessMySQLDump_MultiCol(t *testing.T) {
 			expectedSchema: map[string]ddl.CreateTable{
 				"test": ddl.CreateTable{
 					Name:     "test",
-					ColNames: []string{"a", "b", "c", "synth_id"},
+					ColNames: []string{"a", "b", "c"},
 					ColDefs: map[string]ddl.ColumnDef{
-						"a":        ddl.ColumnDef{Name: "a", T: ddl.Type{Name: ddl.Int64}},
-						"b":        ddl.ColumnDef{Name: "b", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
-						"c":        ddl.ColumnDef{Name: "c", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
-						"synth_id": ddl.ColumnDef{Name: "synth_id", T: ddl.Type{Name: ddl.Int64}},
+						"a": ddl.ColumnDef{Name: "a", T: ddl.Type{Name: ddl.Int64}},
+						"b": ddl.ColumnDef{Name: "b", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
+						"c": ddl.ColumnDef{Name: "c", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
 					},
-					Pks:     []ddl.IndexKey{ddl.IndexKey{Col: "synth_id"}},
-					Indexes: []ddl.CreateIndex{ddl.CreateIndex{Name: "custom_index", Table: "test", Unique: true, Keys: []ddl.IndexKey{ddl.IndexKey{Col: "b", Desc: false}, ddl.IndexKey{Col: "c", Desc: false}}}}}},
+					Pks:     []ddl.IndexKey{ddl.IndexKey{Col: "b"}, ddl.IndexKey{Col: "c"}},
+					Indexes: []ddl.CreateIndex{},
+				}},
 		},
 		{
 			name: "Create index statement",
@@ -487,15 +484,15 @@ func TestProcessMySQLDump_MultiCol(t *testing.T) {
 			expectedSchema: map[string]ddl.CreateTable{
 				"test": ddl.CreateTable{
 					Name:     "test",
-					ColNames: []string{"a", "b", "c", "synth_id"},
+					ColNames: []string{"a", "b", "c"},
 					ColDefs: map[string]ddl.ColumnDef{
-						"a":        ddl.ColumnDef{Name: "a", T: ddl.Type{Name: ddl.Int64}},
-						"b":        ddl.ColumnDef{Name: "b", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
-						"c":        ddl.ColumnDef{Name: "c", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
-						"synth_id": ddl.ColumnDef{Name: "synth_id", T: ddl.Type{Name: ddl.Int64}},
+						"a": ddl.ColumnDef{Name: "a", T: ddl.Type{Name: ddl.Int64}},
+						"b": ddl.ColumnDef{Name: "b", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
+						"c": ddl.ColumnDef{Name: "c", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
 					},
-					Pks:     []ddl.IndexKey{ddl.IndexKey{Col: "synth_id"}},
-					Indexes: []ddl.CreateIndex{ddl.CreateIndex{Name: "custom_index", Table: "test", Unique: true, Keys: []ddl.IndexKey{ddl.IndexKey{Col: "b", Desc: false}, ddl.IndexKey{Col: "c", Desc: false}}}}}},
+					Pks:     []ddl.IndexKey{ddl.IndexKey{Col: "b"}, ddl.IndexKey{Col: "c"}},
+					Indexes: []ddl.CreateIndex{},
+				}},
 		},
 		{
 			name:  "Create table with mysql schema",
diff --git a/sources/postgres/pgdump_test.go b/sources/postgres/pgdump_test.go
index ad66369..d74f8a0 100644
--- a/sources/postgres/pgdump_test.go
+++ b/sources/postgres/pgdump_test.go
@@ -330,15 +330,15 @@ func TestProcessPgDump(t *testing.T) {
 			expectedSchema: map[string]ddl.CreateTable{
 				"test": ddl.CreateTable{
 					Name:     "test",
-					ColNames: []string{"a", "b", "c", "synth_id"},
+					ColNames: []string{"a", "b", "c"},
 					ColDefs: map[string]ddl.ColumnDef{
-						"a":        ddl.ColumnDef{Name: "a", T: ddl.Type{Name: ddl.Int64}},
-						"b":        ddl.ColumnDef{Name: "b", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
-						"c":        ddl.ColumnDef{Name: "c", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
-						"synth_id": ddl.ColumnDef{Name: "synth_id", T: ddl.Type{Name: ddl.Int64}},
+						"a": ddl.ColumnDef{Name: "a", T: ddl.Type{Name: ddl.Int64}},
+						"b": ddl.ColumnDef{Name: "b", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
+						"c": ddl.ColumnDef{Name: "c", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
 					},
-					Pks:     []ddl.IndexKey{ddl.IndexKey{Col: "synth_id"}},
-					Indexes: []ddl.CreateIndex{ddl.CreateIndex{Name: "custom_index", Table: "test", Unique: true, Keys: []ddl.IndexKey{ddl.IndexKey{Col: "c", Desc: true}, ddl.IndexKey{Col: "b", Desc: false}}}}}},
+					Pks:     []ddl.IndexKey{ddl.IndexKey{Col: "c", Desc: true}, ddl.IndexKey{Col: "b", Desc: false}},
+					Indexes: []ddl.CreateIndex{},
+				}},
 		},
 		{
 			name: "Create table with unique constraint",
@@ -351,15 +351,15 @@ func TestProcessPgDump(t *testing.T) {
 			expectedSchema: map[string]ddl.CreateTable{
 				"test": ddl.CreateTable{
 					Name:     "test",
-					ColNames: []string{"a", "b", "c", "synth_id"},
+					ColNames: []string{"a", "b", "c"},
 					ColDefs: map[string]ddl.ColumnDef{
-						"a":        ddl.ColumnDef{Name: "a", T: ddl.Type{Name: ddl.Int64}},
-						"b":        ddl.ColumnDef{Name: "b", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
-						"c":        ddl.ColumnDef{Name: "c", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
-						"synth_id": ddl.ColumnDef{Name: "synth_id", T: ddl.Type{Name: ddl.Int64}},
+						"a": ddl.ColumnDef{Name: "a", T: ddl.Type{Name: ddl.Int64}},
+						"b": ddl.ColumnDef{Name: "b", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
+						"c": ddl.ColumnDef{Name: "c", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
 					},
-					Pks:     []ddl.IndexKey{ddl.IndexKey{Col: "synth_id"}},
-					Indexes: []ddl.CreateIndex{ddl.CreateIndex{Name: "custom_index", Table: "test", Unique: true, Keys: []ddl.IndexKey{ddl.IndexKey{Col: "b", Desc: false}}}}}},
+					Pks:     []ddl.IndexKey{ddl.IndexKey{Col: "b"}},
+					Indexes: []ddl.CreateIndex{},
+				}},
 		},
 		{
 			name: "Alter table add unique constraint",
@@ -372,15 +372,15 @@ func TestProcessPgDump(t *testing.T) {
 			expectedSchema: map[string]ddl.CreateTable{
 				"test": ddl.CreateTable{
 					Name:     "test",
-					ColNames: []string{"a", "b", "c", "synth_id"},
+					ColNames: []string{"a", "b", "c"},
 					ColDefs: map[string]ddl.ColumnDef{
-						"a":        ddl.ColumnDef{Name: "a", T: ddl.Type{Name: ddl.Int64}},
-						"b":        ddl.ColumnDef{Name: "b", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
-						"c":        ddl.ColumnDef{Name: "c", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
-						"synth_id": ddl.ColumnDef{Name: "synth_id", T: ddl.Type{Name: ddl.Int64}},
+						"a": ddl.ColumnDef{Name: "a", T: ddl.Type{Name: ddl.Int64}},
+						"b": ddl.ColumnDef{Name: "b", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
+						"c": ddl.ColumnDef{Name: "c", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
 					},
-					Pks:     []ddl.IndexKey{ddl.IndexKey{Col: "synth_id"}},
-					Indexes: []ddl.CreateIndex{ddl.CreateIndex{Name: "custom_index", Table: "test", Unique: true, Keys: []ddl.IndexKey{ddl.IndexKey{Col: "b", Desc: false}, ddl.IndexKey{Col: "c", Desc: false}}}}}},
+					Pks:     []ddl.IndexKey{ddl.IndexKey{Col: "b"}, ddl.IndexKey{Col: "c"}},
+					Indexes: []ddl.CreateIndex{},
+				}},
 		},
 		{
 			name:  "Create table with pg schema",
@@ -998,15 +998,15 @@ func TestProcessPgDumpPGTarget(t *testing.T) {
 			expectedSchema: map[string]ddl.CreateTable{
 				"test": ddl.CreateTable{
 					Name:     "test",
-					ColNames: []string{"a", "b", "c", "synth_id"},
+					ColNames: []string{"a", "b", "c"},
 					ColDefs: map[string]ddl.ColumnDef{
-						"a":        ddl.ColumnDef{Name: "a", T: ddl.Type{Name: ddl.Int64}},
-						"b":        ddl.ColumnDef{Name: "b", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
-						"c":        ddl.ColumnDef{Name: "c", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
-						"synth_id": ddl.ColumnDef{Name: "synth_id", T: ddl.Type{Name: ddl.Int64}},
+						"a": ddl.ColumnDef{Name: "a", T: ddl.Type{Name: ddl.Int64}},
+						"b": ddl.ColumnDef{Name: "b", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
+						"c": ddl.ColumnDef{Name: "c", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
 					},
-					Pks:     []ddl.IndexKey{ddl.IndexKey{Col: "synth_id"}},
-					Indexes: []ddl.CreateIndex{ddl.CreateIndex{Name: "custom_index", Table: "test", Unique: true, Keys: []ddl.IndexKey{ddl.IndexKey{Col: "c", Desc: true}, ddl.IndexKey{Col: "b", Desc: false}}}}}},
+					Pks:     []ddl.IndexKey{ddl.IndexKey{Col: "c", Desc: true}, ddl.IndexKey{Col: "b", Desc: false}},
+					Indexes: []ddl.CreateIndex{},
+				}},
 		},
 		{
 			name: "Create table with unique constraint",
@@ -1019,15 +1019,15 @@ func TestProcessPgDumpPGTarget(t *testing.T) {
 			expectedSchema: map[string]ddl.CreateTable{
 				"test": ddl.CreateTable{
 					Name:     "test",
-					ColNames: []string{"a", "b", "c", "synth_id"},
+					ColNames: []string{"a", "b", "c"},
 					ColDefs: map[string]ddl.ColumnDef{
-						"a":        ddl.ColumnDef{Name: "a", T: ddl.Type{Name: ddl.Int64}},
-						"b":        ddl.ColumnDef{Name: "b", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
-						"c":        ddl.ColumnDef{Name: "c", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
-						"synth_id": ddl.ColumnDef{Name: "synth_id", T: ddl.Type{Name: ddl.Int64}},
+						"a": ddl.ColumnDef{Name: "a", T: ddl.Type{Name: ddl.Int64}},
+						"b": ddl.ColumnDef{Name: "b", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
+						"c": ddl.ColumnDef{Name: "c", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
 					},
-					Pks:     []ddl.IndexKey{ddl.IndexKey{Col: "synth_id"}},
-					Indexes: []ddl.CreateIndex{ddl.CreateIndex{Name: "custom_index", Table: "test", Unique: true, Keys: []ddl.IndexKey{ddl.IndexKey{Col: "b", Desc: false}}}}}},
+					Pks:     []ddl.IndexKey{ddl.IndexKey{Col: "b"}},
+					Indexes: []ddl.CreateIndex{},
+				}},
 		},
 		{
 			name: "Alter table add unique constraint",
@@ -1040,15 +1040,15 @@ func TestProcessPgDumpPGTarget(t *testing.T) {
 			expectedSchema: map[string]ddl.CreateTable{
 				"test": ddl.CreateTable{
 					Name:     "test",
-					ColNames: []string{"a", "b", "c", "synth_id"},
+					ColNames: []string{"a", "b", "c"},
 					ColDefs: map[string]ddl.ColumnDef{
-						"a":        ddl.ColumnDef{Name: "a", T: ddl.Type{Name: ddl.Int64}},
-						"b":        ddl.ColumnDef{Name: "b", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
-						"c":        ddl.ColumnDef{Name: "c", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
-						"synth_id": ddl.ColumnDef{Name: "synth_id", T: ddl.Type{Name: ddl.Int64}},
+						"a": ddl.ColumnDef{Name: "a", T: ddl.Type{Name: ddl.Int64}},
+						"b": ddl.ColumnDef{Name: "b", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
+						"c": ddl.ColumnDef{Name: "c", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
 					},
-					Pks:     []ddl.IndexKey{ddl.IndexKey{Col: "synth_id"}},
-					Indexes: []ddl.CreateIndex{ddl.CreateIndex{Name: "custom_index", Table: "test", Unique: true, Keys: []ddl.IndexKey{ddl.IndexKey{Col: "b", Desc: false}, ddl.IndexKey{Col: "c", Desc: false}}}}}},
+					Pks:     []ddl.IndexKey{ddl.IndexKey{Col: "b"}, ddl.IndexKey{Col: "c"}},
+					Indexes: []ddl.CreateIndex{},
+				}},
 		},
 		{
 			name:  "Create table with pg schema",
diff --git a/testing/common/testutil.go b/testing/common/testutil.go
index 44b7c77..f754f89 100644
--- a/testing/common/testutil.go
+++ b/testing/common/testutil.go
@@ -27,3 +27,20 @@ func RunCommand(args string, projectID string) error {
 	}
 	return nil
 }
+
+// Clears the env variables specified in the input list and stashes the values
+// in a map.
+func ClearEnvVariables(vars []string) map[string]string {
+	envVars := make(map[string]string)
+	for _, v := range vars {
+		envVars[v] = os.Getenv(v)
+		os.Setenv(v, "")
+	}
+	return envVars
+}
+
+func RestoreEnvVariables(params map[string]string) {
+	for k, v := range params {
+		os.Setenv(k, v)
+	}
+}
diff --git a/testing/mysql/integration_test.go b/testing/mysql/integration_test.go
index 82d927c..6af891a 100644
--- a/testing/mysql/integration_test.go
+++ b/testing/mysql/integration_test.go
@@ -126,6 +126,31 @@ func TestIntegration_MYSQLDUMP_Command(t *testing.T) {
 	checkResults(t, dbURI)
 }
 
+func TestIntegration_MySQL_EvalSubcommand(t *testing.T) {
+	onlyRunForEmulatorTest(t)
+	t.Parallel()
+
+	tmpdir := prepareIntegrationTest(t)
+	defer os.RemoveAll(tmpdir)
+
+	dbName := "mysql-dc-eval-src-params"
+	dbURI := fmt.Sprintf("projects/%s/instances/%s/databases/%s", projectID, instanceID, dbName)
+	filePrefix := filepath.Join(tmpdir, dbName+".")
+
+	host, user, db_name, password := os.Getenv("MYSQLHOST"), os.Getenv("MYSQLUSER"), os.Getenv("MYSQLDATABASE"), os.Getenv("MYSQLPWD")
+	envVars := common.ClearEnvVariables([]string{"MYSQLHOST", "MYSQLUSER", "MYSQLDATABASE", "MYSQLPWD"})
+	args := fmt.Sprintf("eval -source=%s -prefix=%s -source-profile='host=%s,user=%s,db_name=%s,password=%s' -target-profile='instance=%s,dbname=%s'", constants.MYSQL, filePrefix, host, user, db_name, password, instanceID, dbName)
+	err := common.RunCommand(args, projectID)
+	common.RestoreEnvVariables(envVars)
+	if err != nil {
+		t.Fatal(err)
+	}
+	// Drop the database later.
+	defer dropDatabase(t, dbURI)
+
+	checkResults(t, dbURI)
+}
+
 func TestIntegration_MYSQL_Command(t *testing.T) {
 	onlyRunForEmulatorTest(t)
 	t.Parallel()