Repository: cloudspannerecosystem/harbourbridge
Pull request #313 (2022-06-03 16:47:49)

Added best practices to readme
Added best practices section to Data migration

- [x] Tests pass
- [x] Appropriate changes to README are included in PR


Patch:
diff --git a/Makefile b/Makefile
new file mode 100644
index 0000000..ad832ec
--- /dev/null
+++ b/Makefile
@@ -0,0 +1,13 @@
+# Build the default binary
+build:
+	go build
+# Build a static binary
+build-static:
+	go build -a -tags osusergo,netgo -ldflags '-w -extldflags "-static"' -o harbourbridge main.go
+# Run unit tests
+test:
+	go test -v ./...
+# Run code coverage with unit tests
+test-coverage:
+	go test ./... -coverprofile coverage.out -covermode count
+	go tool cover -func coverage.out
diff --git a/README.md b/README.md
index 731f5b8..19cb6c4 100644
--- a/README.md
+++ b/README.md
@@ -398,12 +398,13 @@ Details on HarbourBridge schema conversion can be found here:
 - [SQL Server schema conversion](sources/sqlserver/README.md#schema-conversion)
 - [Oracle DB schema conversion](sources/oracle/README.md#schema-conversion)
 
-## Data Conversion
+## Data Migration
 
-HarbourBridge converts PostgreSQL/MySQL/DynamoDB data to Spanner data based on
+### Data Conversion
+
+HarbourBridge converts data from the source to Spanner data based on
 the Spanner schema it constructs. Conversion for most data types is fairly
-straightforward, but several types deserve discussion. Note that HarbourBridge
-is not intended for databases larger than about 100GB. Details on HarbourBridge
+straightforward, but several types deserve discussion. Details on HarbourBridge
 data conversion can be found here:
 
 - [PostgreSQL data conversion](sources/postgres/README.md#data-conversion)
@@ -412,6 +413,12 @@ data conversion can be found here:
 - [CSV data conversion](sources/csv/README.md#example-csv-usage)
 - [SQL Server data conversion](sources/sqlserver/README.md#data-conversion)
 
+### Data Migration Recommendations
+- We recommend to use this data migration solution only for small databases 
+(smaller than 100GB) without strict downtime requirements. 
+- While using direct connect, it is recommended to use a secondary/read replica
+ to ensure consistency and that and avoid impact from the load on the primary.
+
 ## Troubleshooting Guide
 
 The following steps can help diagnose common issues encountered while running
@@ -551,3 +558,5 @@ encountered.
 
 If you are having problems with HarbourBridge, please [submit an
 issue](https://github.com/cloudspannerecosystem/harbourbridge/issues).
+
+
diff --git a/cmd/cmd.go b/cmd/cmd.go
index 8f5079c..a0c59a9 100644
--- a/cmd/cmd.go
+++ b/cmd/cmd.go
@@ -89,11 +89,11 @@ func CommandLine(ctx context.Context, driver, targetDb, dbURI string, dataOnly,
 	}
 
 	// Populate migration request id and migration type in conv object
-	conv.MigrationRequestId = "HB-" + uuid.New().String()
+	conv.Audit.MigrationRequestId = "HB-" + uuid.New().String()
 	if dataOnly {
-		conv.MigrationType = migration.MigrationData_DATA_ONLY.Enum()
+		conv.Audit.MigrationType = migration.MigrationData_DATA_ONLY.Enum()
 	} else {
-		conv.MigrationType = migration.MigrationData_SCHEMA_AND_DATA.Enum()
+		conv.Audit.MigrationType = migration.MigrationData_SCHEMA_AND_DATA.Enum()
 	}
 
 	adminClient, err := utils.NewDatabaseAdminClient(ctx)
diff --git a/cmd/data.go b/cmd/data.go
index 2773145..b602035 100644
--- a/cmd/data.go
+++ b/cmd/data.go
@@ -154,8 +154,8 @@ func (cmd *DataCmd) Execute(ctx context.Context, f *flag.FlagSet, _ ...interface
 	defer adminClient.Close()
 
 	// Populate migration request id and migration type in conv object
-	conv.MigrationRequestId = "HB-" + uuid.New().String()
-	conv.MigrationType = migration.MigrationData_DATA_ONLY.Enum()
+	conv.Audit.MigrationRequestId = "HB-" + uuid.New().String()
+	conv.Audit.MigrationType = migration.MigrationData_DATA_ONLY.Enum()
 
 	if !sourceProfile.UseTargetSchema() {
 		err = conversion.CreateOrUpdateDatabase(ctx, adminClient, dbURI, sourceProfile.Driver, targetProfile.TargetDb, conv, ioHelper.Out)
@@ -168,18 +168,20 @@ func (cmd *DataCmd) Execute(ctx context.Context, f *flag.FlagSet, _ ...interface
 	dataCoversionStartTime := time.Now()
 	bw, err := conversion.DataConv(ctx, sourceProfile, targetProfile, &ioHelper, client, conv, true, cmd.writeLimit)
 	if err != nil {
-		err = fmt.Errorf("can't finish data conversion for db %s: %v", dbURI, 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)
 			return subcommands.ExitFailure
 		}
 	}
-	dataCoversionEndTime := time.Now()
-	dataCoversionDuration := dataCoversionEndTime.Sub(dataCoversionStartTime)
-	conv.DataConversionDuration = dataCoversionDuration
+
 	banner := utils.GetBanner(now, dbURI)
 	conversion.Report(sourceProfile.Driver, bw.DroppedRowsByTable(), ioHelper.BytesRead, banner, conv, cmd.filePrefix+reportFile, ioHelper.Out)
 	conversion.WriteBadData(bw, conv, banner, cmd.filePrefix+badDataFile, ioHelper.Out)
diff --git a/cmd/schema.go b/cmd/schema.go
index 36565eb..eb02082 100644
--- a/cmd/schema.go
+++ b/cmd/schema.go
@@ -121,7 +121,7 @@ func (cmd *SchemaCmd) Execute(ctx context.Context, f *flag.FlagSet, _ ...interfa
 	}
 
 	schemaCoversionEndTime := time.Now()
-	conv.SchemaConversionDuration = schemaCoversionEndTime.Sub(schemaConversionStartTime)
+	conv.Audit.SchemaConversionDuration = schemaCoversionEndTime.Sub(schemaConversionStartTime)
 	conversion.WriteSchemaFile(conv, schemaCoversionEndTime, cmd.filePrefix+schemaFile, ioHelper.Out)
 	conversion.WriteSessionFile(conv, cmd.filePrefix+sessionFile, ioHelper.Out)
 	conversion.Report(sourceProfile.Driver, nil, ioHelper.BytesRead, "", conv, cmd.filePrefix+reportFile, ioHelper.Out)
diff --git a/cmd/schema_and_data.go b/cmd/schema_and_data.go
index 056c71b..afacfb9 100644
--- a/cmd/schema_and_data.go
+++ b/cmd/schema_and_data.go
@@ -125,10 +125,12 @@ func (cmd *SchemaAndDataCmd) Execute(ctx context.Context, f *flag.FlagSet, _ ...
 	if err != nil {
 		panic(err)
 	}
+	schemaCoversionEndTime := time.Now()
+	conv.Audit.SchemaConversionDuration = schemaCoversionEndTime.Sub(schemaConversionStartTime)
 
 	// Populate migration request id and migration type in conv object
-	conv.MigrationRequestId = "HB-" + uuid.New().String()
-	conv.MigrationType = migration.MigrationData_SCHEMA_AND_DATA.Enum()
+	conv.Audit.MigrationRequestId = "HB-" + uuid.New().String()
+	conv.Audit.MigrationType = migration.MigrationData_SCHEMA_AND_DATA.Enum()
 
 	conversion.WriteSchemaFile(conv, schemaConversionStartTime, cmd.filePrefix+schemaFile, ioHelper.Out)
 	conversion.WriteSessionFile(conv, cmd.filePrefix+sessionFile, ioHelper.Out)
@@ -163,21 +165,22 @@ func (cmd *SchemaAndDataCmd) Execute(ctx context.Context, f *flag.FlagSet, _ ...
 		return subcommands.ExitFailure
 	}
 
-	schemaCoversionEndTime := time.Now()
-	conv.SchemaConversionDuration = schemaCoversionEndTime.Sub(schemaConversionStartTime)
+	dataCoversionStartTime := time.Now()
 	bw, err := conversion.DataConv(ctx, sourceProfile, targetProfile, &ioHelper, client, conv, true, cmd.writeLimit)
 	if err != nil {
-		err = fmt.Errorf("can't finish data conversion for db %s: %v", dbURI, 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)
 			return subcommands.ExitFailure
 		}
 	}
-	dataCoversionEndTime := time.Now()
-	conv.DataConversionDuration = dataCoversionEndTime.Sub(schemaCoversionEndTime)
 	banner := utils.GetBanner(schemaConversionStartTime, dbURI)
 	conversion.Report(sourceProfile.Driver, bw.DroppedRowsByTable(), ioHelper.BytesRead, banner, conv, cmd.filePrefix+reportFile, ioHelper.Out)
 	conversion.WriteBadData(bw, conv, banner, cmd.filePrefix+badDataFile, ioHelper.Out)
diff --git a/common/metrics/metrics.go b/common/metrics/metrics.go
index acdab3f..46eeeef 100644
--- a/common/metrics/metrics.go
+++ b/common/metrics/metrics.go
@@ -14,12 +14,12 @@ import (
 func GetMigrationData(conv *internal.Conv, driver, targetDb, typeOfConv string) *migration.MigrationData {
 
 	migrationData := migration.MigrationData{
-		MigrationRequestId: &conv.MigrationRequestId,
+		MigrationRequestId: &conv.Audit.MigrationRequestId,
 	}
 	if typeOfConv == constants.DataConv {
 		return &migrationData
 	}
-	migrationData.MigrationType = conv.MigrationType
+	migrationData.MigrationType = conv.Audit.MigrationType
 	migrationData.SourceConnectionMechanism, migrationData.Source = getMigrationDataSourceDetails(driver, &migrationData)
 	migrationData.SchemaPatterns = getMigrationDataSchemaPatterns(conv, &migrationData)
 
diff --git a/common/utils/utils.go b/common/utils/utils.go
index 97185a9..4081b2d 100644
--- a/common/utils/utils.go
+++ b/common/utils/utils.go
@@ -249,10 +249,10 @@ func GetPassword() string {
 
 // GetDatabaseName generates database name with driver_date prefix.
 func GetDatabaseName(driver string, now time.Time) (string, error) {
-	return generateName(fmt.Sprintf("%s_%s", driver, now.Format("2006-01-02")))
+	return GenerateName(fmt.Sprintf("%s_%s", driver, now.Format("2006-01-02")))
 }
 
-func generateName(prefix string) (string, error) {
+func GenerateName(prefix string) (string, error) {
 	b := make([]byte, 4)
 	_, err := rand.Read(b)
 	if err != nil {
diff --git a/conversion/conversion.go b/conversion/conversion.go
index 2f2dd7b..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"
@@ -74,14 +75,7 @@ var (
 const migrationMetadataKey = "cloud-spanner-migration-metadata"
 
 // SchemaConv performs the schema conversion
-// TODO: Pass around cmd.SourceProfile instead of sqlConnectionStr and schemaSampleSize.
-// Doing that requires refactoring since that would introduce a circular dependency between
-// conversion.go and cmd/source_profile.go.
-// The sqlConnectionStr param provides the connection details to use the go SQL library.
-// It is empty in the following cases:
-//  - Driver is DynamoDB or a dump file mode.
-//  - This function is called as part of the legacy global CLI flag mode. (This string is constructed from env variables later on)
-// When using source-profile, the sqlConnectionStr is constructed from the input params.
+// The SourceProfile param provides the connection details to use the go SQL library.
 func SchemaConv(sourceProfile profiles.SourceProfile, targetProfile profiles.TargetProfile, ioHelper *utils.IOStreams) (*internal.Conv, error) {
 	switch sourceProfile.Driver {
 	case constants.POSTGRES, constants.MYSQL, constants.DYNAMODB, constants.SQLSERVER, constants.ORACLE:
@@ -94,11 +88,7 @@ func SchemaConv(sourceProfile profiles.SourceProfile, targetProfile profiles.Tar
 }
 
 // DataConv performs the data conversion
-// The sqlConnectionStr param provides the connection details to use the go SQL library.
-// It is empty in the following cases:
-//  - Driver is DynamoDB or a dump file mode.
-//  - This function is called as part of the legacy global CLI flag mode. (This string is constructed from env variables later on)
-// When using source-profile, the sqlConnectionStr and schemaSampleSize are constructed from the input params.
+// The SourceProfile param provides the connection details to use the go SQL library.
 func DataConv(ctx context.Context, sourceProfile profiles.SourceProfile, targetProfile profiles.TargetProfile, ioHelper *utils.IOStreams, client *sp.Client, conv *internal.Conv, dataOnly bool, writeLimit int64) (*writer.BatchWriter, error) {
 	config := writer.BatchWriterConfig{
 		BytesLimit: 100 * 1000 * 1000,
@@ -108,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")
@@ -177,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)
@@ -198,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")
@@ -771,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
@@ -784,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 {
@@ -794,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))
@@ -808,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 6146bb9..df818ca 100644
--- a/go.mod
+++ b/go.mod
@@ -1,10 +1,11 @@
 module github.com/cloudspannerecosystem/harbourbridge
 
-go 1.13
+go 1.16
 
 require (
 	cloud.google.com/go v0.100.2
-	cloud.google.com/go/compute v1.5.0 // indirect
+	cloud.google.com/go/dataflow v0.4.0
+	cloud.google.com/go/datastream v0.4.0
 	cloud.google.com/go/iam v0.3.0 // indirect
 	cloud.google.com/go/spanner v1.30.0
 	cloud.google.com/go/storage v1.21.0
@@ -32,11 +33,10 @@ require (
 	github.com/sijms/go-ora/v2 v2.2.17
 	github.com/stretchr/testify v1.7.0
 	golang.org/x/crypto v0.0.0-20220214200702-86341886e292
-	golang.org/x/net v0.0.0-20220225172249-27dd8689420f
-	golang.org/x/oauth2 v0.0.0-20220223155221-ee480838109b // indirect
+	golang.org/x/net v0.0.0-20220325170049-de3da57026de
 	golang.org/x/tools v0.1.11-0.20220316014157-77aa08bb151a // indirect
-	google.golang.org/api v0.70.0
-	google.golang.org/genproto v0.0.0-20220304144024-325a89244dc8
-	google.golang.org/grpc v1.44.0
-	google.golang.org/protobuf v1.27.1
+	google.golang.org/api v0.74.0
+	google.golang.org/genproto v0.0.0-20220407144326-9054f6ed7bac
+	google.golang.org/grpc v1.45.0
+	google.golang.org/protobuf v1.28.0
 )
diff --git a/go.sum b/go.sum
index 3b88c35..ddd8a83 100644
--- a/go.sum
+++ b/go.sum
@@ -40,8 +40,12 @@ cloud.google.com/go/compute v1.2.0/go.mod h1:xlogom/6gr8RJGBe7nT2eGsQYAFUbbv8dbC
 cloud.google.com/go/compute v1.3.0/go.mod h1:cCZiE1NHEtai4wiufUhW8I8S1JKkAnhnQJWM7YD99wM=
 cloud.google.com/go/compute v1.5.0 h1:b1zWmYuuHz7gO9kDcM/EpHGr06UgsYNRpNJzI2kFiLM=
 cloud.google.com/go/compute v1.5.0/go.mod h1:9SMHyhJlzhlkJqrPAc839t2BZFTSk6Jdj6mkzQJeu0M=
+cloud.google.com/go/dataflow v0.4.0 h1:7P664mAaW1fbKhYLgc8lvB4KpVOpzOfQ8Z/vqO+IwU4=
+cloud.google.com/go/dataflow v0.4.0/go.mod h1:b9jjQUJRNB1PLaMcyWHPkONsJCb7cJstgFYDlAWU164=
 cloud.google.com/go/datastore v1.0.0/go.mod h1:LXYbyblFSglQ5pkeyhO+Qmw7ukd3C+pD7TKLgZqpHYE=
 cloud.google.com/go/datastore v1.1.0/go.mod h1:umbIZjpQpHh4hmRpGhH4tLFup+FVzqBi1b3c64qFpCk=
+cloud.google.com/go/datastream v0.4.0 h1:v3MBGwQgr//lAHV1AAKX2XST3DYD0MjpcAiKhEZ/2IE=
+cloud.google.com/go/datastream v0.4.0/go.mod h1:pd1ABa+05JkJbF8AdMqWGtbUx23o+9OhYRqpTzJ4x98=
 cloud.google.com/go/firestore v1.1.0/go.mod h1:ulACoGHTpvq5r8rxGJ4ddJZBZqakUQqClKRT5SZwBmk=
 cloud.google.com/go/iam v0.1.1/go.mod h1:CKqrcnI/suGpybEHxZ7BMehL0oA4LpdyJdUlTl9jVMw=
 cloud.google.com/go/iam v0.3.0 h1:exkAomrVUuzx9kWFI1wm3KI0uoDeUFPB4kKGzx6x+Gc=
@@ -413,8 +417,10 @@ github.com/google/uuid v1.1.2/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+
 github.com/googleapis/gax-go/v2 v2.0.4/go.mod h1:0Wqv26UfaUD9n4G6kQubkQ+KchISgw+vpHVxEJEs9eg=
 github.com/googleapis/gax-go/v2 v2.0.5/go.mod h1:DWXyrwAJ9X0FpwwEdw+IPEYBICEFu5mhpdKc/us6bOk=
 github.com/googleapis/gax-go/v2 v2.1.0/go.mod h1:Q3nei7sK6ybPYH7twZdmQpAd1MKb7pfu6SK+H1/DsU0=
-github.com/googleapis/gax-go/v2 v2.1.1 h1:dp3bWCh+PPO1zjRRiCSczJav13sBvG4UhNyVTa1KqdU=
 github.com/googleapis/gax-go/v2 v2.1.1/go.mod h1:hddJymUZASv3XPyGkUpKj8pPO47Rmb0eJc8R6ouapiM=
+github.com/googleapis/gax-go/v2 v2.2.0/go.mod h1:as02EH8zWkzwUoLbBaFeQ+arQaj/OthfcblKl4IGNaM=
+github.com/googleapis/gax-go/v2 v2.3.0 h1:nRJtk3y8Fm770D42QV6T90ZnvFZyk7agSo3Q+Z9p3WI=
+github.com/googleapis/gax-go/v2 v2.3.0/go.mod h1:b8LNqSzNabLiUpXKkY7HAR5jr6bIT99EXz9pXxye9YM=
 github.com/gopherjs/gopherjs v0.0.0-20181017120253-0766667cb4d1/go.mod h1:wJfORRmW1u3UXTncJ5qlYoELFm8eSnnEO6hX4iZ3EWY=
 github.com/gorilla/handlers v1.4.2/go.mod h1:Qkdc/uu4tH4g6mTK6auzZ766c4CA0Ng8+o/OAirnOIQ=
 github.com/gorilla/handlers v1.5.1 h1:9lRY6j8DEeeBT10CvO9hGW0gmky0BprnvDI5vfhUHH4=
@@ -508,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=
@@ -565,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=
@@ -573,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=
@@ -927,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=
@@ -1033,8 +1045,9 @@ golang.org/x/net v0.0.0-20210813160813-60bc85c4be6d/go.mod h1:9nx3DQGgdP8bBQD5qx
 golang.org/x/net v0.0.0-20211015210444-4f30a5c0130f/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
 golang.org/x/net v0.0.0-20211112202133-69e39bad7dc2/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
 golang.org/x/net v0.0.0-20220127200216-cd36cc0744dd/go.mod h1:CfG3xpIq0wQ8r1q4Su4UZFWDARRcnwPjda9FqA0JpMk=
-golang.org/x/net v0.0.0-20220225172249-27dd8689420f h1:oA4XRj0qtSt8Yo1Zms0CUlsT3KG69V2UGQWPBxujDmc=
 golang.org/x/net v0.0.0-20220225172249-27dd8689420f/go.mod h1:CfG3xpIq0wQ8r1q4Su4UZFWDARRcnwPjda9FqA0JpMk=
+golang.org/x/net v0.0.0-20220325170049-de3da57026de h1:pZB1TWnKi+o4bENlbzAgLrEbY4RMYmUIRobMcSmfeYc=
+golang.org/x/net v0.0.0-20220325170049-de3da57026de/go.mod h1:CfG3xpIq0wQ8r1q4Su4UZFWDARRcnwPjda9FqA0JpMk=
 golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
 golang.org/x/oauth2 v0.0.0-20190226205417-e64efc72b421/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw=
 golang.org/x/oauth2 v0.0.0-20190604053449-0f29369cfe45/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw=
@@ -1051,8 +1064,9 @@ golang.org/x/oauth2 v0.0.0-20210628180205-a41e5a781914/go.mod h1:KelEdhl1UZF7XfJ
 golang.org/x/oauth2 v0.0.0-20210805134026-6f1e6394065a/go.mod h1:KelEdhl1UZF7XfJ4dDtk6s++YSgaE7mD/BuKKDLBl4A=
 golang.org/x/oauth2 v0.0.0-20210819190943-2bc19b11175f/go.mod h1:KelEdhl1UZF7XfJ4dDtk6s++YSgaE7mD/BuKKDLBl4A=
 golang.org/x/oauth2 v0.0.0-20211104180415-d3ed0bb246c8/go.mod h1:KelEdhl1UZF7XfJ4dDtk6s++YSgaE7mD/BuKKDLBl4A=
-golang.org/x/oauth2 v0.0.0-20220223155221-ee480838109b h1:clP8eMhB30EHdc0bd2Twtq6kgU7yl5ub2cQLSdrv1Dg=
 golang.org/x/oauth2 v0.0.0-20220223155221-ee480838109b/go.mod h1:DAh4E804XQdzx2j+YRIaUnCqCV2RuMz24cGBJ5QYIrc=
+golang.org/x/oauth2 v0.0.0-20220309155454-6242fa91716a h1:qfl7ob3DIEs3Ml9oLuPwY2N04gymzAW04WsUQHIClgM=
+golang.org/x/oauth2 v0.0.0-20220309155454-6242fa91716a/go.mod h1:DAh4E804XQdzx2j+YRIaUnCqCV2RuMz24cGBJ5QYIrc=
 golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20181108010431-42b317875d0f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20181221193216-37e7f081c4d4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
@@ -1151,6 +1165,9 @@ golang.org/x/sys v0.0.0-20211216021012-1d35b9e2eb4e/go.mod h1:oPkhp1MJrh7nUepCBc
 golang.org/x/sys v0.0.0-20220114195835-da31bd327af9/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220128215802-99c3d69c2c27/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220209214540-3681064d5158/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.0.0-20220227234510-4e6760a101f9/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.0.0-20220310020820-b874c991c1a5/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.0.0-20220328115105-d36c6a25d886/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220408201424-a24fb2fb8a0f h1:8w7RhxzTVgUzw/AH/9mUV5q0vMgy40SQRursCcfmkCw=
 golang.org/x/sys v0.0.0-20220408201424-a24fb2fb8a0f/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
@@ -1288,8 +1305,11 @@ google.golang.org/api v0.64.0/go.mod h1:931CdxA8Rm4t6zqTFGSsgwbAEZ2+GMYurbndwSim
 google.golang.org/api v0.66.0/go.mod h1:I1dmXYpX7HGwz/ejRxwQp2qj5bFAz93HiCU1C1oYd9M=
 google.golang.org/api v0.67.0/go.mod h1:ShHKP8E60yPsKNw/w8w+VYaj9H6buA5UqDp8dhbQZ6g=
 google.golang.org/api v0.69.0/go.mod h1:boanBiw+h5c3s+tBPgEzLDRHfFLWV0qXxRHz3ws7C80=
-google.golang.org/api v0.70.0 h1:67zQnAE0T2rB0A3CwLSas0K+SbVzSxP+zTLkQLexeiw=
 google.golang.org/api v0.70.0/go.mod h1:Bs4ZM2HGifEvXwd50TtW70ovgJffJYw2oRCOFU/SkfA=
+google.golang.org/api v0.71.0/go.mod h1:4PyU6e6JogV1f9eA4voyrTY2batOLdgZ5qZ5HOCc4j8=
+google.golang.org/api v0.73.0/go.mod h1:lbd/q6BRFJbdpV6OUCXstVeiI5mL/d3/WifG7iNKnjI=
+google.golang.org/api v0.74.0 h1:ExR2D+5TYIrMphWgs5JCgwRhEDlPDXXrLwHHMgPHTXE=
+google.golang.org/api v0.74.0/go.mod h1:ZpfMZOVRMywNyvJFeqL9HRWBgAuRfSjJFpe9QtRRyDs=
 google.golang.org/appengine v1.1.0/go.mod h1:EbEs0AVv82hx2wNQdGPgUI5lhzA/G0D9YwlJXL52JkM=
 google.golang.org/appengine v1.4.0/go.mod h1:xpcJRLb0r/rnEns0DIKYYv+WjYCduHsrkT7/EB5XEv4=
 google.golang.org/appengine v1.5.0/go.mod h1:xpcJRLb0r/rnEns0DIKYYv+WjYCduHsrkT7/EB5XEv4=
@@ -1372,8 +1392,12 @@ google.golang.org/genproto v0.0.0-20220211171837-173942840c17/go.mod h1:kGP+zUP2
 google.golang.org/genproto v0.0.0-20220216160803-4663080d8bc8/go.mod h1:kGP+zUP2Ddo0ayMi4YuN7C3WZyJvGLZRh8Z5wnAqvEI=
 google.golang.org/genproto v0.0.0-20220218161850-94dd64e39d7c/go.mod h1:kGP+zUP2Ddo0ayMi4YuN7C3WZyJvGLZRh8Z5wnAqvEI=
 google.golang.org/genproto v0.0.0-20220222213610-43724f9ea8cf/go.mod h1:kGP+zUP2Ddo0ayMi4YuN7C3WZyJvGLZRh8Z5wnAqvEI=
-google.golang.org/genproto v0.0.0-20220304144024-325a89244dc8 h1:U9V52f6rAgINH7kT+musA1qF8kWyVOxzF8eYuOVuFwQ=
 google.golang.org/genproto v0.0.0-20220304144024-325a89244dc8/go.mod h1:kGP+zUP2Ddo0ayMi4YuN7C3WZyJvGLZRh8Z5wnAqvEI=
+google.golang.org/genproto v0.0.0-20220310185008-1973136f34c6/go.mod h1:kGP+zUP2Ddo0ayMi4YuN7C3WZyJvGLZRh8Z5wnAqvEI=
+google.golang.org/genproto v0.0.0-20220322021311-435b647f9ef2/go.mod h1:hAL49I2IFola2sVEjAn7MEwsja0xp51I0tlGAf9hz4E=
+google.golang.org/genproto v0.0.0-20220324131243-acbaeb5b85eb/go.mod h1:hAL49I2IFola2sVEjAn7MEwsja0xp51I0tlGAf9hz4E=
+google.golang.org/genproto v0.0.0-20220407144326-9054f6ed7bac h1:qSNTkEN+L2mvWcLgJOR+8bdHX9rN/IdU3A1Ghpfb1Rg=
+google.golang.org/genproto v0.0.0-20220407144326-9054f6ed7bac/go.mod h1:8w6bsBMX6yCPbAVTeqQHvzxW0EIFigd5lZyahWgyfDo=
 google.golang.org/grpc v0.0.0-20180607172857-7a6a684ca69e/go.mod h1:yo6s7OP7yaDglbqo1J04qKzAhqBH6lvTonzMVmEdcZw=
 google.golang.org/grpc v1.12.0/go.mod h1:yo6s7OP7yaDglbqo1J04qKzAhqBH6lvTonzMVmEdcZw=
 google.golang.org/grpc v1.19.0/go.mod h1:mqu4LbDTu4XGKhr4mRzUsmM4RtVoemTSY81AxZiDr8c=
@@ -1404,8 +1428,9 @@ google.golang.org/grpc v1.39.1/go.mod h1:PImNr+rS9TWYb2O4/emRugxiyHZ5JyHW5F+RPnD
 google.golang.org/grpc v1.40.0/go.mod h1:ogyxbiOoUXAkP+4+xa6PZSE9DZgIHtSpzjDTB9KAK34=
 google.golang.org/grpc v1.40.1/go.mod h1:ogyxbiOoUXAkP+4+xa6PZSE9DZgIHtSpzjDTB9KAK34=
 google.golang.org/grpc v1.43.0/go.mod h1:k+4IHHFw41K8+bbowsex27ge2rCb65oeWqe4jJ590SU=
-google.golang.org/grpc v1.44.0 h1:weqSxi/TMs1SqFRMHCtBgXRs8k3X39QIDEZ0pRcttUg=
 google.golang.org/grpc v1.44.0/go.mod h1:k+4IHHFw41K8+bbowsex27ge2rCb65oeWqe4jJ590SU=
+google.golang.org/grpc v1.45.0 h1:NEpgUqV3Z+ZjkqMsxMg11IaDrXY4RY6CQukSGK0uI1M=
+google.golang.org/grpc v1.45.0/go.mod h1:lN7owxKUQEqMfSyQikvvk5tf/6zMPsrK+ONuO11+0rQ=
 google.golang.org/grpc/cmd/protoc-gen-go-grpc v1.1.0/go.mod h1:6Kw0yEErY5E/yWrBtf03jp27GLLJujG4z/JK95pnjjw=
 google.golang.org/protobuf v0.0.0-20200109180630-ec00e32a8dfd/go.mod h1:DFci5gLYBciE7Vtevhsrf46CRTquxDuWsQurQQe4oz8=
 google.golang.org/protobuf v0.0.0-20200221191635-4d8936d0db64/go.mod h1:kwYJMbMJ01Woi6D6+Kah6886xMZcty6N08ah7+eCXa0=
@@ -1419,8 +1444,9 @@ google.golang.org/protobuf v1.24.0/go.mod h1:r/3tXBNzIEhYS9I1OUVjXDlt8tc493IdKGj
 google.golang.org/protobuf v1.25.0/go.mod h1:9JNX74DMeImyA3h4bdi1ymwjUzf21/xIlbajtzgsN7c=
 google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp09yW+WbY/TyQbw=
 google.golang.org/protobuf v1.26.0/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc=
-google.golang.org/protobuf v1.27.1 h1:SnqbnDw1V7RiZcXPx5MEeqPv2s79L9i7BJUlG/+RurQ=
 google.golang.org/protobuf v1.27.1/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc=
+google.golang.org/protobuf v1.28.0 h1:w43yiav+6bVFTBQFZX0r7ipe9JQ1QsbMgHwbBziscLw=
+google.golang.org/protobuf v1.28.0/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqwMG9pJV4I=
 gopkg.in/alecthomas/kingpin.v2 v2.2.6/go.mod h1:FMv+mEhP44yOT+4EoQTLFTRgOQ1FBLkstjWtayDeSgw=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
diff --git a/internal/convert.go b/internal/convert.go
index 48d14b2..7ef775b 100644
--- a/internal/convert.go
+++ b/internal/convert.go
@@ -25,26 +25,23 @@ import (
 
 // Conv contains all schema and data conversion state.
 type Conv struct {
-	mode                     mode                                // Schema mode or data mode.
-	SpSchema                 ddl.Schema                          // Maps Spanner table name to Spanner schema.
-	SyntheticPKeys           map[string]SyntheticPKey            // Maps Spanner table name to synthetic primary key (if needed).
-	SrcSchema                map[string]schema.Table             // Maps source-DB table name to schema information.
-	Issues                   map[string]map[string][]SchemaIssue // Maps source-DB table/col to list of schema conversion issues.
-	ToSpanner                map[string]NameAndCols              // Maps from source-DB table name to Spanner name and column mapping.
-	ToSource                 map[string]NameAndCols              // Maps from Spanner table name to source-DB table name and column mapping.
-	UsedNames                map[string]bool                     // Map storing the names that are already assigned to tables, indices or foreign key contraints.
-	dataSink                 func(table string, cols []string, values []interface{})
-	DataFlush                func()         `json:"-"` // Data flush is used to flush out remaining writes and wait for them to complete.
-	Location                 *time.Location // Timezone (for timestamp conversion).
-	sampleBadRows            rowSamples     // Rows that generated errors during conversion.
-	Stats                    stats
-	TimezoneOffset           string                                 // Timezone offset for timestamp conversion.
-	TargetDb                 string                                 // The target database to which HarbourBridge is writing.
-	UniquePKey               map[string][]string                    // Maps Spanner table name to unique column name being used as primary key (if needed).
-	SchemaConversionDuration time.Duration                          `json:"-"` // Duration of schema conversion.
-	DataConversionDuration   time.Duration                          `json:"-"` // Duration of data conversion.
-	MigrationRequestId       string                                 `json:"-"` // Unique request id generated per migration
-	MigrationType            *migration.MigrationData_MigrationType `json:"-"` // Type of migration: Schema migration, data migration or schema and data migration
+	mode           mode                                // Schema mode or data mode.
+	SpSchema       ddl.Schema                          // Maps Spanner table name to Spanner schema.
+	SyntheticPKeys map[string]SyntheticPKey            // Maps Spanner table name to synthetic primary key (if needed).
+	SrcSchema      map[string]schema.Table             // Maps source-DB table name to schema information.
+	Issues         map[string]map[string][]SchemaIssue // Maps source-DB table/col to list of schema conversion issues.
+	ToSpanner      map[string]NameAndCols              // Maps from source-DB table name to Spanner name and column mapping.
+	ToSource       map[string]NameAndCols              // Maps from Spanner table name to source-DB table name and column mapping.
+	UsedNames      map[string]bool                     // Map storing the names that are already assigned to tables, indices or foreign key contraints.
+	dataSink       func(table string, cols []string, values []interface{})
+	DataFlush      func()         `json:"-"` // Data flush is used to flush out remaining writes and wait for them to complete.
+	Location       *time.Location // Timezone (for timestamp conversion).
+	sampleBadRows  rowSamples     // Rows that generated errors during conversion.
+	Stats          stats
+	TimezoneOffset string              // Timezone offset for timestamp conversion.
+	TargetDb       string              // The target database to which HarbourBridge is writing.
+	UniquePKey     map[string][]string // Maps Spanner table name to unique column name being used as primary key (if needed).
+	Audit          audit               // Stores the audit information for the database conversion
 }
 
 type mode int
@@ -85,6 +82,7 @@ const (
 	Widened
 	Time
 	StringOverflow
+	IllegalName
 )
 
 // NameAndCols contains the name of a table and its columns.
@@ -94,6 +92,13 @@ type NameAndCols struct {
 	Cols map[string]string
 }
 
+//FkeyAndIdxs contains the name of a table, its foreign keys and indexes
+//Used to map between source DB and spanner table name, foreign key name and index names.
+type FkeyAndIdxs struct {
+	Name       string
+	ForeignKey map[string]string
+	Index      map[string]string
+}
 type rowSamples struct {
 	rows       []*row
 	bytes      int64 // Bytes consumed by l.
@@ -128,6 +133,17 @@ type statementStat struct {
 	Error  int64
 }
 
+// Stores the audit information of conversion.
+// Elements that do not affect the migration functionality but are relevant for the conversion report.
+type audit struct {
+	ToSpannerFkIdx           map[string]FkeyAndIdxs                 `json:"-"` // Maps from source-DB table name to Spanner names for table name, foreign key and indexes.
+	ToSourceFkIdx            map[string]FkeyAndIdxs                 `json:"-"` // Maps from Spanner table name to source-DB names for table name, foreign key and indexes.
+	SchemaConversionDuration time.Duration                          `json:"-"` // Duration of schema conversion.
+	DataConversionDuration   time.Duration                          `json:"-"` // Duration of data conversion.
+	MigrationRequestId       string                                 `json:"-"` // Unique request id generated per migration
+	MigrationType            *migration.MigrationData_MigrationType `json:"-"` // Type of migration: Schema migration, data migration or schema and data migration
+}
+
 // MakeConv returns a default-configured Conv.
 func MakeConv() *Conv {
 	return &Conv{
@@ -149,6 +165,10 @@ func MakeConv() *Conv {
 		},
 		TimezoneOffset: "+00:00", // By default, use +00:00 offset which is equal to UTC timezone
 		UniquePKey:     make(map[string][]string),
+		Audit: audit{
+			ToSpannerFkIdx: make(map[string]FkeyAndIdxs),
+			ToSourceFkIdx:  make(map[string]FkeyAndIdxs),
+		},
 	}
 }
 
diff --git a/internal/mapping.go b/internal/mapping.go
index 5bdbfc6..386176a 100644
--- a/internal/mapping.go
+++ b/internal/mapping.go
@@ -49,6 +49,8 @@ func GetSpannerTable(conv *Conv, srcTable string) (string, error) {
 	}
 	conv.ToSpanner[srcTable] = NameAndCols{Name: spTable, Cols: make(map[string]string)}
 	conv.ToSource[spTable] = NameAndCols{Name: srcTable, Cols: make(map[string]string)}
+	conv.Audit.ToSpannerFkIdx[srcTable] = FkeyAndIdxs{Name: spTable, ForeignKey: make(map[string]string), Index: make(map[string]string)}
+	conv.Audit.ToSourceFkIdx[spTable] = FkeyAndIdxs{Name: srcTable, ForeignKey: make(map[string]string), Index: make(map[string]string)}
 	return spTable, nil
 }
 
diff --git a/internal/report.go b/internal/report.go
index 2b27a23..c5ae82d 100644
--- a/internal/report.go
+++ b/internal/report.go
@@ -58,6 +58,8 @@ func GenerateReport(driverName string, conv *Conv, w *bufio.Writer, badWrites ma
 	if isDump {
 		writeStmtStats(driverName, conv, w)
 	}
+	reportNameChanges(conv, w)
+
 	if printTableReports {
 		for _, t := range reports {
 			h := fmt.Sprintf("Table %s", t.SrcTable)
@@ -76,6 +78,7 @@ func GenerateReport(driverName string, conv *Conv, w *bufio.Writer, badWrites ma
 			}
 		}
 	}
+
 	if printUnexpecteds {
 		writeUnexpectedConditions(driverName, conv, w)
 	}
@@ -172,6 +175,20 @@ func buildTableReportBody(conv *Conv, srcTable string, issues map[string][]Schem
 				l = append(l, fmt.Sprintf("UNIQUE constraint on column(s) '%s' replaced with primary key since this table didn't have one. Spanner requires a primary key for every table", strings.Join(uniquePK, ", ")))
 			}
 		}
+
+		if p.severity == note {
+			for srcKeyName, spKeyName := range conv.Audit.ToSpannerFkIdx[srcTable].ForeignKey {
+				if srcKeyName != spKeyName {
+					l = append(l, fmt.Sprintf("%s, Foreign Key '%s' is mapped to '%s'", IssueDB[IllegalName].Brief, srcKeyName, spKeyName))
+				}
+			}
+			for srcIdxName, spIdxName := range conv.Audit.ToSpannerFkIdx[srcTable].Index {
+				if srcIdxName != spIdxName {
+					l = append(l, fmt.Sprintf("%s, Index '%s' is mapped to '%s'", IssueDB[IllegalName].Brief, srcIdxName, spIdxName))
+				}
+			}
+		}
+
 		issueBatcher := make(map[SchemaIssue]bool)
 		for _, srcCol := range cols {
 			for _, i := range issues[srcCol] {
@@ -192,6 +209,9 @@ func buildTableReportBody(conv *Conv, srcTable string, issues map[string][]Schem
 				}
 				srcType := srcSchema.ColDefs[srcCol].Type.Print()
 				spType := spSchema.ColDefs[spCol].T.PrintColumnDefType()
+				srcName := srcSchema.ColDefs[srcCol].Name
+				spName := spSchema.ColDefs[spCol].Name
+
 				// A note on case: Spanner types are case insensitive, but
 				// default to upper case. In particular, the Spanner AST uses
 				// upper case, so spType is upper case. Many source DBs
@@ -215,6 +235,8 @@ func buildTableReportBody(conv *Conv, srcTable string, issues map[string][]Schem
 					l = append(l, fmt.Sprintf("Some columns have source DB type 'datetime' which is mapped to Spanner type timestamp e.g. column '%s'. %s", srcCol, IssueDB[i].Brief))
 				case Widened:
 					l = append(l, fmt.Sprintf("%s e.g. for column '%s', source DB type %s is mapped to Spanner type %s", IssueDB[i].Brief, srcCol, srcType, spType))
+				case IllegalName:
+					l = append(l, fmt.Sprintf("%s, Column '%s' is mapped to '%s'", IssueDB[i].Brief, srcName, spName))
 				default:
 					l = append(l, fmt.Sprintf("Column '%s': type %s is mapped to %s. %s", srcCol, srcType, spType, IssueDB[i].Brief))
 				}
@@ -223,6 +245,7 @@ func buildTableReportBody(conv *Conv, srcTable string, issues map[string][]Schem
 		if len(l) == 0 {
 			continue
 		}
+
 		heading := p.heading
 		if len(l) > 1 {
 			heading = heading + "s"
@@ -277,6 +300,7 @@ var IssueDB = map[SchemaIssue]struct {
 	Time:                  {Brief: "Spanner does not support time/year types", severity: note, batch: true},
 	Widened:               {Brief: "Some columns will consume more storage in Spanner", severity: note, batch: true},
 	StringOverflow:        {Brief: "String overflow issue might occur as maximum supported length in Spanner is 2621440", severity: warning},
+	IllegalName:           {Brief: "Names must adhere to the spanner regular expression {a-z|A-Z}[{a-z|A-Z|0-9|_}+]", severity: note},
 }
 
 type severity int
@@ -325,6 +349,7 @@ func AnalyzeCols(conv *Conv, srcTable, spTable string) (map[string][]SchemaIssue
 // summary rating.
 func rateSchema(cols, warnings int64, missingPKey, summary bool) string {
 	pkMsg := "missing primary key"
+	s := fmt.Sprintf(" (%s%% of %d columns mapped cleanly)", pct(cols, warnings), cols)
 	if summary {
 		pkMsg = "some missing primary keys"
 	}
@@ -332,21 +357,21 @@ func rateSchema(cols, warnings int64, missingPKey, summary bool) string {
 	case cols == 0:
 		return "NONE (no schema found)"
 	case warnings == 0 && !missingPKey:
-		return "EXCELLENT (all columns mapped cleanly)"
+		return fmt.Sprintf("EXCELLENT (all %d columns mapped cleanly)", cols)
 	case warnings == 0 && missingPKey:
 		return fmt.Sprintf("GOOD (all columns mapped cleanly, but %s)", pkMsg)
 	case good(cols, warnings) && !missingPKey:
-		return "GOOD (most columns mapped cleanly)"
+		return "GOOD" + s
 	case good(cols, warnings) && missingPKey:
-		return fmt.Sprintf("GOOD (most columns mapped cleanly, but %s)", pkMsg)
+		return "GOOD" + s + fmt.Sprintf(" + %s", pkMsg)
 	case ok(cols, warnings) && !missingPKey:
-		return "OK (some columns did not map cleanly)"
+		return "OK" + s
 	case ok(cols, warnings) && missingPKey:
-		return fmt.Sprintf("OK (some columns did not map cleanly + %s)", pkMsg)
+		return "OK" + s + fmt.Sprintf(" + %s", pkMsg)
 	case !missingPKey:
-		return "POOR (many columns did not map cleanly)"
+		return "POOR" + s
 	default:
-		return fmt.Sprintf("POOR (many columns did not map cleanly + %s)", pkMsg)
+		return "POOR" + s + fmt.Sprintf(" + %s", pkMsg)
 	}
 }
 
@@ -367,11 +392,11 @@ func rateData(rows int64, badRows int64) string {
 }
 
 func good(total, badCount int64) bool {
-	return badCount < total/20
+	return float64(badCount) < float64(total)/20
 }
 
 func ok(total, badCount int64) bool {
-	return badCount < total/3
+	return float64(badCount) < float64(total)/3
 }
 
 func rateConversion(rows, badRows, cols, warnings int64, missingPKey, summary bool, schemaOnly bool) string {
@@ -433,6 +458,38 @@ func IgnoredStatements(conv *Conv) (l []string) {
 	return l
 }
 
+func reportNameChanges(conv *Conv, w *bufio.Writer) {
+
+	w.WriteString("-----------------------------------------------------------------------------------------------------\n")
+	w.WriteString("Name Changes in Migration\n")
+	w.WriteString("-----------------------------------------------------------------------------------------------------\n")
+	fmt.Fprintf(w, "%25s %15s %25s %25s\n", "Source Table", "Change", "Old Name", "New Name")
+	w.WriteString("-----------------------------------------------------------------------------------------------------\n")
+
+	for srcTableName, spTable := range conv.ToSpanner {
+		if srcTableName != spTable.Name {
+			fmt.Fprintf(w, "%25s %15s %25s %25s\n", srcTableName, "Table Name", srcTableName, spTable.Name)
+		}
+		for srcColName, spColName := range spTable.Cols {
+			if srcColName != spColName {
+				fmt.Fprintf(w, "%25s %15s %25s %25s\n", srcTableName, "Column Name", srcColName, spColName)
+			}
+		}
+		for srcFkName, spFkName := range conv.Audit.ToSpannerFkIdx[srcTableName].ForeignKey {
+			if srcFkName != spFkName {
+				fmt.Fprintf(w, "%25s %15s %25s %25s\n", srcTableName, "Foreign Key", srcFkName, spFkName)
+			}
+		}
+		for srcIdxName, spIdxName := range conv.Audit.ToSpannerFkIdx[srcTableName].Index {
+			if srcIdxName != spIdxName {
+				fmt.Fprintf(w, "%25s %15s %25s %25s\n", srcTableName, "Index", srcIdxName, spIdxName)
+			}
+		}
+	}
+	w.WriteString("-----------------------------------------------------------------------------------------------------\n\n\n")
+
+}
+
 func writeStmtStats(driverName string, conv *Conv, w *bufio.Writer) {
 	type stat struct {
 		statement string
@@ -561,13 +618,13 @@ func writeHeading(w *bufio.Writer, s string) {
 
 func conversionDuration(conv *Conv, w *bufio.Writer) string {
 	res := ""
-	if conv.DataConversionDuration.Microseconds() != 0 || conv.SchemaConversionDuration.Microseconds() != 0 {
+	if conv.Audit.DataConversionDuration.Microseconds() != 0 || conv.Audit.SchemaConversionDuration.Microseconds() != 0 {
 		writeHeading(w, "Time duration of Conversion")
-		if conv.SchemaConversionDuration.Microseconds() != 0 {
-			res += fmt.Sprintf("Schema conversion duration : %s \n", conv.SchemaConversionDuration)
+		if conv.Audit.SchemaConversionDuration.Microseconds() != 0 {
+			res += fmt.Sprintf("Schema conversion duration : %s \n", conv.Audit.SchemaConversionDuration)
 		}
-		if conv.DataConversionDuration.Microseconds() != 0 {
-			res += fmt.Sprintf("Data conversion duration : %s \n", conv.DataConversionDuration)
+		if conv.Audit.DataConversionDuration.Microseconds() != 0 {
+			res += fmt.Sprintf("Data conversion duration : %s \n", conv.Audit.DataConversionDuration)
 		}
 		res += "\n"
 	}
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 7d85ae3..ffb7d78 100644
--- a/profiles/source_profile.go
+++ b/profiles/source_profile.go
@@ -66,20 +66,29 @@ const (
 )
 
 type SourceProfileConnectionMySQL struct {
-	Host string // Same as MYSQLHOST environment variable
-	Port string // Same as MYSQLPORT environment variable
-	User string // Same as MYSQLUSER environment variable
-	Db   string // Same as MYSQLDATABASE environment variable
-	Pwd  string // Same as MYSQLPWD environment variable
+	Host            string // Same as MYSQLHOST environment variable
+	Port            string // Same as MYSQLPORT environment variable
+	User            string // Same as MYSQLUSER environment variable
+	Db              string // Same as MYSQLDATABASE environment variable
+	Pwd             string // Same as MYSQLPWD environment variable
+	StreamingConfig string
 }
 
 func NewSourceProfileConnectionMySQL(params map[string]string) (SourceProfileConnectionMySQL, error) {
 	mysql := SourceProfileConnectionMySQL{}
+
 	host, hostOk := params["host"]
 	user, userOk := params["user"]
 	db, dbOk := params["dbName"]
 	port, portOk := params["port"]
 	pwd, pwdOk := params["password"]
+
+	streamingConfig, cfgOk := params["streamingCfg"]
+	if cfgOk && streamingConfig == "" {
+		return mysql, fmt.Errorf("specify a non-empty streaming config file path")
+	}
+	mysql.StreamingConfig = streamingConfig
+
 	// We don't users to mix and match params from source-profile and environment variables.
 	// We either try to get all params from the source-profile and if none are set, we read from the env variables.
 	if !(hostOk || userOk || dbOk || portOk || pwdOk) {
@@ -252,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) {
@@ -279,15 +289,26 @@ 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
 }
 
 type SourceProfileConnectionOracle struct {
-	Host string
-	Port string
-	User string
-	Db   string
-	Pwd  string
+	Host            string
+	Port            string
+	User            string
+	Db              string
+	Pwd             string
+	StreamingConfig string
 }
 
 func NewSourceProfileConnectionOracle(params map[string]string) (SourceProfileConnectionOracle, error) {
@@ -298,6 +319,12 @@ func NewSourceProfileConnectionOracle(params map[string]string) (SourceProfileCo
 	port, _ := params["port"]
 	pwd, _ := params["password"]
 
+	streamingConfig, cfgOk := params["streamingCfg"]
+	if cfgOk && streamingConfig == "" {
+		return ss, fmt.Errorf("specify a non-empty streaming config file path")
+	}
+	ss.StreamingConfig = streamingConfig
+
 	if hostOk && userOk && dbOk {
 		// All connection params provided through source-profile. Port and password handled later.
 		ss.Host, ss.User, ss.Db, ss.Port, ss.Pwd = host, user, db, port, pwd
@@ -323,6 +350,7 @@ func NewSourceProfileConnectionOracle(params map[string]string) (SourceProfileCo
 
 type SourceProfileConnection struct {
 	Ty        SourceProfileConnectionType
+	Streaming bool
 	Mysql     SourceProfileConnectionMySQL
 	Pg        SourceProfileConnectionPostgreSQL
 	Dydb      SourceProfileConnectionDynamoDB
@@ -341,6 +369,9 @@ func NewSourceProfileConnection(source string, params map[string]string) (Source
 			if err != nil {
 				return conn, err
 			}
+			if conn.Mysql.StreamingConfig != "" {
+				conn.Streaming = true
+			}
 		}
 	case "postgresql", "postgres", "pg":
 		{
@@ -357,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":
@@ -374,6 +408,9 @@ func NewSourceProfileConnection(source string, params map[string]string) (Source
 			if err != nil {
 				return conn, err
 			}
+			if conn.Oracle.StreamingConfig != "" {
+				conn.Streaming = true
+			}
 		}
 	default:
 		return conn, fmt.Errorf("please specify a valid source database using -source flag, received source = %v", source)
@@ -443,6 +480,7 @@ func (src SourceProfile) ToLegacyDriver(source string) (string, error) {
 				return "", fmt.Errorf("please specify a valid source database using -source flag, received source = %v", source)
 			}
 		}
+	// No need to handle unsupported streaming source specified as it is already covered during source profile creation.
 	case SourceProfileTypeConnection:
 		{
 			switch strings.ToLower(source) {
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/common/toddl.go b/sources/common/toddl.go
index e10bb43..05b4f79 100644
--- a/sources/common/toddl.go
+++ b/sources/common/toddl.go
@@ -74,6 +74,9 @@ func SchemaToSpannerDDL(conv *internal.Conv, toddl ToDdl) error {
 			if srcCol.Ignored.ForeignKey {
 				issues = append(issues, internal.ForeignKey)
 			}
+			if srcCol.Name != colName {
+				issues = append(issues, internal.IllegalName)
+			}
 			if srcCol.Ignored.Default {
 				issues = append(issues, internal.DefaultValue)
 			}
@@ -96,7 +99,7 @@ func SchemaToSpannerDDL(conv *internal.Conv, toddl ToDdl) error {
 			ColNames: spColNames,
 			ColDefs:  spColDef,
 			Pks:      cvtPrimaryKeys(conv, srcTable.Name, srcTable.PrimaryKeys),
-			Fks:      cvtForeignKeys(conv, srcTable.Name, srcTable.ForeignKeys),
+			Fks:      cvtForeignKeys(conv, spTableName, srcTable.Name, srcTable.ForeignKeys),
 			Indexes:  cvtIndexes(conv, spTableName, srcTable.Name, srcTable.Indexes),
 			Comment:  comment}
 	}
@@ -127,7 +130,7 @@ func cvtPrimaryKeys(conv *internal.Conv, srcTable string, srcKeys []schema.Key)
 	return spKeys
 }
 
-func cvtForeignKeys(conv *internal.Conv, srcTable string, srcKeys []schema.ForeignKey) []ddl.Foreignkey {
+func cvtForeignKeys(conv *internal.Conv, spTableName string, srcTable string, srcKeys []schema.ForeignKey) []ddl.Foreignkey {
 	var spKeys []ddl.Foreignkey
 	for _, key := range srcKeys {
 		if len(key.Columns) != len(key.ReferColumns) {
@@ -151,13 +154,17 @@ func cvtForeignKeys(conv *internal.Conv, srcTable string, srcKeys []schema.Forei
 			spReferCols = append(spReferCols, spReferCol)
 		}
 		spKeyName := internal.ToSpannerForeignKey(conv, key.Name)
+
 		spKey := ddl.Foreignkey{
 			Name:         spKeyName,
 			Columns:      spCols,
 			ReferTable:   spReferTable,
 			ReferColumns: spReferCols}
 		spKeys = append(spKeys, spKey)
+		conv.Audit.ToSpannerFkIdx[srcTable].ForeignKey[key.Name] = spKeyName
+		conv.Audit.ToSourceFkIdx[spTableName].ForeignKey[spKeyName] = key.Name
 	}
+
 	return spKeys
 }
 
@@ -165,6 +172,7 @@ func cvtIndexes(conv *internal.Conv, spTableName string, srcTable string, srcInd
 	var spIndexes []ddl.CreateIndex
 	for _, srcIndex := range srcIndexes {
 		var spKeys []ddl.IndexKey
+
 		for _, k := range srcIndex.Keys {
 			spCol, err := internal.GetSpannerCol(conv, srcTable, k.Column, true)
 			if err != nil {
@@ -186,6 +194,8 @@ func cvtIndexes(conv *internal.Conv, spTableName string, srcTable string, srcInd
 			Keys:   spKeys,
 		}
 		spIndexes = append(spIndexes, spIndex)
+		conv.Audit.ToSpannerFkIdx[srcTable].Index[srcIndex.Name] = spIndexName
+		conv.Audit.ToSourceFkIdx[spTableName].Index[spIndexName] = srcIndex.Name
 	}
 	return spIndexes
 }
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/mysql/toddl.go b/sources/mysql/toddl.go
index e76a5cf..25944b9 100644
--- a/sources/mysql/toddl.go
+++ b/sources/mysql/toddl.go
@@ -105,8 +105,6 @@ func toSpannerTypeInternal(conv *internal.Conv, id string, mods []int64) (ddl.Ty
 func overrideExperimentalType(columnType schema.Type, originalType ddl.Type) ddl.Type {
 	if len(columnType.ArrayBounds) > 0 {
 		return ddl.Type{Name: ddl.String, Len: ddl.MaxLength}
-	} else if columnType.Name == "date" {
-		return ddl.Type{Name: ddl.String, Len: ddl.MaxLength}
 	} else if columnType.Name == "json" {
 		return ddl.Type{Name: ddl.String, Len: ddl.MaxLength}
 	}
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 dfb7f66..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":
@@ -158,9 +169,7 @@ func toSpannerTypeInternal(conv *internal.Conv, spType string, srcType string, m
 
 // Override the types to map to experimental postgres types.
 func overrideExperimentalType(columnType schema.Type, originalType ddl.Type) ddl.Type {
-	if columnType.Name == "DATE" {
-		return ddl.Type{Name: ddl.String, Len: ddl.MaxLength}
-	} else if columnType.Name == "JSON" {
+	if columnType.Name == "JSON" {
 		return ddl.Type{Name: ddl.String, Len: ddl.MaxLength}
 	}
 	return originalType
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/postgres/toddl.go b/sources/postgres/toddl.go
index fe88787..cdbef60 100644
--- a/sources/postgres/toddl.go
+++ b/sources/postgres/toddl.go
@@ -112,8 +112,6 @@ func toSpannerTypeInternal(conv *internal.Conv, id string, mods []int64) (ddl.Ty
 func overrideExperimentalType(columnType schema.Type, originalType ddl.Type) ddl.Type {
 	if len(columnType.ArrayBounds) > 0 {
 		return ddl.Type{Name: ddl.String, Len: ddl.MaxLength}
-	} else if columnType.Name == "date" {
-		return ddl.Type{Name: ddl.String, Len: ddl.MaxLength}
 	} else if columnType.Name == "json" || columnType.Name == "jsonb" {
 		return ddl.Type{Name: ddl.String, Len: ddl.MaxLength}
 	}
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/sources/sqlserver/toddl.go b/sources/sqlserver/toddl.go
index c120e33..cba1ccd 100644
--- a/sources/sqlserver/toddl.go
+++ b/sources/sqlserver/toddl.go
@@ -16,7 +16,6 @@
 package sqlserver
 
 import (
-	"github.com/cloudspannerecosystem/harbourbridge/common/constants"
 	"github.com/cloudspannerecosystem/harbourbridge/internal"
 	"github.com/cloudspannerecosystem/harbourbridge/schema"
 	"github.com/cloudspannerecosystem/harbourbridge/spanner/ddl"
@@ -32,9 +31,6 @@ type ToDdlImpl struct {
 // conversion issues encountered.
 func (tdi ToDdlImpl) ToSpannerType(conv *internal.Conv, columnType schema.Type) (ddl.Type, []internal.SchemaIssue) {
 	ty, issues := toSpannerTypeInternal(columnType.Name, columnType.Mods)
-	if conv.TargetDb == constants.TargetExperimentalPostgres {
-		ty = overrideExperimentalType(columnType, ty)
-	}
 	return ty, issues
 }
 
@@ -85,11 +81,3 @@ func toSpannerTypeInternal(srcType string, mods []int64) (ddl.Type, []internal.S
 		return ddl.Type{Name: ddl.String, Len: ddl.MaxLength}, []internal.SchemaIssue{internal.NoGoodType}
 	}
 }
-
-// Override the types to map to experimental postgres types.
-func overrideExperimentalType(columnType schema.Type, originalType ddl.Type) ddl.Type {
-	if columnType.Name == "date" {
-		return ddl.Type{Name: ddl.String, Len: ddl.MaxLength}
-	}
-	return originalType
-}
diff --git a/streaming/streaming.go b/streaming/streaming.go
new file mode 100644
index 0000000..ae8d7dc
--- /dev/null
+++ b/streaming/streaming.go
@@ -0,0 +1,321 @@
+// 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 streaming
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"io/ioutil"
+	"time"
+
+	dataflow "cloud.google.com/go/dataflow/apiv1beta3"
+	datastream "cloud.google.com/go/datastream/apiv1alpha1"
+	datastreampb "google.golang.org/genproto/googleapis/cloud/datastream/v1alpha1"
+	dataflowpb "google.golang.org/genproto/googleapis/dataflow/v1beta3"
+	fieldmaskpb "google.golang.org/protobuf/types/known/fieldmaskpb"
+
+	"github.com/cloudspannerecosystem/harbourbridge/common/constants"
+	"github.com/cloudspannerecosystem/harbourbridge/common/utils"
+	"github.com/cloudspannerecosystem/harbourbridge/profiles"
+)
+
+type SrcConnCfg struct {
+	Name     string
+	Location string
+}
+
+type DstConnCfg struct {
+	Name     string
+	Location string
+	Prefix   string
+}
+
+type DatastreamCfg struct {
+	StreamId                    string
+	StreamLocation              string
+	StreamDisplayName           string
+	SourceConnectionConfig      SrcConnCfg
+	DestinationConnectionConfig DstConnCfg
+}
+
+type DataflowCfg struct {
+	JobName  string
+	Location string
+}
+
+type StreamingCfg struct {
+	DatastreamCfg DatastreamCfg
+	DataflowCfg   DataflowCfg
+}
+
+// VerifyAndUpdateCfg checks the fields and errors out if certain fields are empty.
+// It then auto-populates certain empty fields like StreamId and Dataflow JobName.
+func VerifyAndUpdateCfg(streamingCfg *StreamingCfg, dbName string) error {
+	dsCfg := streamingCfg.DatastreamCfg
+	if dsCfg.StreamLocation == "" {
+		return fmt.Errorf("please specify DatastreamCfg.StreamLocation in the streaming config")
+	}
+	srcCfg := dsCfg.SourceConnectionConfig
+	if srcCfg.Name == "" || srcCfg.Location == "" {
+		return fmt.Errorf("please specify Name and Location under DatastreamCfg.SourceConnectionConfig in the streaming config")
+	}
+	dstCfg := dsCfg.DestinationConnectionConfig
+	if dstCfg.Name == "" || dstCfg.Location == "" {
+		return fmt.Errorf("please specify Name and Location under DatastreamCfg.DestinationConnectionConfig in the streaming config")
+	}
+
+	dfCfg := streamingCfg.DataflowCfg
+	if dfCfg.Location == "" {
+		return fmt.Errorf("please specify the Location under DataflowCfg in the streaming config")
+	}
+
+	// If both ID and Display name are empty, generate a new one for both.
+	// If either is present, assign it to the other one.
+	if dsCfg.StreamId == "" && dsCfg.StreamDisplayName == "" {
+		// TODO: Update names to have more info like dbname.
+		streamId, err := utils.GenerateName("hb-stream-" + dbName)
+		if err != nil {
+			return fmt.Errorf("error generating stream name: %v", err)
+		}
+		streamingCfg.DatastreamCfg.StreamId = streamId
+		streamingCfg.DatastreamCfg.StreamDisplayName = streamId
+	} else if dsCfg.StreamId == "" {
+		streamingCfg.DatastreamCfg.StreamId = streamingCfg.DatastreamCfg.StreamDisplayName
+	} else if dsCfg.StreamDisplayName == "" {
+		streamingCfg.DatastreamCfg.StreamDisplayName = streamingCfg.DatastreamCfg.StreamId
+	}
+
+	if dfCfg.JobName == "" {
+		// Update names to have more info like dbname.
+		jobName, err := utils.GenerateName("hb-dataflow-" + dbName)
+		if err != nil {
+			return fmt.Errorf("error generating stream name: %v", err)
+		}
+		streamingCfg.DataflowCfg.JobName = jobName
+	}
+	return nil
+}
+
+// ReadStreamingConfig reads the file and unmarshalls it into the StreamingCfg struct.
+func ReadStreamingConfig(file, dbName string) (StreamingCfg, error) {
+	streamingCfg := StreamingCfg{}
+	cfgFile, err := ioutil.ReadFile(file)
+	if err != nil {
+		return streamingCfg, fmt.Errorf("can't read streaming config file due to: %v", err)
+	}
+	err = json.Unmarshal(cfgFile, &streamingCfg)
+	if err != nil {
+		return streamingCfg, fmt.Errorf("unable to unmarshall json due to: %v", err)
+	}
+	err = VerifyAndUpdateCfg(&streamingCfg, dbName)
+	if err != nil {
+		return streamingCfg, fmt.Errorf("streaming config is incomplete: %v", err)
+	}
+	return streamingCfg, nil
+}
+
+func getMysqlSourceStreamConfig(dbName string) *datastreampb.SourceConfig_MysqlSourceConfig {
+	mydb := &datastreampb.MysqlDatabase{
+		DatabaseName: dbName,
+	}
+	mysqlSrcCfg := &datastreampb.MysqlSourceConfig{
+		Allowlist: &datastreampb.MysqlRdbms{MysqlDatabases: []*datastreampb.MysqlDatabase{mydb}},
+	}
+	return &datastreampb.SourceConfig_MysqlSourceConfig{MysqlSourceConfig: mysqlSrcCfg}
+}
+
+func getOracleSourceStreamConfig(dbName string) *datastreampb.SourceConfig_OracleSourceConfig {
+	oracledb := &datastreampb.OracleSchema{
+		SchemaName: dbName,
+	}
+	oracleSrcCfg := &datastreampb.OracleSourceConfig{
+		Allowlist: &datastreampb.OracleRdbms{OracleSchemas: []*datastreampb.OracleSchema{oracledb}},
+	}
+	return &datastreampb.SourceConfig_OracleSourceConfig{OracleSourceConfig: oracleSrcCfg}
+}
+
+func getSourceStreamConfig(srcCfg *datastreampb.SourceConfig, sourceProfile profiles.SourceProfile) error {
+	switch sourceProfile.Driver {
+	case constants.MYSQL:
+		srcCfg.SourceStreamConfig = getMysqlSourceStreamConfig(sourceProfile.Conn.Mysql.Db)
+		return nil
+	case constants.ORACLE:
+		srcCfg.SourceStreamConfig = getOracleSourceStreamConfig(sourceProfile.Conn.Oracle.Db)
+		return nil
+	default:
+		return fmt.Errorf("only MySQL and Oracle are supported as source streams")
+	}
+}
+
+// LaunchStream populates the parameters from the streaming config and triggers a stream on Cloud Datastream.
+func LaunchStream(ctx context.Context, sourceProfile profiles.SourceProfile, projectID string, datastreamCfg DatastreamCfg) error {
+	fmt.Println("Launching stream ", fmt.Sprintf("projects/%s/locations/%s", projectID, datastreamCfg.StreamLocation))
+	dsClient, err := datastream.NewClient(ctx)
+	if err != nil {
+		return fmt.Errorf("datastream client can not be created: %v", err)
+	}
+	defer dsClient.Close()
+	fmt.Println("Created client...")
+
+	gcsDstCfg := &datastreampb.GcsDestinationConfig{
+		Path:       datastreamCfg.DestinationConnectionConfig.Prefix,
+		FileFormat: &datastreampb.GcsDestinationConfig_AvroFileFormat{},
+	}
+	srcCfg := &datastreampb.SourceConfig{
+		SourceConnectionProfileName: fmt.Sprintf("projects/%s/locations/%s/connectionProfiles/%s", projectID, datastreamCfg.SourceConnectionConfig.Location, datastreamCfg.SourceConnectionConfig.Name),
+	}
+	getSourceStreamConfig(srcCfg, sourceProfile)
+
+	dstCfg := &datastreampb.DestinationConfig{
+		DestinationConnectionProfileName: fmt.Sprintf("projects/%s/locations/%s/connectionProfiles/%s", projectID, datastreamCfg.DestinationConnectionConfig.Location, datastreamCfg.DestinationConnectionConfig.Name),
+		DestinationStreamConfig:          &datastreampb.DestinationConfig_GcsDestinationConfig{GcsDestinationConfig: gcsDstCfg},
+	}
+	streamInfo := &datastreampb.Stream{
+		DisplayName:       datastreamCfg.StreamDisplayName,
+		SourceConfig:      srcCfg,
+		DestinationConfig: dstCfg,
+		State:             datastreampb.Stream_RUNNING,
+		BackfillStrategy:  &datastreampb.Stream_BackfillNone{BackfillNone: &datastreampb.Stream_BackfillNoneStrategy{}},
+	}
+	createStreamRequest := &datastreampb.CreateStreamRequest{
+		Parent:   fmt.Sprintf("projects/%s/locations/%s", projectID, datastreamCfg.StreamLocation),
+		StreamId: datastreamCfg.StreamId,
+		Stream:   streamInfo,
+	}
+
+	fmt.Println("Created stream request..")
+
+	dsOp, err := dsClient.CreateStream(ctx, createStreamRequest)
+	if err != nil {
+		fmt.Printf("createStreamRequest: %+v\n", createStreamRequest)
+		return fmt.Errorf("cannot create stream: %v ", err)
+	}
+
+	_, err = dsOp.Wait(ctx)
+	if err != nil {
+		fmt.Printf("createStreamRequest: %+v\n", createStreamRequest)
+		return fmt.Errorf("datastream create operation failed: %v", err)
+	}
+	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"}},
+		Stream:     streamInfo,
+	}
+	upOp, err := dsClient.UpdateStream(ctx, updateStreamRequest)
+	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("Done")
+	return nil
+}
+
+// LaunchDataflowJob populates the parameters from the streaming config and triggers a Dataflow job.
+func LaunchDataflowJob(ctx context.Context, targetProfile profiles.TargetProfile, datastreamCfg DatastreamCfg, dataflowCfg DataflowCfg) error {
+	project, instance, dbName, _ := targetProfile.GetResourceIds(ctx, time.Now(), "", nil)
+	fmt.Println("Launching dataflow job ", dataflowCfg.JobName, " in ", project, "-", dataflowCfg.Location)
+
+	c, err := dataflow.NewFlexTemplatesClient(ctx)
+	if err != nil {
+		return fmt.Errorf("could not create flex template client: %v", err)
+	}
+	defer c.Close()
+	fmt.Println("Created flex template client...")
+
+	//Creating datastream client to fetch the gcs bucket using target profile.
+	dsClient, err := datastream.NewClient(ctx)
+	if err != nil {
+		return fmt.Errorf("datastream client can not be created: %v", err)
+	}
+	defer dsClient.Close()
+
+	// Fetch the GCS path from the destination connection profile.
+	dstProf := fmt.Sprintf("projects/%s/locations/%s/connectionProfiles/%s", project, datastreamCfg.DestinationConnectionConfig.Location, datastreamCfg.DestinationConnectionConfig.Name)
+	res, err := dsClient.GetConnectionProfile(ctx, &datastreampb.GetConnectionProfileRequest{Name: dstProf})
+	if err != nil {
+		return fmt.Errorf("could not get connection profiles: %v", err)
+	}
+	gcsProfile := res.Profile.(*datastreampb.ConnectionProfile_GcsProfile).GcsProfile
+	inputFilePattern := "gs://" + gcsProfile.BucketName + gcsProfile.RootPath + datastreamCfg.DestinationConnectionConfig.Prefix
+	fmt.Println("Reading files from datastream destination ", inputFilePattern)
+
+	launchParameter := &dataflowpb.LaunchFlexTemplateParameter{
+		JobName:  dataflowCfg.JobName,
+		Template: &dataflowpb.LaunchFlexTemplateParameter_ContainerSpecGcsPath{ContainerSpecGcsPath: "gs://dataflow-templates/latest/flex/Cloud_Datastream_to_Spanner"},
+		Parameters: map[string]string{
+			"inputFilePattern": inputFilePattern,
+			"streamName":       fmt.Sprintf("projects/%s/locations/%s/streams/%s", project, datastreamCfg.StreamLocation, datastreamCfg.StreamId),
+			"instanceId":       instance,
+			"databaseId":       dbName,
+		},
+	}
+
+	req := &dataflowpb.LaunchFlexTemplateRequest{
+		ProjectId:       project,
+		LaunchParameter: launchParameter,
+		Location:        dataflowCfg.Location,
+	}
+	fmt.Println("Created flex template request body...")
+
+	respDf, err := c.LaunchFlexTemplate(ctx, req)
+	if err != nil {
+		fmt.Printf("flexTemplateRequest: %+v\n", req)
+		return fmt.Errorf("unable to launch template: %v", err)
+	}
+	fullStreamName := fmt.Sprintf("projects/%s/locations/%s/streams/%s", project, datastreamCfg.StreamLocation, datastreamCfg.StreamId)
+	dfJobDetails := fmt.Sprintf("project: %s, location: %s, name: %s, id: %s", project, respDf.Job.Location, respDf.Job.Name, respDf.Job.Id)
+	fmt.Println("\n------------------------------------------\n" +
+		"The Datastream job: " + fullStreamName + "and the Dataflow job: " + dfJobDetails +
+		" 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/dynamodb/toddl_test.go b/sources/dynamodb/toddl_test.go
index d86a3f3..ddb7abe 100644
--- a/sources/dynamodb/toddl_test.go
+++ b/sources/dynamodb/toddl_test.go
@@ -17,6 +17,7 @@ package dynamodb
 import (
 	"testing"
 
+	"github.com/cloudspannerecosystem/harbourbridge/common/constants"
 	"github.com/cloudspannerecosystem/harbourbridge/internal"
 	"github.com/cloudspannerecosystem/harbourbridge/schema"
 	"github.com/cloudspannerecosystem/harbourbridge/sources/common"
@@ -79,6 +80,62 @@ func TestToSpannerType(t *testing.T) {
 	assert.Equal(t, expected, actual)
 }
 
+func TestToExperimentalSpannerType(t *testing.T) {
+	conv := internal.MakeConv()
+	conv.SetSchemaMode()
+	conv.TargetDb = constants.TargetExperimentalPostgres
+	name := "test"
+	srcSchema := schema.Table{
+		Name:     name,
+		ColNames: []string{"a", "b", "c", "d", "e", "f", "g", "h", "i", "j", "k"},
+		ColDefs: map[string]schema.Column{
+			"a": {Name: "a", Type: schema.Type{Name: typeString}},
+			"b": {Name: "b", Type: schema.Type{Name: typeNumber}},
+			"c": {Name: "c", Type: schema.Type{Name: typeNumberString}},
+			"d": {Name: "d", Type: schema.Type{Name: typeBool}},
+			"e": {Name: "e", Type: schema.Type{Name: typeBinary}},
+			"f": {Name: "f", Type: schema.Type{Name: typeList}},
+			"g": {Name: "g", Type: schema.Type{Name: typeMap}},
+			"h": {Name: "h", Type: schema.Type{Name: typeStringSet}},
+			"i": {Name: "i", Type: schema.Type{Name: typeBinarySet}},
+			"j": {Name: "j", Type: schema.Type{Name: typeNumberSet}},
+			"k": {Name: "k", Type: schema.Type{Name: typeNumberStringSet}},
+		},
+		PrimaryKeys: []schema.Key{{Column: "a"}, {Column: "b"}},
+		Indexes: []schema.Index{
+			{Name: "index1", Keys: []schema.Key{{Column: "b"}, {Column: "c"}}},
+			{Name: "test", Keys: []schema.Key{{Column: "d"}}},
+		},
+	}
+	conv.SrcSchema[name] = srcSchema
+	assert.Nil(t, common.SchemaToSpannerDDL(conv, ToDdlImpl{}))
+	actual := conv.SpSchema[name]
+	dropComments(&actual) // Don't test comment.
+	expected := ddl.CreateTable{
+		Name:     name,
+		ColNames: []string{"a", "b", "c", "d", "e", "f", "g", "h", "i", "j", "k"},
+		ColDefs: map[string]ddl.ColumnDef{
+			"a": {Name: "a", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
+			"b": {Name: "b", T: ddl.Type{Name: ddl.Numeric}},
+			"c": {Name: "c", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
+			"d": {Name: "d", T: ddl.Type{Name: ddl.Bool}},
+			"e": {Name: "e", T: ddl.Type{Name: ddl.Bytes, Len: ddl.MaxLength}},
+			"f": {Name: "f", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
+			"g": {Name: "g", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
+			"h": {Name: "h", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
+			"i": {Name: "i", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
+			"j": {Name: "j", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
+			"k": {Name: "k", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
+		},
+		Pks: []ddl.IndexKey{{Col: "a"}, {Col: "b"}},
+		Indexes: []ddl.CreateIndex{
+			{Name: "index1", Table: name, Keys: []ddl.IndexKey{{Col: "b"}, {Col: "c"}}},
+			{Name: "test_2", Table: name, Keys: []ddl.IndexKey{{Col: "d"}}},
+		},
+	}
+	assert.Equal(t, expected, actual)
+}
+
 func dropComments(t *ddl.CreateTable) {
 	t.Comment = ""
 	for _, c := range t.ColNames {
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/mysql/report_test.go b/sources/mysql/report_test.go
index b34d036..764c29a 100644
--- a/sources/mysql/report_test.go
+++ b/sources/mysql/report_test.go
@@ -74,7 +74,7 @@ func TestReport(t *testing.T) {
 		`----------------------------
 Summary of Conversion
 ----------------------------
-Schema conversion: GOOD (most columns mapped cleanly, but some missing primary keys).
+Schema conversion: GOOD (99.99412% of 17006 columns mapped cleanly) + some missing primary keys.
 Data conversion: POOR (66% of 6000 rows written to Spanner).
 
 The remainder of this report provides stats on the mysqldump statements
@@ -98,6 +98,14 @@ Analysis of statements in mysqldump output, broken down by statement type.
 See https://github.com/pingcap/parser for definitions of statement types
 (pingcap/tidb/parser is the library we use for parsing mysqldump output).
 
+-----------------------------------------------------------------------------------------------------
+Name Changes in Migration
+-----------------------------------------------------------------------------------------------------
+             Source Table          Change                  Old Name                  New Name
+-----------------------------------------------------------------------------------------------------
+-----------------------------------------------------------------------------------------------------
+
+
 ----------------------------
 Table bad_schema
 ----------------------------
@@ -115,7 +123,7 @@ Note
 ----------------------------
 Table default_value
 ----------------------------
-Schema conversion: POOR (many columns did not map cleanly).
+Schema conversion: POOR (50% of 2 columns mapped cleanly).
 Data conversion: NONE (no data rows found).
 
 Warning
@@ -125,13 +133,13 @@ Warning
 ----------------------------
 Table excellent_schema
 ----------------------------
-Schema conversion: EXCELLENT (all columns mapped cleanly).
+Schema conversion: EXCELLENT (all 2 columns mapped cleanly).
 Data conversion: NONE (no data rows found).
 
 ----------------------------
 Table foreign_key
 ----------------------------
-Schema conversion: EXCELLENT (all columns mapped cleanly).
+Schema conversion: EXCELLENT (all 2 columns mapped cleanly).
 Data conversion: NONE (no data rows found).
 
 ----------------------------
diff --git a/sources/mysql/toddl_test.go b/sources/mysql/toddl_test.go
index c849d57..f42a3e9 100644
--- a/sources/mysql/toddl_test.go
+++ b/sources/mysql/toddl_test.go
@@ -17,6 +17,7 @@ package mysql
 import (
 	"testing"
 
+	"github.com/cloudspannerecosystem/harbourbridge/common/constants"
 	"github.com/cloudspannerecosystem/harbourbridge/internal"
 	"github.com/cloudspannerecosystem/harbourbridge/schema"
 	"github.com/cloudspannerecosystem/harbourbridge/sources/common"
@@ -31,7 +32,7 @@ func TestToSpannerType(t *testing.T) {
 	name := "test"
 	srcSchema := schema.Table{
 		Name:     name,
-		ColNames: []string{"a", "b", "c", "d", "e", "f", "g"},
+		ColNames: []string{"a", "b", "c", "d", "e", "f", "g", "h", "i", "j"},
 		ColDefs: map[string]schema.Column{
 			"a": schema.Column{Name: "a", Type: schema.Type{Name: "int"}},
 			"b": schema.Column{Name: "b", Type: schema.Type{Name: "float"}},
@@ -40,6 +41,9 @@ func TestToSpannerType(t *testing.T) {
 			"e": schema.Column{Name: "e", Type: schema.Type{Name: "numeric"}},
 			"f": schema.Column{Name: "f", Type: schema.Type{Name: "timestamp"}},
 			"g": schema.Column{Name: "g", Type: schema.Type{Name: "json"}},
+			"h": schema.Column{Name: "h", Type: schema.Type{Name: "date"}},
+			"i": schema.Column{Name: "i", Type: schema.Type{Name: "timestamp"}},
+			"j": schema.Column{Name: "j", Type: schema.Type{Name: "bit"}},
 		},
 		PrimaryKeys: []schema.Key{schema.Key{Column: "a"}},
 		ForeignKeys: []schema.ForeignKey{schema.ForeignKey{Name: "fk_test", Columns: []string{"d"}, ReferTable: "ref_table", ReferColumns: []string{"dref"}},
@@ -72,7 +76,7 @@ func TestToSpannerType(t *testing.T) {
 	dropComments(&actual) // Don't test comment.
 	expected := ddl.CreateTable{
 		Name:     name,
-		ColNames: []string{"a", "b", "c", "d", "e", "f", "g"},
+		ColNames: []string{"a", "b", "c", "d", "e", "f", "g", "h", "i", "j"},
 		ColDefs: map[string]ddl.ColumnDef{
 			"a": ddl.ColumnDef{Name: "a", T: ddl.Type{Name: ddl.Int64}},
 			"b": ddl.ColumnDef{Name: "b", T: ddl.Type{Name: ddl.Float64}},
@@ -81,6 +85,87 @@ func TestToSpannerType(t *testing.T) {
 			"e": ddl.ColumnDef{Name: "e", T: ddl.Type{Name: ddl.Numeric}},
 			"f": ddl.ColumnDef{Name: "f", T: ddl.Type{Name: ddl.Timestamp}},
 			"g": ddl.ColumnDef{Name: "g", T: ddl.Type{Name: ddl.JSON}},
+			"h": ddl.ColumnDef{Name: "h", T: ddl.Type{Name: ddl.Date}},
+			"i": ddl.ColumnDef{Name: "i", T: ddl.Type{Name: ddl.Timestamp}},
+			"j": ddl.ColumnDef{Name: "j", T: ddl.Type{Name: ddl.Bytes, Len: ddl.MaxLength}},
+		},
+		Pks: []ddl.IndexKey{ddl.IndexKey{Col: "a"}},
+		Fks: []ddl.Foreignkey{ddl.Foreignkey{Name: "fk_test", Columns: []string{"d"}, ReferTable: "ref_table", ReferColumns: []string{"dref"}},
+			ddl.Foreignkey{Name: "fk_test2", Columns: []string{"a"}, ReferTable: "ref_table2", ReferColumns: []string{"aref"}}},
+		Indexes: []ddl.CreateIndex{ddl.CreateIndex{Name: "index1", Table: name, Unique: true, Keys: []ddl.IndexKey{ddl.IndexKey{Col: "a", Desc: false}, ddl.IndexKey{Col: "d", Desc: true}}}},
+	}
+	assert.Equal(t, expected, actual)
+	expectedIssues := map[string][]internal.SchemaIssue{
+		"a": []internal.SchemaIssue{internal.Widened},
+		"b": []internal.SchemaIssue{internal.Widened},
+	}
+	assert.Equal(t, expectedIssues, conv.Issues[name])
+}
+
+// This is just a very basic smoke-test for toExperimentalSpannerType.
+func TestToExperimentalSpannerType(t *testing.T) {
+	conv := internal.MakeConv()
+	conv.SetSchemaMode()
+	conv.TargetDb = constants.TargetExperimentalPostgres
+	name := "test"
+	srcSchema := schema.Table{
+		Name:     name,
+		ColNames: []string{"a", "b", "c", "d", "e", "f", "g", "h", "i", "j"},
+		ColDefs: map[string]schema.Column{
+			"a": schema.Column{Name: "a", Type: schema.Type{Name: "int"}},
+			"b": schema.Column{Name: "b", Type: schema.Type{Name: "float"}},
+			"c": schema.Column{Name: "c", Type: schema.Type{Name: "tinyint", Mods: []int64{1}}},
+			"d": schema.Column{Name: "d", Type: schema.Type{Name: "varchar", Mods: []int64{6}}},
+			"e": schema.Column{Name: "e", Type: schema.Type{Name: "numeric"}},
+			"f": schema.Column{Name: "f", Type: schema.Type{Name: "timestamp"}},
+			"g": schema.Column{Name: "g", Type: schema.Type{Name: "json"}},
+			"h": schema.Column{Name: "h", Type: schema.Type{Name: "date"}},
+			"i": schema.Column{Name: "i", Type: schema.Type{Name: "timestamp"}},
+			"j": schema.Column{Name: "j", Type: schema.Type{Name: "bit"}},
+		},
+		PrimaryKeys: []schema.Key{schema.Key{Column: "a"}},
+		ForeignKeys: []schema.ForeignKey{schema.ForeignKey{Name: "fk_test", Columns: []string{"d"}, ReferTable: "ref_table", ReferColumns: []string{"dref"}},
+			schema.ForeignKey{Name: "fk_test2", Columns: []string{"a"}, ReferTable: "ref_table2", ReferColumns: []string{"aRef"}}},
+		Indexes: []schema.Index{schema.Index{Name: "index1", Unique: true, Keys: []schema.Key{schema.Key{Column: "a", Desc: false}, schema.Key{Column: "d", Desc: true}}}},
+	}
+	conv.SrcSchema[name] = srcSchema
+	conv.SpSchema["ref_table"] = ddl.CreateTable{
+		Name:     "ref_table",
+		ColNames: []string{"dref", "b", "c"},
+		ColDefs: map[string]ddl.ColumnDef{
+			"dref": ddl.ColumnDef{Name: "dref", T: ddl.Type{Name: ddl.String, Len: int64(6)}},
+			"b":    ddl.ColumnDef{Name: "b", T: ddl.Type{Name: ddl.Float64}},
+			"c":    ddl.ColumnDef{Name: "c", T: ddl.Type{Name: ddl.Bool}},
+		},
+		Pks: []ddl.IndexKey{ddl.IndexKey{Col: "dref"}},
+	}
+	conv.SpSchema["ref_table2"] = ddl.CreateTable{
+		Name:     "ref_table2",
+		ColNames: []string{"aref", "b", "c"},
+		ColDefs: map[string]ddl.ColumnDef{
+			"aref": ddl.ColumnDef{Name: "aref", T: ddl.Type{Name: ddl.Int64}},
+			"b":    ddl.ColumnDef{Name: "b", T: ddl.Type{Name: ddl.Float64}},
+			"c":    ddl.ColumnDef{Name: "c", T: ddl.Type{Name: ddl.Bool}},
+		},
+		Pks: []ddl.IndexKey{ddl.IndexKey{Col: "aref"}},
+	}
+	assert.Nil(t, common.SchemaToSpannerDDL(conv, ToDdlImpl{}))
+	actual := conv.SpSchema[name]
+	dropComments(&actual) // Don't test comment.
+	expected := ddl.CreateTable{
+		Name:     name,
+		ColNames: []string{"a", "b", "c", "d", "e", "f", "g", "h", "i", "j"},
+		ColDefs: map[string]ddl.ColumnDef{
+			"a": ddl.ColumnDef{Name: "a", T: ddl.Type{Name: ddl.Int64}},
+			"b": ddl.ColumnDef{Name: "b", T: ddl.Type{Name: ddl.Float64}},
+			"c": ddl.ColumnDef{Name: "c", T: ddl.Type{Name: ddl.Bool}},
+			"d": ddl.ColumnDef{Name: "d", T: ddl.Type{Name: ddl.String, Len: int64(6)}},
+			"e": ddl.ColumnDef{Name: "e", T: ddl.Type{Name: ddl.Numeric}},
+			"f": ddl.ColumnDef{Name: "f", T: ddl.Type{Name: ddl.Timestamp}},
+			"g": ddl.ColumnDef{Name: "g", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
+			"h": ddl.ColumnDef{Name: "h", T: ddl.Type{Name: ddl.Date}},
+			"i": ddl.ColumnDef{Name: "i", T: ddl.Type{Name: ddl.Timestamp}},
+			"j": ddl.ColumnDef{Name: "j", T: ddl.Type{Name: ddl.Bytes, Len: ddl.MaxLength}},
 		},
 		Pks: []ddl.IndexKey{ddl.IndexKey{Col: "a"}},
 		Fks: []ddl.Foreignkey{ddl.Foreignkey{Name: "fk_test", Columns: []string{"d"}, ReferTable: "ref_table", ReferColumns: []string{"dref"}},
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": {
diff --git a/sources/oracle/toddl_test.go b/sources/oracle/toddl_test.go
index 769d5ae..137693a 100644
--- a/sources/oracle/toddl_test.go
+++ b/sources/oracle/toddl_test.go
@@ -17,6 +17,7 @@ package oracle
 import (
 	"testing"
 
+	"github.com/cloudspannerecosystem/harbourbridge/common/constants"
 	"github.com/cloudspannerecosystem/harbourbridge/internal"
 	"github.com/cloudspannerecosystem/harbourbridge/schema"
 	"github.com/cloudspannerecosystem/harbourbridge/sources/common"
@@ -99,6 +100,89 @@ func TestToSpannerType(t *testing.T) {
 	assert.Equal(t, int64(3), conv.Unexpecteds())
 }
 
+// This is just a very basic smoke-test for toExperimentalSpannerType.
+func TestToExperimentalSpannerType(t *testing.T) {
+	conv := internal.MakeConv()
+	conv.SetSchemaMode()
+	conv.TargetDb = constants.TargetExperimentalPostgres
+	name := "test"
+	srcSchema := schema.Table{
+		Name:     name,
+		ColNames: []string{"a", "b", "c", "d", "e", "f", "g", "h", "i", "j", "k"},
+		ColDefs: map[string]schema.Column{
+			"a": {Name: "a", Type: schema.Type{Name: "NUMBER"}},
+			"b": {Name: "b", Type: schema.Type{Name: "FLOAT"}},
+			"c": {Name: "c", Type: schema.Type{Name: "BFILE"}},
+			"d": {Name: "d", Type: schema.Type{Name: "VARCHAR2", Mods: []int64{20}}},
+			"e": {Name: "e", Type: schema.Type{Name: "DATE"}},
+			"f": {Name: "f", Type: schema.Type{Name: "TIMESTAMP"}},
+			"g": {Name: "g", Type: schema.Type{Name: "LONG"}},
+			"h": {Name: "h", Type: schema.Type{Name: "NUMBER", Mods: []int64{13}}},
+			"i": {Name: "i", Type: schema.Type{Name: "JSON"}},
+			"j": {Name: "j", Type: schema.Type{Name: "NCLOB"}},
+			"k": {Name: "k", Type: schema.Type{Name: "XMLTYPE"}},
+		},
+		PrimaryKeys: []schema.Key{{Column: "a"}},
+		ForeignKeys: []schema.ForeignKey{{Name: "fk_test", Columns: []string{"d"}, ReferTable: "ref_table", ReferColumns: []string{"dref"}},
+			{Name: "fk_fake", Columns: []string{"x"}, ReferTable: "ref_table", ReferColumns: []string{"x"}},
+			{Name: "fk_test2", Columns: []string{"a"}, ReferTable: "ref_table2", ReferColumns: []string{"aRef"}}},
+		Indexes: []schema.Index{{Name: "index1", Unique: true, Keys: []schema.Key{{Column: "a", Desc: false}, {Column: "d", Desc: true}}},
+			{Name: "index_with_0_key", Unique: true, Keys: []schema.Key{{Column: "m", Desc: false}, {Column: "y", Desc: true}}},
+		},
+	}
+	conv.SrcSchema[name] = srcSchema
+	conv.SpSchema["ref_table"] = ddl.CreateTable{
+		Name:     "ref_table",
+		ColNames: []string{"dref", "b", "c"},
+		ColDefs: map[string]ddl.ColumnDef{
+			"dref": {Name: "dref", T: ddl.Type{Name: ddl.String, Len: int64(6)}},
+			"b":    {Name: "b", T: ddl.Type{Name: ddl.Float64}},
+			"c":    {Name: "c", T: ddl.Type{Name: ddl.Bool}},
+		},
+		Pks: []ddl.IndexKey{{Col: "dref"}},
+	}
+	conv.SpSchema["ref_table2"] = ddl.CreateTable{
+		Name:     "ref_table2",
+		ColNames: []string{"aref", "b", "c"},
+		ColDefs: map[string]ddl.ColumnDef{
+			"aref": {Name: "aref", T: ddl.Type{Name: ddl.Numeric}},
+			"b":    {Name: "b", T: ddl.Type{Name: ddl.Float64}},
+			"c":    {Name: "c", T: ddl.Type{Name: ddl.Bool}},
+		},
+		Pks: []ddl.IndexKey{{Col: "aref"}},
+	}
+	assert.Nil(t, common.SchemaToSpannerDDL(conv, ToDdlImpl{}))
+	actual := conv.SpSchema[name]
+	dropComments(&actual) // Don't test comment.
+	expected := ddl.CreateTable{
+		Name:     name,
+		ColNames: []string{"a", "b", "c", "d", "e", "f", "g", "h", "i", "j", "k"},
+		ColDefs: map[string]ddl.ColumnDef{
+			"a": {Name: "a", T: ddl.Type{Name: ddl.Numeric}},
+			"b": {Name: "b", T: ddl.Type{Name: ddl.Float64}},
+			"c": {Name: "c", T: ddl.Type{Name: ddl.Bytes, Len: ddl.MaxLength}},
+			"d": {Name: "d", T: ddl.Type{Name: ddl.String, Len: int64(20)}},
+			"e": {Name: "e", T: ddl.Type{Name: ddl.Date}},
+			"f": {Name: "f", T: ddl.Type{Name: ddl.Timestamp}},
+			"g": {Name: "g", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
+			"h": {Name: "h", T: ddl.Type{Name: ddl.Int64}},
+			"i": {Name: "i", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
+			"j": {Name: "j", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
+			"k": {Name: "k", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
+		},
+		Pks: []ddl.IndexKey{{Col: "a"}},
+		Fks: []ddl.Foreignkey{{Name: "fk_test", Columns: []string{"d"}, ReferTable: "ref_table", ReferColumns: []string{"dref"}},
+			{Name: "fk_test2", Columns: []string{"a"}, ReferTable: "ref_table2", ReferColumns: []string{"aref"}}},
+		Indexes: []ddl.CreateIndex{{Name: "index1", Table: name, Unique: true, Keys: []ddl.IndexKey{{Col: "a", Desc: false}, {Col: "d", Desc: true}}},
+			{Name: "index_with_0_key", Table: name, Unique: true, Keys: nil}},
+	}
+	assert.Equal(t, expected, actual)
+	expectedIssues := map[string][]internal.SchemaIssue{}
+	assert.Equal(t, expectedIssues, conv.Issues[name])
+	// 1 FK issue, 2 index col not found
+	assert.Equal(t, int64(3), conv.Unexpecteds())
+}
+
 func dropComments(t *ddl.CreateTable) {
 	t.Comment = ""
 	for _, c := range t.ColNames {
diff --git a/sources/postgres/pgdump_test.go b/sources/postgres/pgdump_test.go
index 57c65ed..38ec38b 100644
--- a/sources/postgres/pgdump_test.go
+++ b/sources/postgres/pgdump_test.go
@@ -720,7 +720,7 @@ func TestProcessPgDumpPGTarget(t *testing.T) {
 		{"boolean", ddl.Type{Name: ddl.Bool}},
 		{"bytea", ddl.Type{Name: ddl.Bytes, Len: ddl.MaxLength}},
 		{"char(42)", ddl.Type{Name: ddl.String, Len: int64(42)}},
-		{"date", ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
+		{"date", ddl.Type{Name: ddl.Date}},
 		{"decimal", ddl.Type{Name: ddl.Numeric}}, // pg parser maps this to numeric.
 		{"double precision", ddl.Type{Name: ddl.Float64}},
 		{"float8", ddl.Type{Name: ddl.Float64}},
@@ -1281,7 +1281,7 @@ COPY test (id, a, b, c) FROM stdin;
 \.
 `,
 			expectedData: []spannerData{
-				spannerData{table: "test", cols: []string{"id", "a", "b", "c"}, vals: []interface{}{int64(1), "2019-10-29", float64(4.444), float64(5.44444)}}},
+				spannerData{table: "test", cols: []string{"id", "a", "b", "c"}, vals: []interface{}{int64(1), getDate("2019-10-29"), float64(4.444), float64(5.44444)}}},
 		},
 		{
 			name: "Data conversion: int8, int4, int2, numeric",
@@ -1356,25 +1356,24 @@ COPY test (id, a, b, c, d, e) FROM stdin;
 				spannerData{
 					table: "test", cols: []string{"int8", "float8", "bool", "timestamp", "date", "bytea", "arr", "synth_id"},
 					vals: []interface{}{int64(7), float64(42.1), true, getTime(t, "2019-10-29T05:30:00Z"),
-						"2019-10-29", []byte{0x0, 0x1, 0xbe, 0xef},
+						getDate("2019-10-29"), []byte{0x0, 0x1, 0xbe, 0xef},
 						"{42,6}",
 						bitReverse(0)}},
 				spannerData{table: "test", cols: []string{"int8", "synth_id"}, vals: []interface{}{int64(7), bitReverse(1)}},
 				spannerData{table: "test", cols: []string{"float8", "synth_id"}, vals: []interface{}{float64(42.1), bitReverse(2)}},
 				spannerData{table: "test", cols: []string{"bool", "synth_id"}, vals: []interface{}{true, bitReverse(3)}},
 				spannerData{table: "test", cols: []string{"timestamp", "synth_id"}, vals: []interface{}{getTime(t, "2019-10-29T05:30:00Z"), bitReverse(4)}},
-				spannerData{table: "test", cols: []string{"date", "synth_id"}, vals: []interface{}{"2019-10-29", bitReverse(5)}},
-				spannerData{table: "test", cols: []string{"date", "synth_id"}, vals: []interface{}{"2019-10-42", bitReverse(6)}},
-				spannerData{table: "test", cols: []string{"bytea", "synth_id"}, vals: []interface{}{[]byte{0x0, 0x1, 0xbe, 0xef}, bitReverse(7)}},
-				spannerData{table: "test", cols: []string{"arr", "synth_id"}, vals: []interface{}{"{42,6}", bitReverse(8)}},
-				spannerData{table: "test", cols: []string{"arr", "synth_id"}, vals: []interface{}{"{42, 6}", bitReverse(9)}},
+				spannerData{table: "test", cols: []string{"date", "synth_id"}, vals: []interface{}{getDate("2019-10-29"), bitReverse(5)}},
+				spannerData{table: "test", cols: []string{"bytea", "synth_id"}, vals: []interface{}{[]byte{0x0, 0x1, 0xbe, 0xef}, bitReverse(6)}},
+				spannerData{table: "test", cols: []string{"arr", "synth_id"}, vals: []interface{}{"{42,6}", bitReverse(7)}},
+				spannerData{table: "test", cols: []string{"arr", "synth_id"}, vals: []interface{}{"{42, 6}", bitReverse(8)}},
 			},
 		},
 	}
 	for _, tc := range dataErrorTests {
 		conv, rows := runProcessPgDumpPGTarget(tc.input)
 		assert.Equal(t, tc.expectedData, rows, tc.name+": Data rows did not match")
-		assert.Equal(t, conv.BadRows(), int64(5), tc.name+": Error count did not match")
+		assert.Equal(t, conv.BadRows(), int64(6), tc.name+": Error count did not match")
 	}
 }
 
diff --git a/sources/postgres/report_test.go b/sources/postgres/report_test.go
index dcac524..588c1c0 100644
--- a/sources/postgres/report_test.go
+++ b/sources/postgres/report_test.go
@@ -70,7 +70,7 @@ func TestReport(t *testing.T) {
 		`----------------------------
 Summary of Conversion
 ----------------------------
-Schema conversion: OK (some columns did not map cleanly + some missing primary keys).
+Schema conversion: OK (89% of 19006 columns mapped cleanly) + some missing primary keys.
 Data conversion: POOR (66% of 6000 rows written to Spanner).
 
 The remainder of this report provides stats on the pg_dump statements processed,
@@ -93,10 +93,18 @@ Analysis of statements in pg_dump output, broken down by statement type.
 See github.com/pganalyze/pg_query_go for definitions of statement types
 (pganalyze/pg_query_go is the library we use for parsing pg_dump output).
 
+-----------------------------------------------------------------------------------------------------
+Name Changes in Migration
+-----------------------------------------------------------------------------------------------------
+             Source Table          Change                  Old Name                  New Name
+-----------------------------------------------------------------------------------------------------
+-----------------------------------------------------------------------------------------------------
+
+
 ----------------------------
 Table bad_schema
 ----------------------------
-Schema conversion: POOR (many columns did not map cleanly + missing primary key).
+Schema conversion: POOR (50% of 4 columns mapped cleanly) + missing primary key.
 Data conversion: OK (94% of 1000 rows written to Spanner).
 
 Warnings
@@ -114,7 +122,7 @@ Note
 ----------------------------
 Table default_value
 ----------------------------
-Schema conversion: POOR (many columns did not map cleanly).
+Schema conversion: POOR (50% of 2 columns mapped cleanly).
 Data conversion: NONE (no data rows found).
 
 Warning
@@ -124,13 +132,13 @@ Warning
 ----------------------------
 Table excellent_schema
 ----------------------------
-Schema conversion: EXCELLENT (all columns mapped cleanly).
+Schema conversion: EXCELLENT (all 2 columns mapped cleanly).
 Data conversion: NONE (no data rows found).
 
 ----------------------------
 Table foreign_key
 ----------------------------
-Schema conversion: EXCELLENT (all columns mapped cleanly).
+Schema conversion: EXCELLENT (all 2 columns mapped cleanly).
 Data conversion: NONE (no data rows found).
 
 ----------------------------
diff --git a/sources/postgres/toddl_test.go b/sources/postgres/toddl_test.go
index 0cc9bce..c84e756 100644
--- a/sources/postgres/toddl_test.go
+++ b/sources/postgres/toddl_test.go
@@ -156,7 +156,7 @@ func TestToExperimentalSpannerType(t *testing.T) {
 			"c": ddl.ColumnDef{Name: "c", T: ddl.Type{Name: ddl.Bool}},
 			"d": ddl.ColumnDef{Name: "d", T: ddl.Type{Name: ddl.String, Len: int64(6)}},
 			"e": ddl.ColumnDef{Name: "e", T: ddl.Type{Name: ddl.Numeric}},
-			"f": ddl.ColumnDef{Name: "f", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
+			"f": ddl.ColumnDef{Name: "f", T: ddl.Type{Name: ddl.Date}},
 			"g": ddl.ColumnDef{Name: "g", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
 		},
 		Pks: []ddl.IndexKey{ddl.IndexKey{Col: "a"}},
diff --git a/sources/sqlserver/toddl_test.go b/sources/sqlserver/toddl_test.go
index f8eb790..f092643 100644
--- a/sources/sqlserver/toddl_test.go
+++ b/sources/sqlserver/toddl_test.go
@@ -17,6 +17,7 @@ package sqlserver
 import (
 	"testing"
 
+	"github.com/cloudspannerecosystem/harbourbridge/common/constants"
 	"github.com/cloudspannerecosystem/harbourbridge/internal"
 	"github.com/cloudspannerecosystem/harbourbridge/schema"
 	"github.com/cloudspannerecosystem/harbourbridge/sources/common"
@@ -31,7 +32,7 @@ func TestToSpannerType(t *testing.T) {
 	name := "test"
 	srcSchema := schema.Table{
 		Name:     name,
-		ColNames: []string{"a", "b", "c", "d", "e", "f", "g", "h", "i", "j", "k", "l", "m"},
+		ColNames: []string{"a", "b", "c", "d", "e", "f", "g", "h", "i", "j", "k", "l", "m", "n", "o"},
 		ColDefs: map[string]schema.Column{
 			"a": {Name: "a", Type: schema.Type{Name: "int"}},
 			"b": {Name: "b", Type: schema.Type{Name: "float"}},
@@ -46,6 +47,8 @@ func TestToSpannerType(t *testing.T) {
 			"k": {Name: "k", Type: schema.Type{Name: "nvarchar", Mods: []int64{50}}},
 			"l": {Name: "l", Type: schema.Type{Name: "image"}},
 			"m": {Name: "m", Type: schema.Type{Name: "geometry"}},
+			"n": {Name: "n", Type: schema.Type{Name: "bit"}},
+			"o": {Name: "o", Type: schema.Type{Name: "uniqueidentifier"}},
 		},
 		PrimaryKeys: []schema.Key{{Column: "a"}},
 		ForeignKeys: []schema.ForeignKey{{Name: "fk_test", Columns: []string{"d"}, ReferTable: "ref_table", ReferColumns: []string{"dref"}},
@@ -78,7 +81,7 @@ func TestToSpannerType(t *testing.T) {
 	dropComments(&actual) // Don't test comment.
 	expected := ddl.CreateTable{
 		Name:     name,
-		ColNames: []string{"a", "b", "c", "d", "e", "f", "g", "h", "i", "j", "k", "l", "m"},
+		ColNames: []string{"a", "b", "c", "d", "e", "f", "g", "h", "i", "j", "k", "l", "m", "n", "o"},
 		ColDefs: map[string]ddl.ColumnDef{
 			"a": {Name: "a", T: ddl.Type{Name: ddl.Int64}},
 			"b": {Name: "b", T: ddl.Type{Name: ddl.Float64}},
@@ -93,6 +96,100 @@ func TestToSpannerType(t *testing.T) {
 			"k": {Name: "k", T: ddl.Type{Name: ddl.String, Len: int64(50)}},
 			"l": {Name: "l", T: ddl.Type{Name: ddl.Bytes, Len: ddl.MaxLength}},
 			"m": {Name: "m", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
+			"n": {Name: "n", T: ddl.Type{Name: ddl.Bool}},
+			"o": {Name: "o", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
+		},
+
+		Pks: []ddl.IndexKey{{Col: "a"}},
+		Fks: []ddl.Foreignkey{{Name: "fk_test", Columns: []string{"d"}, ReferTable: "ref_table", ReferColumns: []string{"dref"}},
+			{Name: "fk_test2", Columns: []string{"a"}, ReferTable: "ref_table2", ReferColumns: []string{"aref"}}},
+		Indexes: []ddl.CreateIndex{{Name: "index1", Table: name, Unique: true, Keys: []ddl.IndexKey{{Col: "a", Desc: false}, {Col: "d", Desc: true}}}},
+	}
+	assert.Equal(t, expected, actual)
+	expectedIssues := map[string][]internal.SchemaIssue{
+		"a": {internal.Widened},
+		"b": {internal.Widened},
+		"c": {internal.Widened},
+		"j": {internal.Timestamp},
+		"m": {internal.NoGoodType},
+	}
+	assert.Equal(t, expectedIssues, conv.Issues[name])
+}
+
+// This is just a very basic smoke-test for toExperimentalSpannerType.
+func TestToExperimentalSpannerType(t *testing.T) {
+	conv := internal.MakeConv()
+	conv.SetSchemaMode()
+	conv.TargetDb = constants.TargetExperimentalPostgres
+	name := "test"
+	srcSchema := schema.Table{
+		Name:     name,
+		ColNames: []string{"a", "b", "c", "d", "e", "f", "g", "h", "i", "j", "k", "l", "m", "n", "o"},
+		ColDefs: map[string]schema.Column{
+			"a": {Name: "a", Type: schema.Type{Name: "int"}},
+			"b": {Name: "b", Type: schema.Type{Name: "float"}},
+			"c": {Name: "c", Type: schema.Type{Name: "tinyint", Mods: []int64{1}}},
+			"d": {Name: "d", Type: schema.Type{Name: "varchar", Mods: []int64{6}}},
+			"e": {Name: "e", Type: schema.Type{Name: "numeric"}},
+			"f": {Name: "f", Type: schema.Type{Name: "timestamp"}},
+			"g": {Name: "g", Type: schema.Type{Name: "binary", Mods: []int64{4000}}},
+			"h": {Name: "h", Type: schema.Type{Name: "date"}},
+			"i": {Name: "i", Type: schema.Type{Name: "money"}},
+			"j": {Name: "j", Type: schema.Type{Name: "smalldatetime"}},
+			"k": {Name: "k", Type: schema.Type{Name: "nvarchar", Mods: []int64{50}}},
+			"l": {Name: "l", Type: schema.Type{Name: "image"}},
+			"m": {Name: "m", Type: schema.Type{Name: "geometry"}},
+			"n": {Name: "n", Type: schema.Type{Name: "bit"}},
+			"o": {Name: "o", Type: schema.Type{Name: "uniqueidentifier"}},
+		},
+		PrimaryKeys: []schema.Key{{Column: "a"}},
+		ForeignKeys: []schema.ForeignKey{{Name: "fk_test", Columns: []string{"d"}, ReferTable: "ref_table", ReferColumns: []string{"dref"}},
+			{Name: "fk_test2", Columns: []string{"a"}, ReferTable: "ref_table2", ReferColumns: []string{"aRef"}}},
+		Indexes: []schema.Index{{Name: "index1", Unique: true, Keys: []schema.Key{{Column: "a", Desc: false}, {Column: "d", Desc: true}}}},
+	}
+	conv.SrcSchema[name] = srcSchema
+	conv.SpSchema["ref_table"] = ddl.CreateTable{
+		Name:     "ref_table",
+		ColNames: []string{"dref", "b", "c"},
+		ColDefs: map[string]ddl.ColumnDef{
+			"dref": {Name: "dref", T: ddl.Type{Name: ddl.String, Len: int64(6)}},
+			"b":    {Name: "b", T: ddl.Type{Name: ddl.Float64}},
+			"c":    {Name: "c", T: ddl.Type{Name: ddl.Bool}},
+		},
+		Pks: []ddl.IndexKey{{Col: "dref"}},
+	}
+	conv.SpSchema["ref_table2"] = ddl.CreateTable{
+		Name:     "ref_table2",
+		ColNames: []string{"aref", "b", "c"},
+		ColDefs: map[string]ddl.ColumnDef{
+			"aref": {Name: "aref", T: ddl.Type{Name: ddl.Int64}},
+			"b":    {Name: "b", T: ddl.Type{Name: ddl.Float64}},
+			"c":    {Name: "c", T: ddl.Type{Name: ddl.Bool}},
+		},
+		Pks: []ddl.IndexKey{{Col: "aref"}},
+	}
+	assert.Nil(t, common.SchemaToSpannerDDL(conv, ToDdlImpl{}))
+	actual := conv.SpSchema[name]
+	dropComments(&actual) // Don't test comment.
+	expected := ddl.CreateTable{
+		Name:     name,
+		ColNames: []string{"a", "b", "c", "d", "e", "f", "g", "h", "i", "j", "k", "l", "m", "n", "o"},
+		ColDefs: map[string]ddl.ColumnDef{
+			"a": {Name: "a", T: ddl.Type{Name: ddl.Int64}},
+			"b": {Name: "b", T: ddl.Type{Name: ddl.Float64}},
+			"c": {Name: "c", T: ddl.Type{Name: ddl.Int64}},
+			"d": {Name: "d", T: ddl.Type{Name: ddl.String, Len: int64(6)}},
+			"e": {Name: "e", T: ddl.Type{Name: ddl.Numeric}},
+			"f": {Name: "f", T: ddl.Type{Name: ddl.Int64}},
+			"g": {Name: "g", T: ddl.Type{Name: ddl.Bytes, Len: ddl.MaxLength}},
+			"h": {Name: "h", T: ddl.Type{Name: ddl.Date}},
+			"i": {Name: "i", T: ddl.Type{Name: ddl.Numeric}},
+			"j": {Name: "j", T: ddl.Type{Name: ddl.Timestamp}},
+			"k": {Name: "k", T: ddl.Type{Name: ddl.String, Len: int64(50)}},
+			"l": {Name: "l", T: ddl.Type{Name: ddl.Bytes, Len: ddl.MaxLength}},
+			"m": {Name: "m", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
+			"n": {Name: "n", T: ddl.Type{Name: ddl.Bool}},
+			"o": {Name: "o", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
 		},
 
 		Pks: []ddl.IndexKey{{Col: "a"}},