Repository: cloudspannerecosystem/harbourbridge
Pull request #323 (2022-06-23 15:18:16)

Adding interface support for starting streaming migrations
Things added/changed
1. Common functions for all databases in infoschema interface for starting streaming migration
   * StartChangeDataCapture -> This can be implemented to start capturing live changes on the database before bulk migration.
   * StartStreamingMigration -> This can be implemented to start the processing of captured changes after finishing bulk migration
2. Changes to InfoSchemaImpl object for storing information required for starting streaming migrations.
3. Moved code for automatic triggering of Datastream and Dataflow jobs[#286] to the functions above described for MySQL, Oracle.

- [x] Tests pass


Patch:
diff --git a/cmd/common.go b/cmd/common.go
deleted file mode 100644
index f359156..0000000
--- a/cmd/common.go
+++ /dev/null
@@ -1,73 +0,0 @@
-package cmd
-
-import (
-	"context"
-	"fmt"
-	"time"
-
-	sp "cloud.google.com/go/spanner"
-	"github.com/cloudspannerecosystem/harbourbridge/common/utils"
-	"github.com/cloudspannerecosystem/harbourbridge/conversion"
-	"github.com/cloudspannerecosystem/harbourbridge/internal"
-	"github.com/cloudspannerecosystem/harbourbridge/profiles"
-	"github.com/cloudspannerecosystem/harbourbridge/spanner/writer"
-	"github.com/cloudspannerecosystem/harbourbridge/streaming"
-)
-
-func startDatastream(ctx context.Context, sourceProfile profiles.SourceProfile, targetProfile profiles.TargetProfile) (streaming.StreamingCfg, error) {
-	streamingCfg, err := streaming.ReadStreamingConfig(sourceProfile.Conn.Mysql.StreamingConfig, targetProfile.Conn.Sp.Dbname)
-	if err != nil {
-		return streamingCfg, fmt.Errorf("error reading streaming config: %v", err)
-	}
-
-	err = streaming.LaunchStream(ctx, sourceProfile, targetProfile.Conn.Sp.Project, streamingCfg.DatastreamCfg)
-	if err != nil {
-		return streamingCfg, fmt.Errorf("error launching stream: %v", err)
-	}
-	return streamingCfg, nil
-}
-
-func performSnapshotMigration(ctx context.Context, sourceProfile profiles.SourceProfile, targetProfile profiles.TargetProfile, ioHelper utils.IOStreams, client *sp.Client, conv *internal.Conv, writeLimit int64, dbURI string) (*writer.BatchWriter, error) {
-	dataCoversionStartTime := time.Now()
-	bw, err := conversion.DataConv(ctx, sourceProfile, targetProfile, &ioHelper, client, conv, true, writeLimit)
-	if err != nil {
-		return bw, fmt.Errorf("can't finish data conversion for db %s: %v", dbURI, err)
-	}
-	dataCoversionEndTime := time.Now()
-	dataCoversionDuration := dataCoversionEndTime.Sub(dataCoversionStartTime)
-	conv.Audit.DataConversionDuration = dataCoversionDuration
-	return bw, nil
-}
-
-func startDataflow(ctx context.Context, sourceProfile profiles.SourceProfile, targetProfile profiles.TargetProfile, streamingCfg streaming.StreamingCfg) error {
-	err := streaming.LaunchDataflowJob(ctx, targetProfile, streamingCfg.DatastreamCfg, streamingCfg.DataflowCfg)
-	if err != nil {
-		return fmt.Errorf("error launching dataflow: %v", err)
-	}
-	return nil
-}
-
-func migrateData(ctx context.Context, sourceProfile profiles.SourceProfile, targetProfile profiles.TargetProfile, ioHelper utils.IOStreams, client *sp.Client, conv *internal.Conv, writeLimit int64, dbURI string) (*writer.BatchWriter, error) {
-	streamingCfg := streaming.StreamingCfg{}
-	var err error
-	if sourceProfile.Ty == profiles.SourceProfileTypeConnection && sourceProfile.Conn.Streaming {
-		streamingCfg, err = startDatastream(ctx, sourceProfile, targetProfile)
-		if err != nil {
-			err = fmt.Errorf("error starting datastream: %v", err)
-			return nil, err
-		}
-	}
-	bw, err := performSnapshotMigration(ctx, sourceProfile, targetProfile, ioHelper, client, conv, writeLimit, dbURI)
-	if err != nil {
-		err = fmt.Errorf("can't do snapshot migration: %v", err)
-		return nil, err
-	}
-	if sourceProfile.Ty == profiles.SourceProfileTypeConnection && sourceProfile.Conn.Streaming {
-		err = startDataflow(ctx, sourceProfile, targetProfile, streamingCfg)
-		if err != nil {
-			err = fmt.Errorf("error starting dataflow: %v", err)
-			return nil, err
-		}
-	}
-	return bw, nil
-}
diff --git a/cmd/data.go b/cmd/data.go
index 4d31caa..b602035 100644
--- a/cmd/data.go
+++ b/cmd/data.go
@@ -165,11 +165,16 @@ func (cmd *DataCmd) Execute(ctx context.Context, f *flag.FlagSet, _ ...interface
 		}
 	}
 
-	bw, err := migrateData(ctx, sourceProfile, targetProfile, ioHelper, client, conv, cmd.writeLimit, dbURI)
+	dataCoversionStartTime := time.Now()
+	bw, err := conversion.DataConv(ctx, sourceProfile, targetProfile, &ioHelper, client, conv, true, cmd.writeLimit)
 	if err != nil {
-		err = fmt.Errorf("can't do data migration: %v", err)
+		err = fmt.Errorf("can't finish data migration: %v", err)
 		return subcommands.ExitFailure
 	}
+	dataCoversionEndTime := time.Now()
+	dataCoversionDuration := dataCoversionEndTime.Sub(dataCoversionStartTime)
+	conv.Audit.DataConversionDuration = dataCoversionDuration
+
 	if !cmd.skipForeignKeys {
 		if err = conversion.UpdateDDLForeignKeys(ctx, adminClient, dbURI, conv, ioHelper.Out); err != nil {
 			err = fmt.Errorf("can't perform update schema on db %s with foreign keys: %v", dbURI, err)
diff --git a/cmd/schema_and_data.go b/cmd/schema_and_data.go
index 4cb8ea2..afacfb9 100644
--- a/cmd/schema_and_data.go
+++ b/cmd/schema_and_data.go
@@ -165,11 +165,16 @@ func (cmd *SchemaAndDataCmd) Execute(ctx context.Context, f *flag.FlagSet, _ ...
 		return subcommands.ExitFailure
 	}
 
-	bw, err := migrateData(ctx, sourceProfile, targetProfile, ioHelper, client, conv, cmd.writeLimit, dbURI)
+	dataCoversionStartTime := time.Now()
+	bw, err := conversion.DataConv(ctx, sourceProfile, targetProfile, &ioHelper, client, conv, true, cmd.writeLimit)
 	if err != nil {
-		err = fmt.Errorf("can't do data migration: %v", err)
+		err = fmt.Errorf("can't finish data migration: %v", err)
 		return subcommands.ExitFailure
 	}
+	dataCoversionEndTime := time.Now()
+	dataCoversionDuration := dataCoversionEndTime.Sub(dataCoversionStartTime)
+	conv.Audit.DataConversionDuration = dataCoversionDuration
+
 	if !cmd.skipForeignKeys {
 		if err = conversion.UpdateDDLForeignKeys(ctx, adminClient, dbURI, conv, ioHelper.Out); err != nil {
 			err = fmt.Errorf("can't perform update schema on db %s with foreign keys: %v", dbURI, err)
diff --git a/conversion/conversion.go b/conversion/conversion.go
index 81e7ca5..1dbd0ae 100644
--- a/conversion/conversion.go
+++ b/conversion/conversion.go
@@ -43,6 +43,7 @@ import (
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/aws/session"
 	dydb "github.com/aws/aws-sdk-go/service/dynamodb"
+	"github.com/aws/aws-sdk-go/service/dynamodbstreams"
 	adminpb "google.golang.org/genproto/googleapis/spanner/admin/database/v1"
 	"google.golang.org/grpc/metadata"
 	"google.golang.org/protobuf/proto"
@@ -97,7 +98,7 @@ func DataConv(ctx context.Context, sourceProfile profiles.SourceProfile, targetP
 	}
 	switch sourceProfile.Driver {
 	case constants.POSTGRES, constants.MYSQL, constants.DYNAMODB, constants.SQLSERVER, constants.ORACLE:
-		return dataFromDatabase(sourceProfile, config, client, conv)
+		return dataFromDatabase(ctx, sourceProfile, targetProfile, config, conv, client)
 	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")
@@ -166,18 +167,14 @@ func getDbNameFromSQLConnectionStr(driver, sqlConnectionStr string) string {
 func schemaFromDatabase(sourceProfile profiles.SourceProfile, targetProfile profiles.TargetProfile) (*internal.Conv, error) {
 	conv := internal.MakeConv()
 	conv.TargetDb = targetProfile.TargetDb
-	infoSchema, err := GetInfoSchema(sourceProfile)
+	infoSchema, err := GetInfoSchema(sourceProfile, targetProfile)
 	if err != nil {
 		return conv, err
 	}
 	return conv, common.ProcessSchema(conv, infoSchema)
 }
 
-func dataFromDatabase(sourceProfile profiles.SourceProfile, config writer.BatchWriterConfig, client *sp.Client, conv *internal.Conv) (*writer.BatchWriter, error) {
-	infoSchema, err := GetInfoSchema(sourceProfile)
-	if err != nil {
-		return nil, err
-	}
+func performSnapshotMigration(config writer.BatchWriterConfig, conv *internal.Conv, client *sp.Client, infoSchema common.InfoSchema) (*writer.BatchWriter, error) {
 	common.SetRowStats(conv, infoSchema)
 	totalRows := conv.Rows()
 	p := internal.NewProgress(totalRows, "Writing data to Spanner", internal.Verbose(), false)
@@ -187,6 +184,31 @@ func dataFromDatabase(sourceProfile profiles.SourceProfile, config writer.BatchW
 	return batchWriter, nil
 }
 
+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)
+		if err != nil {
+			return nil, err
+		}
+	}
+	bw, err := performSnapshotMigration(config, conv, client, infoSchema)
+	if err != nil {
+		return nil, err
+	}
+	if sourceProfile.Conn.Streaming {
+		err = infoSchema.StartStreamingMigration(ctx, client, conv, streamInfo)
+		if err != nil {
+			return nil, err
+		}
+	}
+	return bw, nil
+}
+
 func getDynamoDBClientConfig() (*aws.Config, error) {
 	cfg := aws.Config{}
 	endpointOverride := os.Getenv("DYNAMODB_ENDPOINT_OVERRIDE")
@@ -760,7 +782,7 @@ func ProcessDump(driver string, conv *internal.Conv, r *internal.Reader) error {
 	}
 }
 
-func GetInfoSchema(sourceProfile profiles.SourceProfile) (common.InfoSchema, error) {
+func GetInfoSchema(sourceProfile profiles.SourceProfile, targetProfile profiles.TargetProfile) (common.InfoSchema, error) {
 	connectionConfig, err := connectionConfig(sourceProfile)
 	if err != nil {
 		return nil, err
@@ -773,7 +795,12 @@ func GetInfoSchema(sourceProfile profiles.SourceProfile) (common.InfoSchema, err
 		if err != nil {
 			return nil, err
 		}
-		return mysql.InfoSchemaImpl{DbName: dbName, Db: db}, nil
+		return mysql.InfoSchemaImpl{
+			DbName:        dbName,
+			Db:            db,
+			SourceProfile: sourceProfile,
+			TargetProfile: targetProfile,
+		}, nil
 	case constants.POSTGRES:
 		db, err := sql.Open(driver, connectionConfig.(string))
 		if err != nil {
@@ -783,7 +810,16 @@ func GetInfoSchema(sourceProfile profiles.SourceProfile) (common.InfoSchema, err
 	case constants.DYNAMODB:
 		mySession := session.Must(session.NewSession())
 		dydbClient := dydb.New(mySession, connectionConfig.(*aws.Config))
-		return dynamodb.InfoSchemaImpl{DynamoClient: dydbClient, SampleSize: profiles.GetSchemaSampleSize(sourceProfile)}, nil
+		var dydbStreamsClient *dynamodbstreams.DynamoDBStreams
+		if sourceProfile.Conn.Streaming {
+			newSession := session.Must(session.NewSession())
+			dydbStreamsClient = dynamodbstreams.New(newSession, connectionConfig.(*aws.Config))
+		}
+		return dynamodb.InfoSchemaImpl{
+			DynamoClient:        dydbClient,
+			SampleSize:          profiles.GetSchemaSampleSize(sourceProfile),
+			DynamoStreamsClient: dydbStreamsClient,
+		}, nil
 	case constants.SQLSERVER:
 		db, err := sql.Open(driver, connectionConfig.(string))
 		dbName := getDbNameFromSQLConnectionStr(driver, connectionConfig.(string))
@@ -797,7 +833,7 @@ func GetInfoSchema(sourceProfile profiles.SourceProfile) (common.InfoSchema, err
 		if err != nil {
 			return nil, err
 		}
-		return oracle.InfoSchemaImpl{DbName: strings.ToUpper(dbName), Db: db}, nil
+		return oracle.InfoSchemaImpl{DbName: strings.ToUpper(dbName), Db: db, SourceProfile: sourceProfile, TargetProfile: targetProfile}, nil
 	default:
 		return nil, fmt.Errorf("driver %s not supported", driver)
 	}
diff --git a/go.sum b/go.sum
index 6aeef52..ddd8a83 100644
--- a/go.sum
+++ b/go.sum
@@ -514,6 +514,7 @@ github.com/julienschmidt/httprouter v1.2.0/go.mod h1:SYymIcj16QtmaHHD7aYtjjsJG7V
 github.com/julienschmidt/httprouter v1.3.0/go.mod h1:JR6WtHb+2LUe8TCKY3cZOxFyyO8IZAc4RVcycCCAKdM=
 github.com/jung-kurt/gofpdf v1.0.3-0.20190309125859-24315acbbda5/go.mod h1:7Id9E/uU8ce6rXgefFLlgrJj/GYY22cpxn+r32jIOes=
 github.com/k0kubun/colorstring v0.0.0-20150214042306-9440f1994b88/go.mod h1:3w7q1U84EfirKl04SVQ/s7nPm1ZPhiXd34z40TNz36k=
+github.com/k0kubun/go-ansi v0.0.0-20180517002512-3bf9e2903213/go.mod h1:vNUNkEQ1e29fT/6vq2aBdFsgNPmy8qMdSay1npru+Sw=
 github.com/kataras/golog v0.0.9/go.mod h1:12HJgwBIZFNGL0EJnMRhmvGA0PQGx8VFwrZtM4CqbAk=
 github.com/kataras/iris/v12 v12.0.1/go.mod h1:udK4vLQKkdDqMGJJVd/msuMtN6hpYJhg/lSzuxjhO+U=
 github.com/kataras/neffos v0.0.10/go.mod h1:ZYmJC07hQPW67eKuzlfY7SO3bC0mw83A3j6im82hfqw=
@@ -571,6 +572,8 @@ github.com/mattn/go-isatty v0.0.14/go.mod h1:7GGIvUiUoEMVVmxf/4nioHXj79iQHKdU27k
 github.com/mattn/go-runewidth v0.0.9/go.mod h1:H031xJmbD/WCDINGzjvQ9THkh0rPKHF+m2gUSrubnMI=
 github.com/mattn/go-runewidth v0.0.12 h1:Y41i/hVW3Pgwr8gV+J23B9YEY0zxjptBuCWEaxmAOow=
 github.com/mattn/go-runewidth v0.0.12/go.mod h1:RAqKPSqVFrSLVXbA8x7dzmKdmGzieGRCM46jaSJTDAk=
+github.com/mattn/go-runewidth v0.0.13 h1:lTGmDsbAYt5DmK6OnoV7EuIF1wEIFAcxld6ypU4OSgU=
+github.com/mattn/go-runewidth v0.0.13/go.mod h1:Jdepj2loyihRzMpdS35Xk/zdY8IAYHsh153qUoGf23w=
 github.com/mattn/goveralls v0.0.2/go.mod h1:8d1ZMHsd7fW6IRPKQh46F2WRpyib5/X4FOpevwGNQEw=
 github.com/matttproud/golang_protobuf_extensions v1.0.1 h1:4hp9jkHxhMHkqkrB3Ix0jegS5sx/RkqARlsWZ6pIwiU=
 github.com/matttproud/golang_protobuf_extensions v1.0.1/go.mod h1:D8He9yQNgCq6Z5Ld7szi9bcBfOoFv/3dc6xSMkL2PC0=
@@ -579,6 +582,8 @@ github.com/mediocregopher/radix/v3 v3.3.0/go.mod h1:EmfVyvspXz1uZEyPBMyGK+kjWiKQ
 github.com/microcosm-cc/bluemonday v1.0.2/go.mod h1:iVP4YcDBq+n/5fb23BhYFvIMq/leAFZyRl6bYmGDlGc=
 github.com/miekg/dns v1.0.14/go.mod h1:W1PPwlIAgtquWBMBEV9nkV9Cazfe8ScdGz/Lj7v3Nrg=
 github.com/mitchellh/cli v1.0.0/go.mod h1:hNIlj7HEI86fIcpObd7a0FcrxTWetlwJDGcceTlRvqc=
+github.com/mitchellh/colorstring v0.0.0-20190213212951-d06e56a500db h1:62I3jR2EmQ4l5rM/4FEfDWcRD+abF5XlKShorW5LRoQ=
+github.com/mitchellh/colorstring v0.0.0-20190213212951-d06e56a500db/go.mod h1:l0dey0ia/Uv7NcFFVbCLtqEBQbrT4OCwCSKTEv6enCw=
 github.com/mitchellh/go-homedir v1.0.0/go.mod h1:SfyaCUpYCn1Vlf4IUYiD9fPX4A5wJrkLzIz1N1q0pr0=
 github.com/mitchellh/go-homedir v1.1.0/go.mod h1:SfyaCUpYCn1Vlf4IUYiD9fPX4A5wJrkLzIz1N1q0pr0=
 github.com/mitchellh/go-testing-interface v1.0.0/go.mod h1:kRemZodwjscx+RGhAo8eIhFbs2+BFgRtFPeD/KE+zxI=
@@ -933,6 +938,7 @@ golang.org/x/crypto v0.0.0-20201016220609-9e8e0b390897/go.mod h1:LzIPMQfyMNhhGPh
 golang.org/x/crypto v0.0.0-20201112155050-0c6587e931a9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/crypto v0.0.0-20210421170649-83a5a9bb288b/go.mod h1:T9bdIzuCu7OtxOm1hfPfRQxPLYneinmdGuTeoZ9dtd4=
 golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
+golang.org/x/crypto v0.0.0-20220131195533-30dcbda58838/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
 golang.org/x/crypto v0.0.0-20220214200702-86341886e292 h1:f+lwQ+GtmgoY+A2YaQxlSOnDjXcQ7ZRLWOHbC6HtRqE=
 golang.org/x/crypto v0.0.0-20220214200702-86341886e292/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
 golang.org/x/exp v0.0.0-20180321215751-8460e604b9de/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
diff --git a/sources/common/infoschema.go b/sources/common/infoschema.go
index d79050d..70b477a 100644
--- a/sources/common/infoschema.go
+++ b/sources/common/infoschema.go
@@ -15,8 +15,11 @@
 package common
 
 import (
+	"context"
 	"fmt"
 
+	sp "cloud.google.com/go/spanner"
+
 	"github.com/cloudspannerecosystem/harbourbridge/internal"
 	"github.com/cloudspannerecosystem/harbourbridge/schema"
 	"github.com/cloudspannerecosystem/harbourbridge/spanner/ddl"
@@ -34,6 +37,8 @@ type InfoSchema interface {
 	GetForeignKeys(conv *internal.Conv, table SchemaAndName) (foreignKeys []schema.ForeignKey, err error)
 	GetIndexes(conv *internal.Conv, table SchemaAndName) ([]schema.Index, error)
 	ProcessData(conv *internal.Conv, srcTable string, srcSchema schema.Table, spTable string, spCols []string, spSchema ddl.CreateTable) error
+	StartChangeDataCapture(ctx context.Context, conv *internal.Conv) (map[string]interface{}, error)
+	StartStreamingMigration(ctx context.Context, client *sp.Client, conv *internal.Conv, streamInfo map[string]interface{}) error
 }
 
 // SchemaAndName contains the schema and name for a table
diff --git a/sources/dynamodb/schema.go b/sources/dynamodb/schema.go
index 868cced..4e78870 100644
--- a/sources/dynamodb/schema.go
+++ b/sources/dynamodb/schema.go
@@ -15,6 +15,7 @@
 package dynamodb
 
 import (
+	"context"
 	"fmt"
 	"log"
 	"math/big"
@@ -24,6 +25,8 @@ import (
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/dynamodb"
 	"github.com/aws/aws-sdk-go/service/dynamodb/dynamodbiface"
+	"github.com/aws/aws-sdk-go/service/dynamodbstreams/dynamodbstreamsiface"
+
 	"github.com/cloudspannerecosystem/harbourbridge/internal"
 	"github.com/cloudspannerecosystem/harbourbridge/schema"
 	"github.com/cloudspannerecosystem/harbourbridge/sources/common"
@@ -48,8 +51,9 @@ const (
 )
 
 type InfoSchemaImpl struct {
-	DynamoClient dynamodbiface.DynamoDBAPI
-	SampleSize   int64
+	DynamoClient        dynamodbiface.DynamoDBAPI
+	DynamoStreamsClient dynamodbstreamsiface.DynamoDBStreamsAPI
+	SampleSize          int64
 }
 
 func (isi InfoSchemaImpl) GetToDdl() common.ToDdl {
@@ -188,6 +192,14 @@ func (isi InfoSchemaImpl) ProcessData(conv *internal.Conv, srcTable string, srcS
 	return nil
 }
 
+func (isi InfoSchemaImpl) StartChangeDataCapture(ctx context.Context, conv *internal.Conv) (map[string]interface{}, error) {
+	return nil, nil
+}
+
+func (isi InfoSchemaImpl) StartStreamingMigration(ctx context.Context, client *sp.Client, conv *internal.Conv, LatestStream map[string]interface{}) error {
+	return nil
+}
+
 func getSchemaIndexStruct(indexName string, keySchema []*dynamodb.KeySchemaElement) schema.Index {
 	var keys []schema.Key
 	for _, j := range keySchema {
diff --git a/sources/mysql/infoschema.go b/sources/mysql/infoschema.go
index 95bf33a..2542877 100644
--- a/sources/mysql/infoschema.go
+++ b/sources/mysql/infoschema.go
@@ -15,23 +15,30 @@
 package mysql
 
 import (
+	"context"
 	"database/sql"
 	"fmt"
 	"sort"
 	"strings"
 
+	sp "cloud.google.com/go/spanner"
+	_ "github.com/go-sql-driver/mysql" // The driver should be used via the database/sql package.
+	_ "github.com/lib/pq"
+
 	"github.com/cloudspannerecosystem/harbourbridge/internal"
+	"github.com/cloudspannerecosystem/harbourbridge/profiles"
 	"github.com/cloudspannerecosystem/harbourbridge/schema"
 	"github.com/cloudspannerecosystem/harbourbridge/sources/common"
 	"github.com/cloudspannerecosystem/harbourbridge/spanner/ddl"
-	_ "github.com/go-sql-driver/mysql" // The driver should be used via the database/sql package.
-	_ "github.com/lib/pq"
+	"github.com/cloudspannerecosystem/harbourbridge/streaming"
 )
 
 // InfoSchemaImpl is MySQL specific implementation for InfoSchema.
 type InfoSchemaImpl struct {
-	DbName string
-	Db     *sql.DB
+	DbName        string
+	Db            *sql.DB
+	SourceProfile profiles.SourceProfile
+	TargetProfile profiles.TargetProfile
 }
 
 // GetToDdl implement the common.InfoSchema interface.
@@ -325,6 +332,31 @@ func (isi InfoSchemaImpl) GetIndexes(conv *internal.Conv, table common.SchemaAnd
 	}
 	return indexes, nil
 }
+
+// StartChangeDataCapture is used for automatic triggering of Datastream job when
+// performing a streaming migration.
+func (isi InfoSchemaImpl) StartChangeDataCapture(ctx context.Context, conv *internal.Conv) (map[string]interface{}, error) {
+	mp := make(map[string]interface{})
+	streamingCfg, err := streaming.StartDatastream(ctx, isi.SourceProfile, isi.TargetProfile)
+	if err != nil {
+		err = fmt.Errorf("error starting datastream: %v", err)
+		return nil, err
+	}
+	mp["streamingCfg"] = streamingCfg
+	return mp, err
+}
+
+// StartStreamingMigration is used for automatic triggering of Dataflow job when
+// 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)
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
 func toType(dataType string, columnType string, charLen sql.NullInt64, numericPrecision, numericScale sql.NullInt64) schema.Type {
 	switch {
 	case dataType == "set":
diff --git a/sources/oracle/infoschema.go b/sources/oracle/infoschema.go
index 31786ed..119ead2 100644
--- a/sources/oracle/infoschema.go
+++ b/sources/oracle/infoschema.go
@@ -15,20 +15,27 @@
 package oracle
 
 import (
+	"context"
 	"database/sql"
 	"fmt"
 	"sort"
 	"strings"
 
+	sp "cloud.google.com/go/spanner"
+
 	"github.com/cloudspannerecosystem/harbourbridge/internal"
+	"github.com/cloudspannerecosystem/harbourbridge/profiles"
 	"github.com/cloudspannerecosystem/harbourbridge/schema"
 	"github.com/cloudspannerecosystem/harbourbridge/sources/common"
 	"github.com/cloudspannerecosystem/harbourbridge/spanner/ddl"
+	"github.com/cloudspannerecosystem/harbourbridge/streaming"
 )
 
 type InfoSchemaImpl struct {
-	DbName string
-	Db     *sql.DB
+	DbName        string
+	Db            *sql.DB
+	SourceProfile profiles.SourceProfile
+	TargetProfile profiles.TargetProfile
 }
 
 // GetToDdl function below implement the common.InfoSchema interface.
@@ -66,6 +73,8 @@ func getSelectQuery(srcDb string, schemaName string, tableName string, colNames
 				FROM TABLE ("%s"."%s")) AS "%s"`, tableName, cn, cn)
 		} else {
 			switch colDefs[cn].Type.Name {
+			case "NUMBER":
+				s = fmt.Sprintf(`TO_CHAR("%s") AS "%s"`, cn, cn)
 			case "XMLTYPE":
 				s = fmt.Sprintf(`CAST(XMLTYPE.getStringVal("%s") AS VARCHAR2(4000)) AS "%s"`, cn, cn)
 			case "SDO_GEOMETRY":
@@ -394,6 +403,30 @@ func (isi InfoSchemaImpl) GetIndexes(conv *internal.Conv, table common.SchemaAnd
 	return indexes, nil
 }
 
+// StartChangeDataCapture is used for automatic triggering of Datastream job when
+// performing a streaming migration.
+func (isi InfoSchemaImpl) StartChangeDataCapture(ctx context.Context, conv *internal.Conv) (map[string]interface{}, error) {
+	mp := make(map[string]interface{})
+	streamingCfg, err := streaming.StartDatastream(ctx, isi.SourceProfile, isi.TargetProfile)
+	if err != nil {
+		err = fmt.Errorf("error starting datastream: %v", err)
+		return nil, err
+	}
+	mp["streamingCfg"] = streamingCfg
+	return mp, err
+}
+
+// StartStreamingMigration is used for automatic triggering of Dataflow job when
+// 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)
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
 func toType(dataType string, typecode, elementDataType sql.NullString, charLen sql.NullInt64, numericPrecision, numericScale, elementCharMaxLen, elementNumericPrecision, elementNumericScale sql.NullInt64) schema.Type {
 	switch {
 	case typecode.Valid && typecode.String == "COLLECTION":
diff --git a/sources/oracle/toddl.go b/sources/oracle/toddl.go
index e6807e0..9c07ddc 100644
--- a/sources/oracle/toddl.go
+++ b/sources/oracle/toddl.go
@@ -88,12 +88,23 @@ func toSpannerTypeInternal(conv *internal.Conv, spType string, srcType string, m
 		case ddl.String:
 			return ddl.Type{Name: ddl.String, Len: ddl.MaxLength}, nil
 		default:
-			// If no scale is avalible then map it to int64, and numeric elsewhere.
-			if len(mods) == 1 && mods[0] >= 1 && mods[0] < 19 {
-				return ddl.Type{Name: ddl.Int64}, nil
-			} else {
+			modsLen := len(mods)
+			if modsLen == 0 {
 				return ddl.Type{Name: ddl.Numeric}, nil
+			} else if modsLen == 1 { // Only precision is available.
+				if mods[0] > 29 {
+					// Max precision in Oracle is 38. String representation of the number should not have more than 50 characters
+					// https://docs.oracle.com/cd/B19306_01/server.102/b14237/limits001.htm#i287903
+					return ddl.Type{Name: ddl.String, Len: 50}, nil
+				}
+				return ddl.Type{Name: ddl.Int64}, nil
+			} else if mods[0] > 29 || mods[1] > 9 { // When both precision and scale are available and within limit
+				// Max precision in Oracle is 38. String representation of the number should not have more than 50 characters
+				// https://docs.oracle.com/cd/B19306_01/server.102/b14237/limits001.htm#i287903
+				return ddl.Type{Name: ddl.String, Len: 50}, nil
 			}
+
+			return ddl.Type{Name: ddl.Numeric}, nil
 		}
 
 	case "BFILE", "BLOB":
diff --git a/sources/postgres/infoschema.go b/sources/postgres/infoschema.go
index cfea57a..bfbd6d6 100644
--- a/sources/postgres/infoschema.go
+++ b/sources/postgres/infoschema.go
@@ -15,6 +15,7 @@
 package postgres
 
 import (
+	"context"
 	"database/sql"
 	"fmt"
 	"math/bits"
@@ -24,6 +25,7 @@ import (
 	"time"
 
 	"cloud.google.com/go/civil"
+	sp "cloud.google.com/go/spanner"
 	_ "github.com/lib/pq" // we will use database/sql package instead of using this package directly
 
 	"github.com/cloudspannerecosystem/harbourbridge/internal"
@@ -37,6 +39,15 @@ type InfoSchemaImpl struct {
 	Db *sql.DB
 }
 
+// We leave the 2 functions below empty to be able to pass this as an infoSchema interface. We don't need these for now.
+func (isi InfoSchemaImpl) StartChangeDataCapture(ctx context.Context, conv *internal.Conv) (map[string]interface{}, error) {
+	return nil, nil
+}
+
+func (isi InfoSchemaImpl) StartStreamingMigration(ctx context.Context, client *sp.Client, conv *internal.Conv, streamingInfo map[string]interface{}) error {
+	return nil
+}
+
 // GetToDdl function below implement the common.InfoSchema interface.
 func (isi InfoSchemaImpl) GetToDdl() common.ToDdl {
 	return ToDdlImpl{}
diff --git a/sources/spanner/infoschema.go b/sources/spanner/infoschema.go
index 5cc701c..df47ae6 100644
--- a/sources/spanner/infoschema.go
+++ b/sources/spanner/infoschema.go
@@ -45,7 +45,7 @@ func (isi InfoSchemaImpl) GetToDdl() common.ToDdl {
 	return ToDdlImpl{}
 }
 
-// We leave the 3 functions below empty to be able to pass this as an infoSchema interface. We don't need these for now.
+// We leave the 5 functions below empty to be able to pass this as an infoSchema interface. We don't need these for now.
 func (isi InfoSchemaImpl) ProcessData(conv *internal.Conv, srcTable string, srcSchema schema.Table, spTable string, spCols []string, spSchema ddl.CreateTable) error {
 	return nil
 }
@@ -58,6 +58,14 @@ func (isi InfoSchemaImpl) GetRowsFromTable(conv *internal.Conv, srcTable string)
 	return nil, nil
 }
 
+func (isi InfoSchemaImpl) StartChangeDataCapture(ctx context.Context, conv *internal.Conv) (map[string]interface{}, error) {
+	return nil, nil
+}
+
+func (isi InfoSchemaImpl) StartStreamingMigration(ctx context.Context, client *spanner.Client, conv *internal.Conv, streamingInfo map[string]interface{}) error {
+	return nil
+}
+
 // GetTableName returns table name.
 func (isi InfoSchemaImpl) GetTableName(schema string, tableName string) string {
 	if isi.TargetDb == constants.TargetExperimentalPostgres {
diff --git a/sources/sqlserver/infoschema.go b/sources/sqlserver/infoschema.go
index a238cca..09cadba 100644
--- a/sources/sqlserver/infoschema.go
+++ b/sources/sqlserver/infoschema.go
@@ -15,11 +15,14 @@
 package sqlserver
 
 import (
+	"context"
 	"database/sql"
 	"fmt"
 	"sort"
 	"strings"
 
+	sp "cloud.google.com/go/spanner"
+
 	"github.com/cloudspannerecosystem/harbourbridge/internal"
 	"github.com/cloudspannerecosystem/harbourbridge/schema"
 	"github.com/cloudspannerecosystem/harbourbridge/sources/common"
@@ -50,6 +53,15 @@ func (isi InfoSchemaImpl) GetToDdl() common.ToDdl {
 	return ToDdlImpl{}
 }
 
+// We leave the 2 functions below empty to be able to pass this as an infoSchema interface. We don't need these for now.
+func (isi InfoSchemaImpl) StartChangeDataCapture(ctx context.Context, conv *internal.Conv) (map[string]interface{}, error) {
+	return nil, nil
+}
+
+func (isi InfoSchemaImpl) StartStreamingMigration(ctx context.Context, client *sp.Client, conv *internal.Conv, streamingInfo map[string]interface{}) error {
+	return nil
+}
+
 // GetTableName returns table name.
 func (isi InfoSchemaImpl) GetTableName(schema string, tableName string) string {
 	if schema == "dbo" { // Drop 'dbo' prefix.
diff --git a/streaming/streaming.go b/streaming/streaming.go
index 6831e95..ae8d7dc 100644
--- a/streaming/streaming.go
+++ b/streaming/streaming.go
@@ -210,6 +210,7 @@ func LaunchStream(ctx context.Context, sourceProfile profiles.SourceProfile, pro
 	}
 	fmt.Println("Successfully created stream ", datastreamCfg.StreamId)
 
+	fmt.Print("Setting stream state to RUNNING...")
 	streamInfo.Name = fmt.Sprintf("projects/%s/locations/%s/streams/%s", projectID, datastreamCfg.StreamLocation, datastreamCfg.StreamId)
 	updateStreamRequest := &datastreampb.UpdateStreamRequest{
 		UpdateMask: &fieldmaskpb.FieldMask{Paths: []string{"state"}},
@@ -219,12 +220,11 @@ func LaunchStream(ctx context.Context, sourceProfile profiles.SourceProfile, pro
 	if err != nil {
 		return fmt.Errorf("could not create update request: %v", err)
 	}
-
 	_, err = upOp.Wait(ctx)
 	if err != nil {
 		return fmt.Errorf("update stream operation failed: %v", err)
 	}
-	fmt.Println("Stream state set to RUNNING...")
+	fmt.Println("Done")
 	return nil
 }
 
@@ -287,3 +287,35 @@ func LaunchDataflowJob(ctx context.Context, targetProfile profiles.TargetProfile
 		" will have to be manually cleaned up via he UI. HarbourBridge will not delete them post completion of the migration.")
 	return nil
 }
+
+func getStreamingConfig(sourceProfile profiles.SourceProfile, targetProfile profiles.TargetProfile) (StreamingCfg, error) {
+	switch sourceProfile.Conn.Ty {
+	case profiles.SourceProfileConnectionTypeMySQL:
+		return ReadStreamingConfig(sourceProfile.Conn.Mysql.StreamingConfig, targetProfile.Conn.Sp.Dbname)
+	case profiles.SourceProfileConnectionTypeOracle:
+		return ReadStreamingConfig(sourceProfile.Conn.Oracle.StreamingConfig, targetProfile.Conn.Sp.Dbname)
+	default:
+		return StreamingCfg{}, fmt.Errorf("only MySQL and Oracle are supported as source streams")
+	}
+}
+
+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)
+	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) error {
+	err := LaunchDataflowJob(ctx, targetProfile, streamingCfg.DatastreamCfg, streamingCfg.DataflowCfg)
+	if err != nil {
+		return fmt.Errorf("error launching dataflow: %v", err)
+	}
+	return nil
+}


Test patch:
diff --git a/sources/dynamodb/schema_test.go b/sources/dynamodb/schema_test.go
index 4f7a00f..8c7d3f8 100644
--- a/sources/dynamodb/schema_test.go
+++ b/sources/dynamodb/schema_test.go
@@ -21,11 +21,12 @@ import (
 
 	"github.com/aws/aws-sdk-go/service/dynamodb"
 	"github.com/aws/aws-sdk-go/service/dynamodb/dynamodbiface"
+	"github.com/stretchr/testify/assert"
+
 	"github.com/cloudspannerecosystem/harbourbridge/internal"
 	"github.com/cloudspannerecosystem/harbourbridge/schema"
 	"github.com/cloudspannerecosystem/harbourbridge/sources/common"
 	"github.com/cloudspannerecosystem/harbourbridge/spanner/ddl"
-	"github.com/stretchr/testify/assert"
 )
 
 type mockDynamoClient struct {
@@ -167,7 +168,7 @@ func TestProcessSchema(t *testing.T) {
 	sampleSize := int64(10000)
 
 	conv := internal.MakeConv()
-	err := common.ProcessSchema(conv, InfoSchemaImpl{client, sampleSize})
+	err := common.ProcessSchema(conv, InfoSchemaImpl{client, nil, sampleSize})
 
 	assert.Nil(t, err)
 	expectedSchema := map[string]ddl.CreateTable{
@@ -270,7 +271,7 @@ func TestProcessSchema_FullDataTypes(t *testing.T) {
 	sampleSize := int64(10000)
 
 	conv := internal.MakeConv()
-	err := common.ProcessSchema(conv, InfoSchemaImpl{client, sampleSize})
+	err := common.ProcessSchema(conv, InfoSchemaImpl{client, nil, sampleSize})
 
 	assert.Nil(t, err)
 	expectedSchema := map[string]ddl.CreateTable{
@@ -353,7 +354,7 @@ func TestProcessData(t *testing.T) {
 		func(table string, cols []string, vals []interface{}) {
 			rows = append(rows, spannerData{table: table, cols: cols, vals: vals})
 		})
-	common.ProcessData(conv, InfoSchemaImpl{client, 10})
+	common.ProcessData(conv, InfoSchemaImpl{client, nil, 10})
 	assert.Equal(t,
 		[]spannerData{
 			{
@@ -532,7 +533,7 @@ func TestInfoSchemaImpl_GetIndexes(t *testing.T) {
 
 	dySchema := common.SchemaAndName{Name: "test"}
 	conv := internal.MakeConv()
-	isi := InfoSchemaImpl{client, 10}
+	isi := InfoSchemaImpl{client, nil, 10}
 	indexes, err := isi.GetIndexes(conv, dySchema)
 	assert.Nil(t, err)
 
@@ -587,7 +588,7 @@ func TestInfoSchemaImpl_GetConstraints(t *testing.T) {
 
 	dySchema := common.SchemaAndName{Name: "test"}
 	conv := internal.MakeConv()
-	isi := InfoSchemaImpl{client, 10}
+	isi := InfoSchemaImpl{client, nil, 10}
 	primaryKeys, constraints, err := isi.GetConstraints(conv, dySchema)
 	assert.Nil(t, err)
 
@@ -608,7 +609,7 @@ func TestInfoSchemaImpl_GetTables(t *testing.T) {
 	client := &mockDynamoClient{
 		listTableOutputs: listTableOutputs,
 	}
-	isi := InfoSchemaImpl{client, 10}
+	isi := InfoSchemaImpl{client, nil, 10}
 	tables, err := isi.GetTables()
 	assert.Nil(t, err)
 	assert.Equal(t, []common.SchemaAndName{{"", "table-a"}, {"", "table-b"}}, tables)
@@ -618,7 +619,7 @@ func TestInfoSchemaImpl_GetTableName(t *testing.T) {
 	tableNameA := "table-a"
 
 	client := &mockDynamoClient{}
-	isi := InfoSchemaImpl{client, 10}
+	isi := InfoSchemaImpl{client, nil, 10}
 	table := isi.GetTableName("", tableNameA)
 	assert.Equal(t, tableNameA, table)
 }
@@ -662,7 +663,7 @@ func TestInfoSchemaImpl_GetColumns(t *testing.T) {
 	}
 	dySchema := common.SchemaAndName{Name: "test"}
 
-	isi := InfoSchemaImpl{client, 10}
+	isi := InfoSchemaImpl{client, nil, 10}
 
 	colDefs, colNames, err := isi.GetColumns(conv, dySchema, nil, nil)
 	assert.Nil(t, err)
@@ -680,7 +681,7 @@ func TestInfoSchemaImpl_GetForeignKeys(t *testing.T) {
 	dySchema := common.SchemaAndName{Name: "test"}
 	conv := internal.MakeConv()
 	client := &mockDynamoClient{}
-	isi := InfoSchemaImpl{client, 10}
+	isi := InfoSchemaImpl{client, nil, 10}
 	fk, err := isi.GetForeignKeys(conv, dySchema)
 	assert.Nil(t, err)
 	assert.Nil(t, fk)
@@ -703,7 +704,7 @@ func TestInfoSchemaImpl_GetRowCount(t *testing.T) {
 		describeTableOutputs: describeTableOutputs,
 	}
 
-	isi := InfoSchemaImpl{client, 10}
+	isi := InfoSchemaImpl{client, nil, 10}
 	dySchema := common.SchemaAndName{Name: tableNameA}
 
 	rowCount, err := isi.GetRowCount(dySchema)
@@ -735,7 +736,7 @@ func TestInfoSchemaImpl_GetRowsFromTable(t *testing.T) {
 		scanOutputs: scanOutputs,
 	}
 	tableName := "testtable"
-	isi := InfoSchemaImpl{client, 10}
+	isi := InfoSchemaImpl{client, nil, 10}
 
 	rows, err := isi.GetRowsFromTable(conv, tableName)
 	assert.Nil(t, err)
@@ -771,7 +772,7 @@ func TestInfoSchemaImpl_ProcessData(t *testing.T) {
 	client := &mockDynamoClient{
 		scanOutputs: scanOutputs,
 	}
-	isi := InfoSchemaImpl{client, 10}
+	isi := InfoSchemaImpl{client, nil, 10}
 
 	tableName := "testtable"
 	cols := []string{"a", "b", "c", "d"}
@@ -906,7 +907,7 @@ func TestSetRowStats(t *testing.T) {
 		describeTableOutputs: describeTableOutputs,
 	}
 
-	common.SetRowStats(conv, InfoSchemaImpl{client, 10})
+	common.SetRowStats(conv, InfoSchemaImpl{client, nil, 10})
 
 	assert.Equal(t, tableItemCountA, conv.Stats.Rows[tableNameA])
 	assert.Equal(t, tableItemCountB, conv.Stats.Rows[tableNameB])
diff --git a/sources/mysql/infoschema_test.go b/sources/mysql/infoschema_test.go
index 0f1c841..dc1e101 100644
--- a/sources/mysql/infoschema_test.go
+++ b/sources/mysql/infoschema_test.go
@@ -20,11 +20,13 @@ import (
 	"testing"
 
 	"github.com/DATA-DOG/go-sqlmock"
+	"github.com/stretchr/testify/assert"
+
 	"github.com/cloudspannerecosystem/harbourbridge/internal"
+	"github.com/cloudspannerecosystem/harbourbridge/profiles"
 	"github.com/cloudspannerecosystem/harbourbridge/schema"
 	"github.com/cloudspannerecosystem/harbourbridge/sources/common"
 	"github.com/cloudspannerecosystem/harbourbridge/spanner/ddl"
-	"github.com/stretchr/testify/assert"
 )
 
 type mockSpec struct {
@@ -194,7 +196,8 @@ func TestProcessSchemaMYSQL(t *testing.T) {
 	}
 	db := mkMockDB(t, ms)
 	conv := internal.MakeConv()
-	err := common.ProcessSchema(conv, InfoSchemaImpl{"test", db})
+	isi := InfoSchemaImpl{"test", db, profiles.SourceProfile{}, profiles.TargetProfile{}}
+	err := common.ProcessSchema(conv, isi)
 	assert.Nil(t, err)
 	expectedSchema := map[string]ddl.CreateTable{
 		"user": ddl.CreateTable{
@@ -317,7 +320,8 @@ func TestProcessData(t *testing.T) {
 		func(table string, cols []string, vals []interface{}) {
 			rows = append(rows, spannerData{table: table, cols: cols, vals: vals})
 		})
-	common.ProcessData(conv, InfoSchemaImpl{"test", db})
+	isi := InfoSchemaImpl{"test", db, profiles.SourceProfile{}, profiles.TargetProfile{}}
+	common.ProcessData(conv, isi)
 	assert.Equal(t,
 		[]spannerData{
 			spannerData{table: "te_st", cols: []string{"a_a", "Ab", "Ac_"}, vals: []interface{}{float64(42.3), int64(3), "cat"}},
@@ -373,7 +377,8 @@ func TestProcessData_MultiCol(t *testing.T) {
 	}
 	db := mkMockDB(t, ms)
 	conv := internal.MakeConv()
-	err := common.ProcessSchema(conv, InfoSchemaImpl{"test", db})
+	isi := InfoSchemaImpl{"test", db, profiles.SourceProfile{}, profiles.TargetProfile{}}
+	err := common.ProcessSchema(conv, isi)
 	assert.Nil(t, err)
 	expectedSchema := map[string]ddl.CreateTable{
 		"test": ddl.CreateTable{
@@ -397,7 +402,7 @@ func TestProcessData_MultiCol(t *testing.T) {
 		func(table string, cols []string, vals []interface{}) {
 			rows = append(rows, spannerData{table: table, cols: cols, vals: vals})
 		})
-	common.ProcessData(conv, InfoSchemaImpl{"test", db})
+	common.ProcessData(conv, isi)
 	assert.Equal(t, []spannerData{
 		{table: "test", cols: []string{"a", "b", "synth_id"}, vals: []interface{}{"cat", float64(42.3), int64(0)}},
 		{table: "test", cols: []string{"a", "c", "synth_id"}, vals: []interface{}{"dog", int64(22), int64(-9223372036854775808)}}},
@@ -425,7 +430,8 @@ func TestSetRowStats(t *testing.T) {
 	db := mkMockDB(t, ms)
 	conv := internal.MakeConv()
 	conv.SetDataMode()
-	common.SetRowStats(conv, InfoSchemaImpl{"test", db})
+	isi := InfoSchemaImpl{"test", db, profiles.SourceProfile{}, profiles.TargetProfile{}}
+	common.SetRowStats(conv, isi)
 	assert.Equal(t, int64(5), conv.Stats.Rows["test1"])
 	assert.Equal(t, int64(142), conv.Stats.Rows["test2"])
 	assert.Equal(t, int64(0), conv.Unexpecteds())
diff --git a/sources/oracle/infoschema_test.go b/sources/oracle/infoschema_test.go
index 0097e09..6a254d9 100644
--- a/sources/oracle/infoschema_test.go
+++ b/sources/oracle/infoschema_test.go
@@ -20,10 +20,12 @@ import (
 	"testing"
 
 	"github.com/DATA-DOG/go-sqlmock"
+	"github.com/stretchr/testify/assert"
+
 	"github.com/cloudspannerecosystem/harbourbridge/internal"
+	"github.com/cloudspannerecosystem/harbourbridge/profiles"
 	"github.com/cloudspannerecosystem/harbourbridge/sources/common"
 	"github.com/cloudspannerecosystem/harbourbridge/spanner/ddl"
-	"github.com/stretchr/testify/assert"
 )
 
 type mockSpec struct {
@@ -151,7 +153,7 @@ func TestProcessSchemaOracle(t *testing.T) {
 	}
 	db := mkMockDB(t, ms)
 	conv := internal.MakeConv()
-	err := common.ProcessSchema(conv, InfoSchemaImpl{"test", db})
+	err := common.ProcessSchema(conv, InfoSchemaImpl{"test", db, profiles.SourceProfile{}, profiles.TargetProfile{}})
 	assert.Nil(t, err)
 	expectedSchema := map[string]ddl.CreateTable{
 		"USER": {