Repository: cloudspannerecosystem/harbourbridge
Pull request #510 (2023-04-10 10:28:10)

CLI support for sharded migrations for MySQL
Implements the following - 

1. Configuration interface for accepting sharded configurations.
2. Serial bulk migrations via HB CLI.
3. Parallel minimal downtime sharded migrations via HB CLI.

Bulk Migration - 

```
./harbourbridge schema-and-data -source=mysql --target-profile="project=span-cloud-testing,instance=manit-testing,dbName=bulkyboo" --source-profile="config=shardbulk.cfg" --log-level=DEBUG
```

Minimal Downtime Migration - 

```
./harbourbridge schema-and-data -source=mysql --target-profile="project=span-cloud-testing,instance=manit-testing,dbName=minimal" --source-profile="config=shardConfigNew.cfg" --log-level=DEBUG
```

Note: Currently sharded migrations are only supported for MySQL source.

manitgupta: Resolved all comments, and have also improved the unit tests a bit.
bharadwaj-aditya: Please mention in the description of the PR, the support is only for MySQL.
manitgupta: > Please mention in the description of the PR, the support is only for MySQL.

Its part of the PR title 😛, also updated the description.
bharadwaj-aditya: LGTM Thanks !

Patch:
diff --git a/common/constants/constants.go b/common/constants/constants.go
index 93d1349..bd452a0 100644
--- a/common/constants/constants.go
+++ b/common/constants/constants.go
@@ -71,4 +71,10 @@ const (
 	// Rule types
 	GlobalDataTypeChange = "global_datatype_change"
 	AddIndex             = "add_index"
+	//bulk migration type 
+	BULK_MIGRATION = "bulk"
+	//dataflow migration type
+	DATAFLOW_MIGRATION = "dataflow"
+	//DMS migration type 
+	DMS_MIGRATION = "dms"
 )
diff --git a/conversion/conversion.go b/conversion/conversion.go
index f5cff41..8fd8574 100644
--- a/conversion/conversion.go
+++ b/conversion/conversion.go
@@ -62,6 +62,7 @@ import (
 	"github.com/cloudspannerecosystem/harbourbridge/sources/sqlserver"
 	"github.com/cloudspannerecosystem/harbourbridge/spanner/ddl"
 	"github.com/cloudspannerecosystem/harbourbridge/spanner/writer"
+	"github.com/cloudspannerecosystem/harbourbridge/streaming"
 	"go.uber.org/zap"
 	adminpb "google.golang.org/genproto/googleapis/spanner/admin/database/v1"
 	"google.golang.org/grpc/metadata"
@@ -166,12 +167,65 @@ func getDbNameFromSQLConnectionStr(driver, sqlConnectionStr string) string {
 	return ""
 }
 
+func getInfoSchemaForShard(shardConnInfo profiles.DirectConnectionConfig, driver string, targetProfile profiles.TargetProfile) (common.InfoSchema, error) {
+	params := make(map[string]string)
+	params["host"] = shardConnInfo.Host
+	params["user"] = shardConnInfo.User
+	params["dbName"] = shardConnInfo.DbName
+	params["port"] = shardConnInfo.Port
+	params["password"] = shardConnInfo.Password
+	//while adding other sources, a switch-case will be added here on the basis of the driver input param passed.
+	//pased on the driver name, profiles.NewSourceProfileConnection<DBName> will need to be called to create
+	//the source profile information.
+	sourceProfileConnectionMySQL, err := profiles.NewSourceProfileConnectionMySQL(params)
+	if err != nil {
+		return nil, fmt.Errorf("cannot parse connection configuration for the primary shard")
+	}
+	sourceProfileConnection := profiles.SourceProfileConnection{Mysql: sourceProfileConnectionMySQL, Ty: profiles.SourceProfileConnectionTypeMySQL}
+	//create a source profile which contains the sourceProfileConnection object for the primary shard
+	//this is done because GetSQLConnectionStr() should not be aware of sharding
+	newSourceProfile := profiles.SourceProfile{Conn: sourceProfileConnection, Ty: profiles.SourceProfileTypeConnection}
+	newSourceProfile.Driver = driver
+	infoSchema, err := GetInfoSchema(newSourceProfile, targetProfile)
+	if err != nil {
+		return nil, err
+	}
+	return infoSchema, nil
+}
+
 func schemaFromDatabase(sourceProfile profiles.SourceProfile, targetProfile profiles.TargetProfile) (*internal.Conv, error) {
 	conv := internal.MakeConv()
 	conv.SpDialect = targetProfile.Conn.Sp.Dialect
-	infoSchema, err := GetInfoSchema(sourceProfile, targetProfile)
-	if err != nil {
-		return conv, err
+	//handle fetching schema differently for sharded migrations, we only connect to the primary shard to
+	//fetch the schema. We reuse the SourceProfileConnection object for this purpose.
+	var infoSchema common.InfoSchema
+	var err error
+	switch sourceProfile.Ty {
+	case profiles.SourceProfileTypeConfig:
+		//Find Primary Shard Name
+		if sourceProfile.Config.ConfigType == constants.BULK_MIGRATION {
+			schemaSource := sourceProfile.Config.ShardConfigurationBulk.SchemaSource
+			infoSchema, err = getInfoSchemaForShard(schemaSource, sourceProfile.Driver, targetProfile)
+			if err != nil {
+				return conv, err
+			}
+		} else if sourceProfile.Config.ConfigType == constants.DATAFLOW_MIGRATION {
+			schemaSource := sourceProfile.Config.ShardConfigurationDataflow.SchemaSource
+			infoSchema, err = getInfoSchemaForShard(schemaSource, sourceProfile.Driver, targetProfile)
+			if err != nil {
+				return conv, err
+			}
+		} else if sourceProfile.Config.ConfigType == constants.DMS_MIGRATION {
+			// TODO: Define the schema processing logic for DMS migrations here.
+			return conv, fmt.Errorf("dms based migrations are not implemented yet")
+		} else {
+			return conv, fmt.Errorf("unknown type of migration, please select one of bulk, dataflow or dms")
+		}
+	default:
+		infoSchema, err = GetInfoSchema(sourceProfile, targetProfile)
+		if err != nil {
+			return conv, err
+		}
 	}
 	return conv, common.ProcessSchema(conv, infoSchema, common.DefaultWorkers)
 }
@@ -201,27 +255,103 @@ func snapshotMigrationHandler(sourceProfile profiles.SourceProfile, config write
 }
 
 func dataFromDatabase(ctx context.Context, sourceProfile profiles.SourceProfile, targetProfile profiles.TargetProfile, config writer.BatchWriterConfig, conv *internal.Conv, client *sp.Client) (*writer.BatchWriter, error) {
-	infoSchema, err := GetInfoSchema(sourceProfile, targetProfile)
-	if err != nil {
-		return nil, err
-	}
-	var streamInfo map[string]interface{}
-	if sourceProfile.Conn.Streaming {
-		streamInfo, err = infoSchema.StartChangeDataCapture(ctx, conv)
+	//handle migrating data for sharded migrations differently
+	//sharded migrations are identified via the config= flag, if that flag is not present
+	//carry on with the existing code path in the else block
+	switch sourceProfile.Ty {
+	case profiles.SourceProfileTypeConfig:
+		////There are three cases to cover here, bulk migrations and sharded migrations (and later DMS)
+		//We provide an if-else based handling for each within the sharded code branch
+		//This will be determined via the configType, which can be "bulk", "dataflow" or "dms"
+		if sourceProfile.Config.ConfigType == constants.BULK_MIGRATION {
+			return dataFromDatabaseForBulkMigration(sourceProfile, targetProfile, config, conv, client)
+		} else if sourceProfile.Config.ConfigType == constants.DATAFLOW_MIGRATION {
+			return dataFromDatabaseForDataflowMigration(targetProfile, ctx, sourceProfile, conv)
+		} else if sourceProfile.Config.ConfigType == constants.DMS_MIGRATION {
+			return dataFromDatabaseForDMSMigration()
+		} else {
+			return nil, fmt.Errorf("configType should be one of 'bulk', 'dataflow' or 'dms'")
+		}
+	default:
+		infoSchema, err := GetInfoSchema(sourceProfile, targetProfile)
 		if err != nil {
 			return nil, err
 		}
-		bw, err := snapshotMigrationHandler(sourceProfile, config, conv, client, infoSchema)
+		var streamInfo map[string]interface{}
+		if sourceProfile.Conn.Streaming {
+			streamInfo, err = infoSchema.StartChangeDataCapture(ctx, conv)
+			if err != nil {
+				return nil, err
+			}
+			bw, err := snapshotMigrationHandler(sourceProfile, config, conv, client, infoSchema)
+			if err != nil {
+				return nil, err
+			}
+			err = infoSchema.StartStreamingMigration(ctx, client, conv, streamInfo)
+			if err != nil {
+				return nil, err
+			}
+			return bw, nil
+		}
+		return performSnapshotMigration(config, conv, client, infoSchema), nil
+	}
+}
+
+// TODO: Define the data processing logic for DMS migrations here.
+func dataFromDatabaseForDMSMigration() (*writer.BatchWriter, error) {
+	return nil, fmt.Errorf("dms configType is not implemented yet, please use one of 'bulk' or 'dataflow'")
+}
+
+//1. Create batch for each physical shard
+//2. Create streaming cfg from the config source type.
+//3. Verify the CFG and update it with HB defaults
+//4. Launch the stream for the physical shard
+//5. Perform streaming migration via dataflow
+func dataFromDatabaseForDataflowMigration(targetProfile profiles.TargetProfile, ctx context.Context, sourceProfile profiles.SourceProfile, conv *internal.Conv) (*writer.BatchWriter, error) {
+	asyncProcessShards := func(p *profiles.DataShard, mutex *sync.Mutex) common.TaskResult[*profiles.DataShard] {
+
+		streamingCfg := streaming.CreateStreamingConfig(*p)
+
+		err := streaming.VerifyAndUpdateCfg(&streamingCfg, targetProfile.Conn.Sp.Dbname)
+		if err != nil {
+			err = fmt.Errorf("failed to process shard: %s, there seems to be an error in the sharding configuration, error: %v", p.DataShardId, err)
+			return common.TaskResult[*profiles.DataShard]{Result: p, Err: err}
+		}
+		fmt.Printf("Initiating migration for shard: %v\n", p.DataShardId)
+
+		err = streaming.LaunchStream(ctx, sourceProfile.Driver, p.LogicalShards, targetProfile.Conn.Sp.Project, streamingCfg.DatastreamCfg)
 		if err != nil {
-			return nil, err
+			return common.TaskResult[*profiles.DataShard]{Result: p, Err: err}
 		}
-		err = infoSchema.StartStreamingMigration(ctx, client, conv, streamInfo)
+
+		err = streaming.StartDataflow(ctx, targetProfile, streamingCfg, conv)
+		return common.TaskResult[*profiles.DataShard]{Result: p, Err: err}
+	}
+	_, err := common.RunParallelTasks(sourceProfile.Config.ShardConfigurationDataflow.DataShards, 5, asyncProcessShards, true)
+	if err != nil {
+		return nil, fmt.Errorf("unable to start minimal downtime migrations: %v", err)
+	}
+	return &writer.BatchWriter{}, nil
+}
+
+//1. Migrate the data from the data shards, the schema shard needs to be specified here again.
+//2. Create a connection profile object for it
+//3. Perform a snapshot migration for the shard
+//4. Once all shard migrations are complete, return the batch writer object
+func dataFromDatabaseForBulkMigration(sourceProfile profiles.SourceProfile, targetProfile profiles.TargetProfile, config writer.BatchWriterConfig, conv *internal.Conv, client *sp.Client) (*writer.BatchWriter, error) {
+	var bw *writer.BatchWriter
+	for _, dataShard := range sourceProfile.Config.ShardConfigurationBulk.DataShards {
+
+		fmt.Printf("Initiating migration for shard: %v\n", dataShard.DbName)
+		infoSchema, err := getInfoSchemaForShard(dataShard, sourceProfile.Driver, targetProfile)
 		if err != nil {
 			return nil, err
 		}
-		return bw, nil
+
+		bw = performSnapshotMigration(config, conv, client, infoSchema)
 	}
-	return performSnapshotMigration(config, conv, client, infoSchema), nil
+
+	return bw, nil
 }
 
 func getDynamoDBClientConfig() (*aws.Config, error) {
diff --git a/profiles/source_profile.go b/profiles/source_profile.go
index 32a4a29..9530639 100644
--- a/profiles/source_profile.go
+++ b/profiles/source_profile.go
@@ -15,7 +15,9 @@
 package profiles
 
 import (
+	"encoding/json"
 	"fmt"
+	"io/ioutil"
 	"os"
 	"strconv"
 	"strings"
@@ -324,8 +326,8 @@ func NewSourceProfileConnectionOracle(params map[string]string) (SourceProfileCo
 	host, hostOk := params["host"]
 	user, userOk := params["user"]
 	db, dbOk := params["dbName"]
-	port, _ := params["port"]
-	pwd, _ := params["password"]
+	port := params["port"]
+	pwd := params["password"]
 
 	streamingConfig, cfgOk := params["streamingCfg"]
 	if cfgOk && streamingConfig == "" {
@@ -429,12 +431,80 @@ func NewSourceProfileConnection(source string, params map[string]string) (Source
 	return conn, nil
 }
 
+type DirectConnectionConfig struct {
+	DataShardId string `json:"dataShardId"`
+	Host        string `json:"host"`
+	User        string `json:"user"`
+	Password    string `json:"password"`
+	Port        string `json:"port"`
+	DbName      string `json:"dbName"`
+}
+
+type DatastreamConnProfile struct {
+	Name     string `json:"name"`
+	Location string `json:"location"`
+}
+
+type DataflowConfig struct {
+	Location      string `json:"location"`
+	Network       string `json:"network"`
+	Subnetwork    string `json:"subnetwork"`
+	HostProjectId string `json:"hostProjectId"`
+}
+
+type DataShard struct {
+	DataShardId          string                `json:"dataShardId"`
+	SrcConnectionProfile DatastreamConnProfile `json:"srcConnectionProfile"`
+	DstConnectionProfile DatastreamConnProfile `json:"dstConnectionProfile"`
+	DataflowConfig       DataflowConfig        `json:"dataflowConfig"`
+	TmpDir               string                `json:"tmpDir"`
+	StreamLocation       string                `json:"streamLocation"`
+	LogicalShards        []LogicalShard        `json:"databases"`
+}
+
+type LogicalShard struct {
+	DbName         string `json:"dbName"`
+	LogicalShardId string `json:"databaseId"`
+	RefDataShardId string `json:"refDataShardId"`
+}
+
+type ShardConfigurationDataflow struct {
+	SchemaSource DirectConnectionConfig `json:"schemaSource"`
+	DataShards   []*DataShard           `json:"dataShards"`
+}
+
+type ShardConfigurationBulk struct {
+	SchemaSource DirectConnectionConfig   `json:"schemaSource"`
+	DataShards   []DirectConnectionConfig `json:"dataShards"`
+}
+
+// TODO: Define the sharding structure for DMS migrations here.
+type ShardConfigurationDMS struct {
+}
+
 type SourceProfileConfig struct {
-	path string
+	ConfigType                 string                     `json:"configType"`
+	ShardConfigurationBulk     ShardConfigurationBulk     `json:"shardConfigurationBulk"`
+	ShardConfigurationDataflow ShardConfigurationDataflow `json:"shardConfigurationDataflow"`
+	ShardConfigurationDMS      ShardConfigurationDMS      `json:"shardConfigurationDMS"`
 }
 
-func NewSourceProfileConfig(path string) SourceProfileConfig {
-	return SourceProfileConfig{path: path}
+func NewSourceProfileConfig(source string, path string) (SourceProfileConfig, error) {
+	//given the source, the fact that this 'config=', determine the appropiate object to marshal into
+	switch source {
+	case constants.MYSQL:
+		//load the JSON configuration into file
+		configFile, err := ioutil.ReadFile(path)
+		if err != nil {
+			return SourceProfileConfig{}, fmt.Errorf("cannot read config file due to: %v", err)
+		}
+		sourceProfileConfig := SourceProfileConfig{}
+		//unmarshal the JSON into object
+		err = json.Unmarshal(configFile, &sourceProfileConfig)
+		return sourceProfileConfig, err
+	default:
+		return SourceProfileConfig{}, fmt.Errorf("sharded migrations are currrently only supported for MySQL databases")
+	}
 }
 
 type SourceProfileCsv struct {
@@ -510,7 +580,14 @@ func (src SourceProfile) ToLegacyDriver(source string) (string, error) {
 			}
 		}
 	case SourceProfileTypeConfig:
-		return "", fmt.Errorf("specifying source-profile using config not implemented")
+		{
+			switch strings.ToLower(source) {
+			case constants.MYSQL:
+				return constants.MYSQL, nil
+			default:
+				return "", fmt.Errorf("specifying source-profile using config for non-mysql databases not implemented")
+			}
+		}
 	case SourceProfileTypeCsv:
 		return constants.CSV, nil
 	default:
@@ -553,8 +630,8 @@ func NewSourceProfile(s string, source string) (SourceProfile, error) {
 		// File is not passed in from stdin or specified using "file" flag.
 		return SourceProfile{Ty: SourceProfileTypeFile}, fmt.Errorf("file not specified, but format set to %v", format)
 	} else if file, ok := params["config"]; ok {
-		config := NewSourceProfileConfig(file)
-		return SourceProfile{Ty: SourceProfileTypeConfig, Config: config}, fmt.Errorf("source-profile type config not yet implemented")
+		config, err := NewSourceProfileConfig(strings.ToLower(source), file)
+		return SourceProfile{Ty: SourceProfileTypeConfig, Config: config}, err
 	} else {
 		// Assume connection profile type connection by default, since
 		// connection parameters could be specified as part of environment
diff --git a/sources/mysql/infoschema.go b/sources/mysql/infoschema.go
index f470e47..6ad7656 100644
--- a/sources/mysql/infoschema.go
+++ b/sources/mysql/infoschema.go
@@ -67,7 +67,7 @@ func (isi InfoSchemaImpl) GetRowsFromTable(conv *internal.Conv, tableId string)
 	// Ideally we would pass schema/name as a query parameter,
 	// but MySQL doesn't support this. So we quote it instead.
 	colNameList := buildColNameList(srcSchema, srcCols)
-	q := fmt.Sprintf("SELECT %s FROM `%s`.`%s`;", colNameList, srcSchema.Schema, srcSchema.Name)
+	q := fmt.Sprintf("SELECT %s FROM `%s`.`%s`;", colNameList, isi.DbName, srcSchema.Name)
 	rows, err := isi.Db.Query(q)
 	return rows, err
 }
@@ -376,7 +376,7 @@ func (isi InfoSchemaImpl) StartChangeDataCapture(ctx context.Context, conv *inte
 func (isi InfoSchemaImpl) StartStreamingMigration(ctx context.Context, client *sp.Client, conv *internal.Conv, streamingInfo map[string]interface{}) error {
 	streamingCfg, _ := streamingInfo["streamingCfg"].(streaming.StreamingCfg)
 
-	err := streaming.StartDataflow(ctx, isi.SourceProfile, isi.TargetProfile, streamingCfg, conv)
+	err := streaming.StartDataflow(ctx, isi.TargetProfile, streamingCfg, conv)
 	if err != nil {
 		err = fmt.Errorf("error starting dataflow: %v", err)
 		return err
diff --git a/sources/oracle/infoschema.go b/sources/oracle/infoschema.go
index e9a6993..f317b33 100644
--- a/sources/oracle/infoschema.go
+++ b/sources/oracle/infoschema.go
@@ -439,7 +439,7 @@ func (isi InfoSchemaImpl) StartChangeDataCapture(ctx context.Context, conv *inte
 // performing a streaming migration.
 func (isi InfoSchemaImpl) StartStreamingMigration(ctx context.Context, client *sp.Client, conv *internal.Conv, streamingInfo map[string]interface{}) error {
 	streamingCfg, _ := streamingInfo["streamingCfg"].(streaming.StreamingCfg)
-	err := streaming.StartDataflow(ctx, isi.SourceProfile, isi.TargetProfile, streamingCfg, conv)
+	err := streaming.StartDataflow(ctx, isi.TargetProfile, streamingCfg, conv)
 	if err != nil {
 		return err
 	}
diff --git a/sources/postgres/infoschema.go b/sources/postgres/infoschema.go
index 36fc3bf..6976dd1 100644
--- a/sources/postgres/infoschema.go
+++ b/sources/postgres/infoschema.go
@@ -75,7 +75,7 @@ func (isi InfoSchemaImpl) StartChangeDataCapture(ctx context.Context, conv *inte
 func (isi InfoSchemaImpl) StartStreamingMigration(ctx context.Context, client *sp.Client, conv *internal.Conv, streamingInfo map[string]interface{}) error {
 	streamingCfg, _ := streamingInfo["streamingCfg"].(streaming.StreamingCfg)
 
-	err := streaming.StartDataflow(ctx, isi.SourceProfile, isi.TargetProfile, streamingCfg, conv)
+	err := streaming.StartDataflow(ctx, isi.TargetProfile, streamingCfg, conv)
 	if err != nil {
 		err = fmt.Errorf("error starting dataflow: %v", err)
 		return err
diff --git a/streaming/streaming.go b/streaming/streaming.go
index 15d3a54..31ea464 100644
--- a/streaming/streaming.go
+++ b/streaming/streaming.go
@@ -160,12 +160,17 @@ func ReadStreamingConfig(file, dbName string) (StreamingCfg, error) {
 	return streamingCfg, nil
 }
 
-func getMysqlSourceStreamConfig(dbName string) *datastreampb.SourceConfig_MysqlSourceConfig {
-	mydb := &datastreampb.MysqlDatabase{
-		Database: dbName,
+func getMysqlSourceStreamConfig(dbList []profiles.LogicalShard) *datastreampb.SourceConfig_MysqlSourceConfig {
+	includeDbList := []*datastreampb.MysqlDatabase{}
+	for _, db := range dbList {
+		//create include db object
+		includeDb := &datastreampb.MysqlDatabase{
+			Database: db.DbName,
+		}
+		includeDbList = append(includeDbList, includeDb)
 	}
 	mysqlSrcCfg := &datastreampb.MysqlSourceConfig{
-		IncludeObjects: &datastreampb.MysqlRdbms{MysqlDatabases: []*datastreampb.MysqlDatabase{mydb}},
+		IncludeObjects: &datastreampb.MysqlRdbms{MysqlDatabases: includeDbList},
 	}
 	return &datastreampb.SourceConfig_MysqlSourceConfig{MysqlSourceConfig: mysqlSrcCfg}
 }
@@ -204,14 +209,14 @@ func getPostgreSQLSourceStreamConfig(properties string) (*datastreampb.SourceCon
 	return &datastreampb.SourceConfig_PostgresqlSourceConfig{PostgresqlSourceConfig: postgresSrcCfg}, nil
 }
 
-func getSourceStreamConfig(srcCfg *datastreampb.SourceConfig, sourceProfile profiles.SourceProfile, datastreamCfg DatastreamCfg) error {
-	switch sourceProfile.Driver {
+func getSourceStreamConfig(srcCfg *datastreampb.SourceConfig, driver string, dbList []profiles.LogicalShard, datastreamCfg DatastreamCfg) error {
+	switch driver {
 	case constants.MYSQL:
-		srcCfg.SourceStreamConfig = getMysqlSourceStreamConfig(sourceProfile.Conn.Mysql.Db)
+		srcCfg.SourceStreamConfig = getMysqlSourceStreamConfig(dbList)
 		return nil
 	case constants.ORACLE:
 		// For Oracle, the User name denotes the name of the schema while the dbName parameter has the SID.
-		srcCfg.SourceStreamConfig = getOracleSourceStreamConfig(sourceProfile.Conn.Oracle.User)
+		srcCfg.SourceStreamConfig = getOracleSourceStreamConfig(dbList[0].DbName)
 		return nil
 	case constants.POSTGRES:
 		sourceStreamConfig, err := getPostgreSQLSourceStreamConfig(datastreamCfg.Properties)
@@ -225,7 +230,7 @@ func getSourceStreamConfig(srcCfg *datastreampb.SourceConfig, sourceProfile prof
 }
 
 // LaunchStream populates the parameters from the streaming config and triggers a stream on Cloud Datastream.
-func LaunchStream(ctx context.Context, sourceProfile profiles.SourceProfile, projectID string, datastreamCfg DatastreamCfg) error {
+func LaunchStream(ctx context.Context, driver string, dbList []profiles.LogicalShard, projectID string, datastreamCfg DatastreamCfg) error {
 	fmt.Println("Launching stream ", fmt.Sprintf("projects/%s/locations/%s", projectID, datastreamCfg.StreamLocation))
 	dsClient, err := datastream.NewClient(ctx)
 	if err != nil {
@@ -241,7 +246,7 @@ func LaunchStream(ctx context.Context, sourceProfile profiles.SourceProfile, pro
 	srcCfg := &datastreampb.SourceConfig{
 		SourceConnectionProfile: fmt.Sprintf("projects/%s/locations/%s/connectionProfiles/%s", projectID, datastreamCfg.SourceConnectionConfig.Location, datastreamCfg.SourceConnectionConfig.Name),
 	}
-	err = getSourceStreamConfig(srcCfg, sourceProfile, datastreamCfg)
+	err = getSourceStreamConfig(srcCfg, driver, dbList, datastreamCfg)
 	if err != nil {
 		return fmt.Errorf("could not get source stream config: %v", err)
 	}
@@ -447,20 +452,51 @@ func getStreamingConfig(sourceProfile profiles.SourceProfile, targetProfile prof
 	}
 }
 
+func CreateStreamingConfig(pl profiles.DataShard) StreamingCfg {
+	//create dataflowcfg from pl receiver object
+	inputDataflowConfig := pl.DataflowConfig
+	dataflowCfg := DataflowCfg{Location: inputDataflowConfig.Location,
+		Network:       inputDataflowConfig.Network,
+		HostProjectId: inputDataflowConfig.HostProjectId,
+		Subnetwork:    inputDataflowConfig.Subnetwork}
+	//create src and dst datastream from pl receiver object
+	datastreamCfg := DatastreamCfg{StreamLocation: pl.StreamLocation} 
+	//set src connection profile
+	inputSrcConnProfile := pl.SrcConnectionProfile
+	srcConnCfg := SrcConnCfg{Location: inputSrcConnProfile.Location, Name: inputSrcConnProfile.Name}
+	datastreamCfg.SourceConnectionConfig = srcConnCfg
+	//set dst connection profile
+	inputDstConnProfile := pl.DstConnectionProfile
+	dstConnCfg := DstConnCfg{Name: inputDstConnProfile.Name, Location: inputDstConnProfile.Location}
+	datastreamCfg.DestinationConnectionConfig = dstConnCfg
+	//create the streamingCfg object
+	streamingCfg := StreamingCfg{DataflowCfg: dataflowCfg, DatastreamCfg: datastreamCfg, TmpDir: pl.TmpDir}
+	return streamingCfg
+}
+
 func StartDatastream(ctx context.Context, sourceProfile profiles.SourceProfile, targetProfile profiles.TargetProfile) (StreamingCfg, error) {
 	streamingCfg, err := getStreamingConfig(sourceProfile, targetProfile)
 	if err != nil {
 		return streamingCfg, fmt.Errorf("error reading streaming config: %v", err)
 	}
-
-	err = LaunchStream(ctx, sourceProfile, targetProfile.Conn.Sp.Project, streamingCfg.DatastreamCfg)
+	driver := sourceProfile.Driver
+	var dbList []profiles.LogicalShard
+	switch driver {
+	case constants.MYSQL:
+		dbList = append(dbList, profiles.LogicalShard{DbName: sourceProfile.Conn.Mysql.Db})
+	case constants.ORACLE:
+		dbList = append(dbList, profiles.LogicalShard{DbName: sourceProfile.Conn.Oracle.User})
+	case constants.POSTGRES:
+		dbList = append(dbList, profiles.LogicalShard{DbName: streamingCfg.DatastreamCfg.Properties})
+	}
+	err = LaunchStream(ctx, driver, dbList, targetProfile.Conn.Sp.Project, streamingCfg.DatastreamCfg)
 	if err != nil {
 		return streamingCfg, fmt.Errorf("error launching stream: %v", err)
 	}
 	return streamingCfg, nil
 }
 
-func StartDataflow(ctx context.Context, sourceProfile profiles.SourceProfile, targetProfile profiles.TargetProfile, streamingCfg StreamingCfg, conv *internal.Conv) error {
+func StartDataflow(ctx context.Context, targetProfile profiles.TargetProfile, streamingCfg StreamingCfg, conv *internal.Conv) error {
 
 	convJSON, err := json.MarshalIndent(conv, "", " ")
 	if err != nil {


Test patch:
diff --git a/profiles/source_profile_test.go b/profiles/source_profile_test.go
index a06dc39..8aae609 100644
--- a/profiles/source_profile_test.go
+++ b/profiles/source_profile_test.go
@@ -15,6 +15,7 @@
 package profiles
 
 import (
+	"path/filepath"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
@@ -68,6 +69,73 @@ func TestNewSourceProfileFile(t *testing.T) {
 	}
 }
 
+func TestNewSourceProfileConfigFile(t *testing.T) {
+	type validationFn func(SourceProfileConfig) 
+	testCases := []struct {
+		name          string
+		source        string
+		path          string
+		errorExpected bool
+		validationFn validationFn 
+	}{
+		{
+			name:          "bulk config for mysql",
+			source:        "mysql",
+			path:          filepath.Join("..", "test_data", "mysql_shard_bulk.cfg"),
+			errorExpected: false,
+			validationFn: func(spc SourceProfileConfig) {
+				assert.NotNil(t, spc.ShardConfigurationBulk)
+				assert.NotNil(t, spc.ShardConfigurationBulk.SchemaSource)
+				assert.NotEmpty(t, spc.ShardConfigurationBulk.SchemaSource.DbName)
+				assert.NotEmpty(t, spc.ShardConfigurationBulk.SchemaSource.Host)
+				assert.NotEmpty(t, spc.ShardConfigurationBulk.SchemaSource.Password)
+				assert.NotEmpty(t, spc.ShardConfigurationBulk.SchemaSource.Port)
+				assert.NotEmpty(t, spc.ShardConfigurationBulk.SchemaSource.User)
+				assert.NotNil(t, spc.ShardConfigurationBulk.DataShards)
+				assert.NotEmpty(t, spc.ShardConfigurationBulk.DataShards[0].DataShardId)
+				assert.NotEmpty(t, spc.ShardConfigurationBulk.DataShards[0].DbName)
+				assert.NotEmpty(t, spc.ShardConfigurationBulk.DataShards[0].User)
+				assert.NotEmpty(t, spc.ShardConfigurationBulk.DataShards[0].Password)
+				assert.NotEmpty(t, spc.ShardConfigurationBulk.DataShards[0].Port)
+			},
+		},
+		{
+			name:          "streaming config for mysql",
+			source:        "mysql",
+			path:          filepath.Join("..", "test_data", "mysql_shard_streaming.cfg"),
+			errorExpected: false,
+			validationFn: func(spc SourceProfileConfig) {
+				assert.NotNil(t, spc.ShardConfigurationDataflow)
+				assert.NotNil(t, spc.ShardConfigurationDataflow.SchemaSource)
+				assert.NotEmpty(t, spc.ShardConfigurationDataflow.SchemaSource.DbName)
+				assert.NotEmpty(t, spc.ShardConfigurationDataflow.SchemaSource.Host)
+				assert.NotEmpty(t, spc.ShardConfigurationDataflow.SchemaSource.Password)
+				assert.NotEmpty(t, spc.ShardConfigurationDataflow.SchemaSource.Port)
+				assert.NotEmpty(t, spc.ShardConfigurationDataflow.SchemaSource.User)
+				assert.NotNil(t, spc.ShardConfigurationDataflow.DataShards)
+				assert.NotEmpty(t, spc.ShardConfigurationDataflow.DataShards[0].DataShardId)
+				assert.NotEmpty(t, spc.ShardConfigurationDataflow.DataShards[0].TmpDir)
+				assert.NotEmpty(t, spc.ShardConfigurationDataflow.DataShards[0].StreamLocation)
+				assert.NotEmpty(t, spc.ShardConfigurationDataflow.DataShards[0].DataflowConfig)
+				assert.NotEmpty(t, spc.ShardConfigurationDataflow.DataShards[0].DstConnectionProfile)
+				assert.NotEmpty(t, spc.ShardConfigurationDataflow.DataShards[0].SrcConnectionProfile)
+			},
+		},
+		{
+			name:          "config for non-mysql",
+			source:        "postgres",
+			path:          "",
+			errorExpected: true,
+			validationFn: func(spc SourceProfileConfig) {},
+		},
+	}
+	for _, tc := range testCases {
+		sourceProfileConfig, err := NewSourceProfileConfig(tc.source, tc.path)
+		assert.Equal(t, tc.errorExpected, err != nil)
+		tc.validationFn(sourceProfileConfig)
+	}
+}
+
 func TestNewSourceProfileConnectionSQL(t *testing.T) {
 	// Avoid getting/settinng env variables in the unit tests.
 	testCases := []struct {
diff --git a/test_data/mysql_shard_bulk.cfg b/test_data/mysql_shard_bulk.cfg
new file mode 100644
index 0000000..c502544
--- /dev/null
+++ b/test_data/mysql_shard_bulk.cfg
@@ -0,0 +1,22 @@
+{
+    "configType":"bulk",
+    "shardConfigurationBulk":{
+       "schemaSource":{
+          "host":"test",
+          "user":"test",
+          "password":"test",
+          "port":"3306",
+          "dbName":"test"
+       },
+       "dataShards":[
+          {
+            "dataShardId":"X1",
+             "host":"test",
+             "user":"test",
+             "password":"test",
+             "port":"3306",
+             "dbName":"test"
+          }
+       ]
+    }
+ }
\ No newline at end of file
diff --git a/test_data/mysql_shard_streaming.cfg b/test_data/mysql_shard_streaming.cfg
new file mode 100644
index 0000000..60fa6e8
--- /dev/null
+++ b/test_data/mysql_shard_streaming.cfg
@@ -0,0 +1,47 @@
+{
+   "configType":"dataflow",
+   "shardConfigurationDataflow":{
+      "schemaSource":{
+         "host":"localhost",
+         "user":"root",
+         "password":"test",
+         "port":"3306",
+         "dbName":"shard3"
+      },
+      "dataShards":[
+         {
+            "dataShardId":"physical1",
+            "tmpDir":"gs://hb-test/",
+            "streamLocation":"asia-east2",
+            "srcConnectionProfile":{
+               "name":"test1",
+               "location":"asia-east2"
+            },
+            "dstConnectionProfile":{
+               "name":"test2",
+               "location":"asia-east2"
+            },
+            "dataflowConfig":{
+               "location":"asia-east2",
+               "network":"",
+               "subnetwork":"",
+               "hostProjectId":""
+            },
+            "databases":[
+               {
+                  "dbName":"shard3",
+                  "databaseId":"logical1",
+                  "tableInclude":[
+                     "numbers"
+                  ],
+                  "tableExclude":[
+                     "numbers2",
+                     "numbers3"
+                  ],
+                  "refDataShardId":"physical1"
+               }
+            ]
+         }
+      ]
+   }
+}
\ No newline at end of file