Repository: cloudspannerecosystem/harbourbridge
Pull request #320 (2022-06-23 08:03:16)

Static file embedding for web
- Static files are embedded for web.
- Go embed is supported only by go versions 1.16 and above, thus edited the version in go.mod here.

shellymathew98: > Can you also add details around how you validated the change?
> 
> An ideal test would be like so:
> 
> 1. Build the binary
> 2. Copy it over to some unknown location
> 3. run the binary
> 4. the webUI should work as expected.

- I tried placing the binary on a different location and webUI works fine from there.

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.mod b/go.mod
index 8905860..df818ca 100644
--- a/go.mod
+++ b/go.mod
@@ -1,6 +1,6 @@
 module github.com/cloudspannerecosystem/harbourbridge
 
-go 1.13
+go 1.16
 
 require (
 	cloud.google.com/go v0.100.2
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/main.go b/main.go
index c53e04e..76ae6d2 100644
--- a/main.go
+++ b/main.go
@@ -19,6 +19,7 @@ package main
 
 import (
 	"context"
+	"embed"
 	"flag"
 	"fmt"
 	"os"
@@ -39,6 +40,9 @@ import (
 	"github.com/google/subcommands"
 )
 
+//go:embed frontend/*
+var frontendDir embed.FS
+
 var (
 	dbNameOverride   string
 	instanceOverride string
@@ -126,6 +130,7 @@ func main() {
 
 	// Note: the web interface does not use any commandline flags.
 	if webapi {
+		web.FrontendDir = frontendDir
 		web.App()
 		return
 	}
diff --git a/profiles/source_profile.go b/profiles/source_profile.go
index ef4d47a..ffb7d78 100644
--- a/profiles/source_profile.go
+++ b/profiles/source_profile.go
@@ -261,6 +261,7 @@ type SourceProfileConnectionDynamoDB struct {
 	AwsRegion          string // Same as AWS_REGION environment variable
 	DydbEndpoint       string // Same as DYNAMODB_ENDPOINT_OVERRIDE environment variable
 	SchemaSampleSize   int64  // Number of rows to use for inferring schema (default 100,000)
+	enableStreaming    string // Used for confirming streaming migration (valid options: `yes`,`no`,`true`,`false`)
 }
 
 func NewSourceProfileConnectionDynamoDB(params map[string]string) (SourceProfileConnectionDynamoDB, error) {
@@ -288,6 +289,16 @@ func NewSourceProfileConnectionDynamoDB(params map[string]string) (SourceProfile
 	if dydb.DydbEndpoint, ok = params["dydb-endpoint"]; ok {
 		os.Setenv("DYNAMODB_ENDPOINT_OVERRIDE", dydb.DydbEndpoint)
 	}
+	if dydb.enableStreaming, ok = params["enableStreaming"]; ok {
+		switch dydb.enableStreaming {
+		case "yes", "true":
+			dydb.enableStreaming = "yes"
+		case "no", "false":
+			dydb.enableStreaming = "no"
+		default:
+			return dydb, fmt.Errorf("please specify a valid choice for enableStreaming: available choices(yes, no, true, false)")
+		}
+	}
 	return dydb, nil
 }
 
@@ -377,6 +388,9 @@ func NewSourceProfileConnection(source string, params map[string]string) (Source
 			if err != nil {
 				return conn, err
 			}
+			if conn.Dydb.enableStreaming == "yes" {
+				conn.Streaming = true
+			}
 		}
 
 	case "sqlserver", "mssql":
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..178b502 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,33 @@ func (isi InfoSchemaImpl) ProcessData(conv *internal.Conv, srcTable string, srcS
 	return nil
 }
 
+// StartChangeDataCapture initializes the DynamoDB Streams for the source database. It
+// returns the latestStreamArn for all tables in the source database.
+func (isi InfoSchemaImpl) StartChangeDataCapture(ctx context.Context, conv *internal.Conv) (map[string]interface{}, error) {
+	fmt.Println("Starting DynamoDB Streams initialization...")
+
+	latestStreamArn := make(map[string]interface{})
+	orderTableNames := ddl.OrderTables(conv.SpSchema)
+
+	for _, spannerTable := range orderTableNames {
+		srcTable, _ := internal.GetSourceTable(conv, spannerTable)
+		streamArn, err := NewDynamoDBStream(isi.DynamoClient, srcTable)
+		if err != nil {
+			conv.Unexpected(fmt.Sprintf("Couldn't initialize DynamoDB Stream for table %s: %s", srcTable, err))
+			continue
+		}
+		latestStreamArn[srcTable] = streamArn
+	}
+
+	fmt.Println("DynamoDB Streams initialized successfully.")
+	return latestStreamArn, nil
+}
+
+func (isi InfoSchemaImpl) StartStreamingMigration(ctx context.Context, client *sp.Client, conv *internal.Conv, latestStreamArn map[string]interface{}) error {
+	// TODO(nareshz): work in progress
+	return nil
+}
+
 func getSchemaIndexStruct(indexName string, keySchema []*dynamodb.KeySchemaElement) schema.Index {
 	var keys []schema.Key
 	for _, j := range keySchema {
diff --git a/sources/dynamodb/streaming.go b/sources/dynamodb/streaming.go
new file mode 100644
index 0000000..e2bb5eb
--- /dev/null
+++ b/sources/dynamodb/streaming.go
@@ -0,0 +1,60 @@
+// Copyright 2022 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+package dynamodb
+
+import (
+	"fmt"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/dynamodb"
+	"github.com/aws/aws-sdk-go/service/dynamodb/dynamodbiface"
+)
+
+// NewDynamoDBStream initializes a new DynamoDB Stream for a table with NEW_AND_OLD_IMAGES
+// StreamViewType. If there exists a stream for a given table then it must be of type
+// NEW_IMAGE or NEW_AND_OLD_IMAGES otherwise streaming changes for this table won't be captured.
+// It returns latest Stream Arn for the table along with any error if encountered.
+func NewDynamoDBStream(client dynamodbiface.DynamoDBAPI, srcTable string) (string, error) {
+	describeTableInput := &dynamodb.DescribeTableInput{
+		TableName: aws.String(srcTable),
+	}
+	result, err := client.DescribeTable(describeTableInput)
+	if err != nil {
+		return "", fmt.Errorf("unexpected call to DescribeTable: %v", err)
+	}
+	if result.Table.StreamSpecification != nil {
+		switch *result.Table.StreamSpecification.StreamViewType {
+		case dynamodb.StreamViewTypeKeysOnly:
+			return "", fmt.Errorf("error! there exists a stream with KEYS_ONLY StreamViewType")
+		case dynamodb.StreamViewTypeOldImage:
+			return "", fmt.Errorf("error! there exists a stream with OLD_IMAGE StreamViewType")
+		default:
+			return *result.Table.LatestStreamArn, nil
+		}
+	} else {
+		streamSpecification := &dynamodb.StreamSpecification{
+			StreamEnabled:  aws.Bool(true),
+			StreamViewType: aws.String(dynamodb.StreamViewTypeNewAndOldImages),
+		}
+		updateTableInput := &dynamodb.UpdateTableInput{
+			StreamSpecification: streamSpecification,
+			TableName:           aws.String(srcTable),
+		}
+		res, err := client.UpdateTable(updateTableInput)
+		if err != nil {
+			return "", fmt.Errorf("unexpected call to UpdateTable: %v", err)
+		}
+		return *res.TableDescription.LatestStreamArn, nil
+	}
+}
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
+}
diff --git a/web/routes.go b/web/routes.go
index 677a17a..679428d 100644
--- a/web/routes.go
+++ b/web/routes.go
@@ -15,14 +15,19 @@
 package web
 
 import (
+	"embed"
+	"io/fs"
 	"net/http"
 
 	"github.com/gorilla/mux"
 )
 
+var FrontendDir embed.FS
+
 func getRoutes() *mux.Router {
 	router := mux.NewRouter().StrictSlash(true)
-	staticFileDirectory := http.Dir("./frontend/")
+	frontendRoot, _ := fs.Sub(FrontendDir, "frontend")
+	frontendStatic := http.FileServer(http.FS(frontendRoot))
 	router.HandleFunc("/connect", databaseConnection).Methods("POST")
 	router.HandleFunc("/convert/infoschema", convertSchemaSQL).Methods("GET")
 	router.HandleFunc("/convert/dump", convertSchemaDump).Methods("POST")
@@ -49,6 +54,6 @@ func getRoutes() *mux.Router {
 	router.HandleFunc("/rename/indexes", renameIndexes).Methods("POST")
 	router.HandleFunc("/add/indexes", addIndexes).Methods("POST")
 
-	router.PathPrefix("/").Handler(http.FileServer(staticFileDirectory))
+	router.PathPrefix("/").Handler(frontendStatic)
 	return router
 }


Test patch:
diff --git a/sources/dynamodb/schema_test.go b/sources/dynamodb/schema_test.go
index 4f7a00f..a41276b 100644
--- a/sources/dynamodb/schema_test.go
+++ b/sources/dynamodb/schema_test.go
@@ -15,17 +15,22 @@
 package dynamodb
 
 import (
+	"context"
 	"fmt"
 	"math/big"
+	"reflect"
 	"testing"
 
+	"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"
+	"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 {
@@ -35,6 +40,8 @@ type mockDynamoClient struct {
 	describeTableOutputs   []dynamodb.DescribeTableOutput
 	scanCallCount          int
 	scanOutputs            []dynamodb.ScanOutput
+	updateTableCallCount   int
+	updateTableOutputs     []dynamodb.UpdateTableOutput
 	dynamodbiface.DynamoDBAPI
 }
 
@@ -62,6 +69,14 @@ func (m *mockDynamoClient) Scan(input *dynamodb.ScanInput) (*dynamodb.ScanOutput
 	return &m.scanOutputs[m.scanCallCount-1], nil
 }
 
+func (m *mockDynamoClient) UpdateTable(input *dynamodb.UpdateTableInput) (*dynamodb.UpdateTableOutput, error) {
+	if m.updateTableCallCount >= len(m.updateTableOutputs) {
+		return nil, fmt.Errorf("unexpected call to UpdateTable: %v", input)
+	}
+	m.updateTableCallCount++
+	return &m.updateTableOutputs[m.updateTableCallCount-1], nil
+}
+
 func TestProcessSchema(t *testing.T) {
 	tableNameA := "test_a"
 	tableNameB := "test_b"
@@ -167,7 +182,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 +285,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 +368,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 +547,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 +602,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 +623,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 +633,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 +677,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 +695,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 +718,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 +750,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 +786,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,8 +921,145 @@ 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])
 }
+
+func TestInfoSchemaImpl_StartChangeDataCapture(t *testing.T) {
+	tableName := "testtable"
+	attrNameA := "a"
+	latestStreamArn := "arn:aws:dynamodb:dydb_endpoint:test_stream"
+
+	cols := []string{attrNameA}
+	spSchema := ddl.CreateTable{
+		Name:     tableName,
+		ColNames: cols,
+		ColDefs: map[string]ddl.ColumnDef{
+			attrNameA: {Name: attrNameA, T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
+		},
+		Pks: []ddl.IndexKey{{Col: attrNameA}},
+	}
+	srcTable := schema.Table{
+		Name:     tableName,
+		ColNames: cols,
+		ColDefs: map[string]schema.Column{
+			attrNameA: {Name: attrNameA, Type: schema.Type{Name: typeString}},
+		},
+		PrimaryKeys: []schema.Key{{Column: attrNameA}},
+	}
+	conv := buildConv(spSchema, srcTable)
+	type fields struct {
+		DynamoClient        dynamodbiface.DynamoDBAPI
+		DynamoStreamsClient *dynamodbstreams.DynamoDBStreams
+		SampleSize          int64
+	}
+	type args struct {
+		ctx  context.Context
+		conv *internal.Conv
+	}
+	arguments := args{
+		ctx:  context.Background(),
+		conv: conv,
+	}
+	tests := []struct {
+		name    string
+		fields  fields
+		args    args
+		want    map[string]interface{}
+		wantErr bool
+	}{
+		{
+			name: "test for checking correctness of output when stream exists already",
+			fields: fields{
+				DynamoClient: &mockDynamoClient{
+					describeTableOutputs: []dynamodb.DescribeTableOutput{
+						{
+							Table: &dynamodb.TableDescription{
+								TableName: &tableName,
+								StreamSpecification: &dynamodb.StreamSpecification{
+									StreamEnabled:  aws.Bool(true),
+									StreamViewType: aws.String(dynamodb.StreamViewTypeNewImage),
+								},
+								LatestStreamArn: &latestStreamArn,
+							},
+						},
+					},
+				},
+			},
+			args: arguments,
+			want: map[string]interface{}{
+				tableName: latestStreamArn,
+			},
+			wantErr: false,
+		},
+		{
+			name: "test for checking correctness of output when a new stream is created",
+			fields: fields{
+				DynamoClient: &mockDynamoClient{
+					describeTableOutputs: []dynamodb.DescribeTableOutput{
+						{
+							Table: &dynamodb.TableDescription{
+								TableName: &tableName,
+							},
+						},
+					},
+					updateTableOutputs: []dynamodb.UpdateTableOutput{
+						{
+							TableDescription: &dynamodb.TableDescription{
+								LatestStreamArn: &latestStreamArn,
+								StreamSpecification: &dynamodb.StreamSpecification{
+									StreamEnabled:  aws.Bool(true),
+									StreamViewType: aws.String(dynamodb.StreamViewTypeNewAndOldImages),
+								},
+							},
+						},
+					},
+				},
+			},
+			args: arguments,
+			want: map[string]interface{}{
+				tableName: latestStreamArn,
+			},
+			wantErr: false,
+		},
+		{
+			name: "test for handling api calls failure",
+			fields: fields{
+				DynamoClient: &mockDynamoClient{
+					describeTableOutputs: []dynamodb.DescribeTableOutput{
+						{
+							Table: &dynamodb.TableDescription{
+								TableName: &tableName,
+							},
+						},
+					},
+				},
+			},
+			args:    arguments,
+			want:    map[string]interface{}{},
+			wantErr: false,
+		},
+	}
+	totalUnexpecteds := int64(0)
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			isi := InfoSchemaImpl{
+				DynamoClient:        tt.fields.DynamoClient,
+				DynamoStreamsClient: tt.fields.DynamoStreamsClient,
+				SampleSize:          tt.fields.SampleSize,
+			}
+			got, err := isi.StartChangeDataCapture(tt.args.ctx, tt.args.conv)
+			if (err != nil) != tt.wantErr {
+				t.Errorf("InfoSchemaImpl.StartChangeDataCapture() error = %v, wantErr %v", err, tt.wantErr)
+				return
+			}
+			if !reflect.DeepEqual(got, tt.want) {
+				t.Errorf("InfoSchemaImpl.StartChangeDataCapture() = %v, want %v", got, tt.want)
+			}
+			totalUnexpecteds += conv.Unexpecteds()
+		})
+	}
+	assert.Equal(t, int64(1), totalUnexpecteds)
+}
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": {