Repository: cloudspannerecosystem/harbourbridge
Pull request #302 (2022-05-19 04:52:49)

Oracle number mapping update
- Maps Oracle's NUMBER type to Spanner's STRING type if the source has precision > 29 or scale > 9


anupam-searce: Adding screenshots of the conversion for reference

![image](https://user-images.githubusercontent.com/92134223/169223966-f69331f2-72b5-4912-ba96-df1a43f508b2.png)

![image](https://user-images.githubusercontent.com/92134223/169224034-7788ab20-f392-49b2-bc72-5438ba32273a.png)


Patch:
diff --git a/.github/workflows/chat-notification.yaml b/.github/workflows/chat-notification.yaml
new file mode 100644
index 0000000..88030de
--- /dev/null
+++ b/.github/workflows/chat-notification.yaml
@@ -0,0 +1,30 @@
+# 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.
+
+on:
+  push:
+    branches:
+      - master
+name: notification
+jobs:
+  notify:
+    runs-on: ubuntu-latest
+    steps:
+    - name: Google Chat Notification
+      uses: Co-qn/google-chat-notification@releases/v1
+      with:
+        name: ${{ github.event_name }} by ${{ github.actor }}
+        url: ${{ secrets.GOOGLE_CHAT_WEBHOOK }}
+        status: ${{ job.status }}
+
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 ab616f7..731f5b8 100644
--- a/README.md
+++ b/README.md
@@ -374,9 +374,9 @@ such as `csv`, `avro` etc.
 HarbourBridge accepts the following options for --target-profile,
 specified as "key1=value1,key2=value,..." pairs:
 
-`dbname` Specifies the name of the Spanner database to create. This must be a
-new database. If dbname is not specified, HarbourBridge creates a new unique
-dbname.
+`dbName` Specifies the name of the Spanner database to create. This must be a
+new database. If dbName is not specified, HarbourBridge creates a new unique
+dbName.
 
 `instance` Specifies the Spanner instance to use. The new database will be
 created in this instance. If not specified, the tool automatically determines an
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/common.go b/cmd/common.go
new file mode 100644
index 0000000..f359156
--- /dev/null
+++ b/cmd/common.go
@@ -0,0 +1,73 @@
+package cmd
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	sp "cloud.google.com/go/spanner"
+	"github.com/cloudspannerecosystem/harbourbridge/common/utils"
+	"github.com/cloudspannerecosystem/harbourbridge/conversion"
+	"github.com/cloudspannerecosystem/harbourbridge/internal"
+	"github.com/cloudspannerecosystem/harbourbridge/profiles"
+	"github.com/cloudspannerecosystem/harbourbridge/spanner/writer"
+	"github.com/cloudspannerecosystem/harbourbridge/streaming"
+)
+
+func startDatastream(ctx context.Context, sourceProfile profiles.SourceProfile, targetProfile profiles.TargetProfile) (streaming.StreamingCfg, error) {
+	streamingCfg, err := streaming.ReadStreamingConfig(sourceProfile.Conn.Mysql.StreamingConfig, targetProfile.Conn.Sp.Dbname)
+	if err != nil {
+		return streamingCfg, fmt.Errorf("error reading streaming config: %v", err)
+	}
+
+	err = streaming.LaunchStream(ctx, sourceProfile, targetProfile.Conn.Sp.Project, streamingCfg.DatastreamCfg)
+	if err != nil {
+		return streamingCfg, fmt.Errorf("error launching stream: %v", err)
+	}
+	return streamingCfg, nil
+}
+
+func performSnapshotMigration(ctx context.Context, sourceProfile profiles.SourceProfile, targetProfile profiles.TargetProfile, ioHelper utils.IOStreams, client *sp.Client, conv *internal.Conv, writeLimit int64, dbURI string) (*writer.BatchWriter, error) {
+	dataCoversionStartTime := time.Now()
+	bw, err := conversion.DataConv(ctx, sourceProfile, targetProfile, &ioHelper, client, conv, true, writeLimit)
+	if err != nil {
+		return bw, fmt.Errorf("can't finish data conversion for db %s: %v", dbURI, err)
+	}
+	dataCoversionEndTime := time.Now()
+	dataCoversionDuration := dataCoversionEndTime.Sub(dataCoversionStartTime)
+	conv.Audit.DataConversionDuration = dataCoversionDuration
+	return bw, nil
+}
+
+func startDataflow(ctx context.Context, sourceProfile profiles.SourceProfile, targetProfile profiles.TargetProfile, streamingCfg streaming.StreamingCfg) error {
+	err := streaming.LaunchDataflowJob(ctx, targetProfile, streamingCfg.DatastreamCfg, streamingCfg.DataflowCfg)
+	if err != nil {
+		return fmt.Errorf("error launching dataflow: %v", err)
+	}
+	return nil
+}
+
+func migrateData(ctx context.Context, sourceProfile profiles.SourceProfile, targetProfile profiles.TargetProfile, ioHelper utils.IOStreams, client *sp.Client, conv *internal.Conv, writeLimit int64, dbURI string) (*writer.BatchWriter, error) {
+	streamingCfg := streaming.StreamingCfg{}
+	var err error
+	if sourceProfile.Ty == profiles.SourceProfileTypeConnection && sourceProfile.Conn.Streaming {
+		streamingCfg, err = startDatastream(ctx, sourceProfile, targetProfile)
+		if err != nil {
+			err = fmt.Errorf("error starting datastream: %v", err)
+			return nil, err
+		}
+	}
+	bw, err := performSnapshotMigration(ctx, sourceProfile, targetProfile, ioHelper, client, conv, writeLimit, dbURI)
+	if err != nil {
+		err = fmt.Errorf("can't do snapshot migration: %v", err)
+		return nil, err
+	}
+	if sourceProfile.Ty == profiles.SourceProfileTypeConnection && sourceProfile.Conn.Streaming {
+		err = startDataflow(ctx, sourceProfile, targetProfile, streamingCfg)
+		if err != nil {
+			err = fmt.Errorf("error starting dataflow: %v", err)
+			return nil, err
+		}
+	}
+	return bw, nil
+}
diff --git a/cmd/data.go b/cmd/data.go
index 2773145..4d31caa 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)
@@ -165,10 +165,9 @@ 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)
+	bw, err := migrateData(ctx, sourceProfile, targetProfile, ioHelper, client, conv, cmd.writeLimit, dbURI)
 	if err != nil {
-		err = fmt.Errorf("can't finish data conversion for db %s: %v", dbURI, err)
+		err = fmt.Errorf("can't do data migration: %v", err)
 		return subcommands.ExitFailure
 	}
 	if !cmd.skipForeignKeys {
@@ -177,9 +176,7 @@ func (cmd *DataCmd) Execute(ctx context.Context, f *flag.FlagSet, _ ...interface
 			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..4cb8ea2 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,11 +165,9 @@ func (cmd *SchemaAndDataCmd) Execute(ctx context.Context, f *flag.FlagSet, _ ...
 		return subcommands.ExitFailure
 	}
 
-	schemaCoversionEndTime := time.Now()
-	conv.SchemaConversionDuration = schemaCoversionEndTime.Sub(schemaConversionStartTime)
-	bw, err := conversion.DataConv(ctx, sourceProfile, targetProfile, &ioHelper, client, conv, true, cmd.writeLimit)
+	bw, err := migrateData(ctx, sourceProfile, targetProfile, ioHelper, client, conv, cmd.writeLimit, dbURI)
 	if err != nil {
-		err = fmt.Errorf("can't finish data conversion for db %s: %v", dbURI, err)
+		err = fmt.Errorf("can't do data migration: %v", err)
 		return subcommands.ExitFailure
 	}
 	if !cmd.skipForeignKeys {
@@ -176,8 +176,6 @@ func (cmd *SchemaAndDataCmd) Execute(ctx context.Context, f *flag.FlagSet, _ ...
 			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 202457c..81e7ca5 100644
--- a/conversion/conversion.go
+++ b/conversion/conversion.go
@@ -74,14 +74,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 +87,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,
@@ -264,7 +253,7 @@ func dataFromDump(driver string, config writer.BatchWriterConfig, ioHelper *util
 
 func dataFromCSV(ctx context.Context, sourceProfile profiles.SourceProfile, targetProfile profiles.TargetProfile, config writer.BatchWriterConfig, conv *internal.Conv, client *sp.Client) (*writer.BatchWriter, error) {
 	if targetProfile.Conn.Sp.Dbname == "" {
-		return nil, fmt.Errorf("dbname is mandatory in target-profile for csv source")
+		return nil, fmt.Errorf("dbName is mandatory in target-profile for csv source")
 	}
 	conv.TargetDb = targetProfile.ToLegacyTargetDb()
 	dialect, err := targetProfile.FetchTargetDialect(ctx)
diff --git a/go.mod b/go.mod
index 6146bb9..8905860 100644
--- a/go.mod
+++ b/go.mod
@@ -4,7 +4,8 @@ go 1.13
 
 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..6aeef52 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=
@@ -1033,8 +1039,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 +1058,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 +1159,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 +1299,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 +1386,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 +1422,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 +1438,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/performance/README.md b/performance/README.md
index 733694a..a777610 100644
--- a/performance/README.md
+++ b/performance/README.md
@@ -43,8 +43,8 @@ Run migration for the test database created.
 We can specify the maximum number of writer threads in write-limit flag.
 Enter the password for connecting to MYSQL db and run the below command.
     ```sh
-    go run main.go schema-and-data -source=mysql -source-profile='host=localhost,user=root,db_name=testdb,password='
-     -target-profile='instance=new-test-instance,dbname=testdb' -write-limit 40
+    go run main.go schema-and-data -source=mysql -source-profile='host=localhost,user=root,dbName=testdb,password='
+     -target-profile='instance=new-test-instance,dbName=testdb' -write-limit 40
     ```
 - **Cleanup:**
 Once the migration is complete, the last step is cleanup. It involves deleting the spanner instance and droping the MYSQL database.
diff --git a/performance/setup.sh b/performance/setup.sh
index 3097118..9132352 100644
--- a/performance/setup.sh
+++ b/performance/setup.sh
@@ -37,7 +37,7 @@ if type mysql >/dev/null 2>&1; then
             do
                 echo "Write limit: $writeLimit"
                 #update source profile password before running the benchmark
-                go run main.go schema-and-data -source=mysql -source-profile='host=localhost,user=root,db_name=testdb,password=' -target-profile='instance=new-test-instance,dbname=testdb' -write-limit $writeLimit
+                go run main.go schema-and-data -source=mysql -source-profile='host=localhost,user=root,dbName=testdb,password=' -target-profile='instance=new-test-instance,dbName=testdb' -write-limit $writeLimit
             done
             yes | gcloud spanner instances delete test-instance
         done
@@ -57,7 +57,7 @@ if type mysql >/dev/null 2>&1; then
             do
                 echo "Write limit: $writeLimit"
                 #update source profile password before running the benchmark
-                go run main.go schema-and-data -source=mysql -source-profile='host=localhost,user=root,db_name=testdb,password=' -target-profile='instance=new-test-instance,dbname=testdb' -write-limit $writeLimit
+                go run main.go schema-and-data -source=mysql -source-profile='host=localhost,user=root,dbName=testdb,password=' -target-profile='instance=new-test-instance,dbName=testdb' -write-limit $writeLimit
             done
             yes | gcloud spanner instances delete test-instance
         done
diff --git a/profiles/common.go b/profiles/common.go
index 14f1855..16828ae 100644
--- a/profiles/common.go
+++ b/profiles/common.go
@@ -91,8 +91,8 @@ func GeneratePGSQLConnectionStr() (string, error) {
 	server := os.Getenv("PGHOST")
 	port := os.Getenv("PGPORT")
 	user := os.Getenv("PGUSER")
-	dbname := os.Getenv("PGDATABASE")
-	if server == "" || port == "" || user == "" || dbname == "" {
+	dbName := os.Getenv("PGDATABASE")
+	if server == "" || port == "" || user == "" || dbName == "" {
 		fmt.Printf("Please specify host, port, user and database using PGHOST, PGPORT, PGUSER and PGDATABASE environment variables\n")
 		return "", fmt.Errorf("could not connect to source database")
 	}
@@ -100,19 +100,19 @@ func GeneratePGSQLConnectionStr() (string, error) {
 	if password == "" {
 		password = utils.GetPassword()
 	}
-	return getPGSQLConnectionStr(server, port, user, password, dbname), nil
+	return getPGSQLConnectionStr(server, port, user, password, dbName), nil
 }
 
-func getPGSQLConnectionStr(server, port, user, password, dbname string) string {
-	return fmt.Sprintf("host=%s port=%s user=%s password=%s dbname=%s sslmode=disable", server, port, user, password, dbname)
+func getPGSQLConnectionStr(server, port, user, password, dbName string) string {
+	return fmt.Sprintf("host=%s port=%s user=%s password=%s dbname=%s sslmode=disable", server, port, user, password, dbName)
 }
 
 func GenerateMYSQLConnectionStr() (string, error) {
 	server := os.Getenv("MYSQLHOST")
 	port := os.Getenv("MYSQLPORT")
 	user := os.Getenv("MYSQLUSER")
-	dbname := os.Getenv("MYSQLDATABASE")
-	if server == "" || port == "" || user == "" || dbname == "" {
+	dbName := os.Getenv("MYSQLDATABASE")
+	if server == "" || port == "" || user == "" || dbName == "" {
 		fmt.Printf("Please specify host, port, user and database using MYSQLHOST, MYSQLPORT, MYSQLUSER and MYSQLDATABASE environment variables\n")
 		return "", fmt.Errorf("could not connect to source database")
 	}
@@ -120,15 +120,15 @@ func GenerateMYSQLConnectionStr() (string, error) {
 	if password == "" {
 		password = utils.GetPassword()
 	}
-	return getMYSQLConnectionStr(server, port, user, password, dbname), nil
+	return getMYSQLConnectionStr(server, port, user, password, dbName), nil
 }
 
-func getMYSQLConnectionStr(server, port, user, password, dbname string) string {
-	return fmt.Sprintf("%s:%s@tcp(%s:%s)/%s", user, password, server, port, dbname)
+func getMYSQLConnectionStr(server, port, user, password, dbName string) string {
+	return fmt.Sprintf("%s:%s@tcp(%s:%s)/%s", user, password, server, port, dbName)
 }
 
-func getSQLSERVERConnectionStr(server, port, user, password, dbname string) string {
-	return fmt.Sprintf(`sqlserver://%s:%s@%s:%s?database=%s`, user, password, server, port, dbname)
+func getSQLSERVERConnectionStr(server, port, user, password, dbName string) string {
+	return fmt.Sprintf(`sqlserver://%s:%s@%s:%s?database=%s`, user, password, server, port, dbName)
 }
 
 func GetSchemaSampleSize(sourceProfile SourceProfile) int64 {
@@ -143,7 +143,7 @@ func GetSchemaSampleSize(sourceProfile SourceProfile) int64 {
 	return schemaSampleSize
 }
 
-func getORACLEConnectionStr(server, port, user, password, dbname string) string {
+func getORACLEConnectionStr(server, port, user, password, dbName string) string {
 	portNumber, _ := strconv.Atoi(port)
-	return go_ora.BuildUrl(server, portNumber, dbname, user, password, nil)
+	return go_ora.BuildUrl(server, portNumber, dbName, user, password, nil)
 }
diff --git a/profiles/source_profile.go b/profiles/source_profile.go
index 68b6103..ef4d47a 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["db_name"]
+	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) {
@@ -96,21 +105,21 @@ func NewSourceProfileConnectionMySQL(params map[string]string) (SourceProfileCon
 			return mysql, fmt.Errorf("found empty string for MYSQLHOST/MYSQLUSER/MYSQLDATABASE. Please specify these environment variables with correct values")
 		}
 	} else if hostOk && userOk && dbOk {
-		// If atleast host, username and dbname are provided through source-profile,
+		// If atleast host, username and dbName are provided through source-profile,
 		// go ahead and use source-profile. Port and password handled later even if they are empty.
 		mysql.Host, mysql.User, mysql.Db, mysql.Port, mysql.Pwd = host, user, db, port, pwd
 		// Throw error if the input entered is empty.
 		if mysql.Host == "" || mysql.User == "" || mysql.Db == "" {
-			return mysql, fmt.Errorf("found empty string for host/user/db_name. Please specify host, port, user and db_name in the source-profile")
+			return mysql, fmt.Errorf("found empty string for host/user/dbName. Please specify host, port, user and dbName in the source-profile")
 		}
 	} else {
 		// Partial params provided through source-profile. Ask user to provide all through the source-profile.
-		return mysql, fmt.Errorf("please specify host, port, user and db_name in the source-profile")
+		return mysql, fmt.Errorf("please specify host, port, user and dbName in the source-profile")
 	}
 
 	// Throw same error if the input entered is empty.
 	if mysql.Host == "" || mysql.User == "" || mysql.Db == "" {
-		return mysql, fmt.Errorf("found empty string for host/user/db. please specify host, port, user and db_name in the source-profile")
+		return mysql, fmt.Errorf("found empty string for host/user/db. please specify host, port, user and dbName in the source-profile")
 	}
 
 	if mysql.Port == "" {
@@ -136,7 +145,7 @@ func NewSourceProfileConnectionPostgreSQL(params map[string]string) (SourceProfi
 	pg := SourceProfileConnectionPostgreSQL{}
 	host, hostOk := params["host"]
 	user, userOk := params["user"]
-	db, dbOk := params["db_name"]
+	db, dbOk := params["dbName"]
 	port, portOk := params["port"]
 	pwd, pwdOk := params["password"]
 	// We don't users to mix and match params from source-profile and environment variables.
@@ -159,11 +168,11 @@ func NewSourceProfileConnectionPostgreSQL(params map[string]string) (SourceProfi
 		pg.Host, pg.User, pg.Db, pg.Port, pg.Pwd = host, user, db, port, pwd
 		// Throw error if the input entered is empty.
 		if pg.Host == "" || pg.User == "" || pg.Db == "" {
-			return pg, fmt.Errorf("found empty string for host/user/db_name. Please specify host, port, user and db_name in the source-profile")
+			return pg, fmt.Errorf("found empty string for host/user/dbName. Please specify host, port, user and dbName in the source-profile")
 		}
 	} else {
 		// Partial params provided through source-profile. Ask user to provide all through the source-profile.
-		return pg, fmt.Errorf("please specify host, port, user and db_name in the source-profile")
+		return pg, fmt.Errorf("please specify host, port, user and dbName in the source-profile")
 	}
 
 	if pg.Port == "" {
@@ -189,7 +198,7 @@ func NewSourceProfileConnectionSqlServer(params map[string]string) (SourceProfil
 	ss := SourceProfileConnectionSqlServer{}
 	host, hostOk := params["host"]
 	user, userOk := params["user"]
-	db, dbOk := params["db_name"]
+	db, dbOk := params["dbName"]
 	port, portOk := params["port"]
 	pwd, pwdOk := params["password"]
 
@@ -218,11 +227,11 @@ func NewSourceProfileConnectionSqlServer(params map[string]string) (SourceProfil
 		ss.Host, ss.User, ss.Db, ss.Port, ss.Pwd = host, user, db, port, pwd
 		// Throw error if the input entered is empty.
 		if ss.Host == "" || ss.User == "" || ss.Db == "" {
-			return ss, fmt.Errorf("found empty string for host/user/db_name. Please specify host, port, user and db_name in the source-profile")
+			return ss, fmt.Errorf("found empty string for host/user/dbName. Please specify host, port, user and dbName in the source-profile")
 		}
 	} else {
 		// Partial params provided through source-profile. Ask user to provide all through the source-profile.
-		return ss, fmt.Errorf("please specify host, port, user and db_name in the source-profile")
+		return ss, fmt.Errorf("please specify host, port, user and dbName in the source-profile")
 	}
 
 	if ss.Port == "" {
@@ -283,31 +292,38 @@ func NewSourceProfileConnectionDynamoDB(params map[string]string) (SourceProfile
 }
 
 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) {
 	ss := SourceProfileConnectionOracle{}
 	host, hostOk := params["host"]
 	user, userOk := params["user"]
-	db, dbOk := params["db_name"]
+	db, dbOk := params["dbName"]
 	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
 		// Throw error if the input entered is empty.
 		if ss.Host == "" || ss.User == "" || ss.Db == "" {
-			return ss, fmt.Errorf("found empty string for host/user/db_name. Please specify host, port, user and db_name in the source-profile")
+			return ss, fmt.Errorf("found empty string for host/user/dbName. Please specify host, port, user and dbName in the source-profile")
 		}
 	} else {
 		// Partial params provided through source-profile. Ask user to provide all through the source-profile.
-		return ss, fmt.Errorf("please specify host, port, user and db_name in the source-profile")
+		return ss, fmt.Errorf("please specify host, port, user and dbName in the source-profile")
 	}
 
 	if ss.Port == "" {
@@ -323,6 +339,7 @@ func NewSourceProfileConnectionOracle(params map[string]string) (SourceProfileCo
 
 type SourceProfileConnection struct {
 	Ty        SourceProfileConnectionType
+	Streaming bool
 	Mysql     SourceProfileConnectionMySQL
 	Pg        SourceProfileConnectionPostgreSQL
 	Dydb      SourceProfileConnectionDynamoDB
@@ -341,6 +358,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":
 		{
@@ -374,6 +394,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 +466,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/profiles/target_profile.go b/profiles/target_profile.go
index cdd9539..d8d8983 100644
--- a/profiles/target_profile.go
+++ b/profiles/target_profile.go
@@ -82,7 +82,7 @@ func (trg TargetProfile) ToLegacyTargetDb() string {
 	}
 }
 
-// This expects that GetResourceIds has already been called once and the project, instance and dbname
+// This expects that GetResourceIds has already been called once and the project, instance and dbName
 // fields in target profile are populated.
 func (trg TargetProfile) FetchTargetDialect(ctx context.Context) (string, error) {
 	// TODO: consider moving all clients to target profile instead of passing them around the codebase.
@@ -141,11 +141,11 @@ func (targetProfile *TargetProfile) GetResourceIds(ctx context.Context, now time
 // correspond to regular Cloud Spanner database and PG Cloud Spanner database
 // respectively.
 //
-// If dbname is not specified, then HarbourBridge will autogenerate the same
+// If dbName is not specified, then HarbourBridge will autogenerate the same
 // and create a database with the same name.
 //
-// Example: -target-profile="instance=my-instance1,dbname=my-new-db1"
-// Example: -target-profile="instance=my-instance1,dbname=my-new-db1,dialect=PostgreSQL"
+// Example: -target-profile="instance=my-instance1,dbName=my-new-db1"
+// Example: -target-profile="instance=my-instance1,dbName=my-new-db1,dialect=PostgreSQL"
 //
 func NewTargetProfile(s string) (TargetProfile, error) {
 	params, err := parseProfile(s)
@@ -163,8 +163,8 @@ func NewTargetProfile(s string) (TargetProfile, error) {
 	if instance, ok := params["instance"]; ok {
 		sp.Instance = instance
 	}
-	if dbname, ok := params["dbname"]; ok {
-		sp.Dbname = dbname
+	if dbName, ok := params["dbName"]; ok {
+		sp.Dbname = dbName
 	}
 	if dialect, ok := params["dialect"]; ok {
 		sp.Dialect = dialect
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/csv/README.md b/sources/csv/README.md
index 0157300..057b758 100644
--- a/sources/csv/README.md
+++ b/sources/csv/README.md
@@ -20,7 +20,7 @@ Harbourbridge will migrate `table_name.csv` to a Spanner table named
 `table_name` in the database specified via the target profile.
 
 ```sh
-harbourbridge data -source=csv -target-profile="instance=my-instance,dbname=my-db,dialect=postgresql" 
+harbourbridge data -source=csv -target-profile="instance=my-instance,dbName=my-db,dialect=postgresql" 
 ```
 
 - **Providing a manifest input:**
@@ -29,7 +29,7 @@ locations (local system as well as Google Cloud Storage). You can also provide
 multiple csv file paths for a single table using the manifest.
 
 ```sh
-harbourbridge data -source=csv -source-profile="manifest=path/to/manifest/file" -target-profile="instance=my-instance,dbname=my-db" 
+harbourbridge data -source=csv -source-profile="manifest=path/to/manifest/file" -target-profile="instance=my-instance,dbName=my-db" 
 ```
 
 ### Manifest File
@@ -81,7 +81,7 @@ profile.
 For example, if you want to use `|` as the delimiter and `NULL` as the null value, 
 you can use 
 ```sh
-harbourbridge data -source=csv -source-profile="delimiter=|,nullStr=NULL" -target-profile="instance=my-instance,dbname=my-db" 
+harbourbridge data -source=csv -source-profile="delimiter=|,nullStr=NULL" -target-profile="instance=my-instance,dbName=my-db" 
 ```
 
 
diff --git a/sources/mysql/README.md b/sources/mysql/README.md
index da1306b..9d85083 100644
--- a/sources/mysql/README.md
+++ b/sources/mysql/README.md
@@ -36,7 +36,7 @@ By default, HarbourBridge will generate a new Spanner database name to populate.
 You can override this and specify the database name to use by:
 
 ```sh
-harbourbridge data -session=mydb.session.json -source=mysql -target-profile="instance=my-spanner-instance,dbname=my-spanner-database-name" < my_mysqldump_file
+harbourbridge data -session=mydb.session.json -source=mysql -target-profile="instance=my-spanner-instance,dbName=my-spanner-database-name" < my_mysqldump_file
 ```
 
 You can also run HarbourBridge in a schema-and-data mode, where it will perform both
@@ -69,12 +69,12 @@ that HarbourBridge will not create directories as it writes these files.
 
 In this case, HarbourBridge connects directly to the MySQL database to retrieve
 table schema and data. Set the `-source=mysql` and corresponding source profile
-connection parameters `host`, `port`, `user`, `db_name` and `password`.
+connection parameters `host`, `port`, `user`, `dbName` and `password`.
 
 For example to perform schema conversion, run
 
 ```sh
-harbourbridge schema -source=mysql -source-profile="host=<>,port=<>,user=<>,db_name=<>"
+harbourbridge schema -source=mysql -source-profile="host=<>,port=<>,user=<>,dbName=<>"
 ```
 
 Parameters `port` and `password` are optional. Port (`port`) defaults to `3306`
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/README.md b/sources/oracle/README.md
index 326f8b1..9c0a3ae 100644
--- a/sources/oracle/README.md
+++ b/sources/oracle/README.md
@@ -20,18 +20,18 @@ in the [Installing HarbourBridge](https://github.com/cloudspannerecosystem/harbo
 
 In this case, HarbourBridge connects directly to the Oracle database to
 retrieve table schema and data. Set the `-source=oracle` and corresponding
-source profile connection parameters `host`, `port`, `user`, `db_name` and
+source profile connection parameters `host`, `port`, `user`, `dbName` and
 `password`.
 
 For example to perform schema conversion, run
 
 ```sh
-harbourbridge schema -source=oracle -source-profile="host=<>,port=<>,user=<>,db_name=<>,password=<>"
+harbourbridge schema -source=oracle -source-profile="host=<>,port=<>,user=<>,dbName=<>,password=<>"
 ```
 
 In Oracle DB, USER is the account name, SCHEMA is the set of objects owned by that user. Oracle creates the SCHEMA object as part of the CREATE USER statement and the SCHEMA has the same name as the USER. 
 
-db_name will be the SID of the Database used. The Oracle System ID (SID) is used to uniquely identify a particular database on a system.
+dbName will be the SID of the Database used. The Oracle System ID (SID) is used to uniquely identify a particular database on a system.
 
 ## Schema Conversion
 
diff --git a/sources/oracle/infoschema.go b/sources/oracle/infoschema.go
index 31786ed..b311fa5 100644
--- a/sources/oracle/infoschema.go
+++ b/sources/oracle/infoschema.go
@@ -66,6 +66,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":
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/README.md b/sources/postgres/README.md
index eb293d4..b295295 100644
--- a/sources/postgres/README.md
+++ b/sources/postgres/README.md
@@ -39,7 +39,7 @@ By default, HarbourBridge will generate a new Spanner database name to populate.
 You can override this and specify the database name to use by:
 
 ```sh
-harbourbridge data -session=mydb.session.json -source=pg -target-profile="instance=my-spanner-instance,dbname=my-spanner-database-name" < my_pg_dump_file
+harbourbridge data -session=mydb.session.json -source=pg -target-profile="instance=my-spanner-instance,dbName=my-spanner-database-name" < my_pg_dump_file
 ```
 
 You can also run HarbourBridge in a schema-and-data mode, where it will perform both
@@ -72,13 +72,13 @@ that HarbourBridge will not create directories as it writes these files.
 
 In this case, HarbourBridge connects directly to the PostgreSQL database to
 retrieve table schema and data. Set the `-source=postgres` and corresponding
-source profile connection parameters `host`, `port`, `user`, `db_name` and
+source profile connection parameters `host`, `port`, `user`, `dbName` and
 `password`.
 
 For example to perform schema conversion, run
 
 ```sh
-harbourbridge schema -source=postgres -source-profile="host=<>,port=<>,user=<>,db_name=<>"
+harbourbridge schema -source=postgres -source-profile="host=<>,port=<>,user=<>,dbName=<>"
 ```
 
 Parameters `port` and `password` are optional. Port (`port`) defaults to `5432`
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/sqlserver/README.md b/sources/sqlserver/README.md
index 963cd38..96c8024 100644
--- a/sources/sqlserver/README.md
+++ b/sources/sqlserver/README.md
@@ -16,17 +16,17 @@ HarbourBridge can be run directly on a SQL Server database (via go's database/sq
 The following examples assume a `harbourbridge` alias has been setup as described
 in the [Installing HarbourBridge](https://github.com/cloudspannerecosystem/harbourbridge#installing-harbourbridge) section of the main README.
 
-### Directly connecting to a Sql server database
+### Directly connecting to a SQL Server database
 
-In this case, HarbourBridge connects directly to the Sql server database to
+In this case, HarbourBridge connects directly to the SQL Server database to
 retrieve table schema and data. Set the `-source=sqlserver` and corresponding
-source profile connection parameters `host`, `port`, `user`, `db_name` and
+source profile connection parameters `host`, `port`, `user`, `dbName` and
 `password`.
 
 For example to perform schema conversion, run
 
 ```sh
-harbourbridge schema -source=sqlserver -source-profile="host=<>,port=<>,user=<>,db_name=<>"
+harbourbridge schema -source=sqlserver -source-profile="host=<>,port=<>,user=<>,dbName=<>"
 ```
 
 Parameters `port` and `password` are optional. Port (`port`) defaults to `1433`
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..6831e95
--- /dev/null
+++ b/streaming/streaming.go
@@ -0,0 +1,289 @@
+// 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)
+
+	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("Stream state set to RUNNING...")
+	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
+}
diff --git a/ui/package-lock.json b/ui/package-lock.json
index 2dad57b..212ca54 100644
--- a/ui/package-lock.json
+++ b/ui/package-lock.json
@@ -3211,9 +3211,9 @@
       }
     },
     "node_modules/async": {
-      "version": "2.6.3",
-      "resolved": "https://registry.npmjs.org/async/-/async-2.6.3.tgz",
-      "integrity": "sha512-zflvls11DCy+dQWzTW2dzuilv8Z5X/pjfmZOWba6TNIVDm+2UDaJmXSOXlasHKfNBs8oo3M0aT50fDEWfKZjXg==",
+      "version": "2.6.4",
+      "resolved": "https://registry.npmjs.org/async/-/async-2.6.4.tgz",
+      "integrity": "sha512-mzo5dfJYwAn29PeiJ0zvwTo04zj8HDJj0Mn8TD7sno7q12prdbnasKJHhkm2c1LgrhlJ0teaea8860oxi51mGA==",
       "dev": true,
       "dependencies": {
         "lodash": "^4.17.14"
@@ -7632,9 +7632,9 @@
       }
     },
     "node_modules/minimist": {
-      "version": "1.2.5",
-      "resolved": "https://registry.npmjs.org/minimist/-/minimist-1.2.5.tgz",
-      "integrity": "sha512-FM9nNUYrRBAELZQT3xeZQ7fmMOBg6nWNmJKTcgsJeaLstP/UODVpGsr5OhXhhXg6f+qtJ8uiZ+PUxkDWcgIXLw==",
+      "version": "1.2.6",
+      "resolved": "https://registry.npmjs.org/minimist/-/minimist-1.2.6.tgz",
+      "integrity": "sha512-Jsjnk4bw3YJqYzbdyBiNsPWHPfO++UGG749Cxs6peCu5Xg4nrena6OVxOYxrQTqww0Jmwt+Ref8rggumkTLz9Q==",
       "dev": true
     },
     "node_modules/minipass": {
@@ -7858,9 +7858,9 @@
       "optional": true
     },
     "node_modules/node-forge": {
-      "version": "1.2.1",
-      "resolved": "https://registry.npmjs.org/node-forge/-/node-forge-1.2.1.tgz",
-      "integrity": "sha512-Fcvtbb+zBcZXbTTVwqGA5W+MKBj56UjVRevvchv5XrcyXbmNdesfZL37nlcWOfpgHhgmxApw3tQbTr4CqNmX4w==",
+      "version": "1.3.1",
+      "resolved": "https://registry.npmjs.org/node-forge/-/node-forge-1.3.1.tgz",
+      "integrity": "sha512-dPEtOeMvF9VMcYV/1Wb8CPoVAXtp6MKMlcbAt4ddqmGqUJ6fQZFXkNZNkNlfevtNkGtaSoXf/vNNNSvgrdXwtA==",
       "dev": true,
       "engines": {
         "node": ">= 6.13.0"
@@ -13739,9 +13739,9 @@
       "dev": true
     },
     "async": {
-      "version": "2.6.3",
-      "resolved": "https://registry.npmjs.org/async/-/async-2.6.3.tgz",
-      "integrity": "sha512-zflvls11DCy+dQWzTW2dzuilv8Z5X/pjfmZOWba6TNIVDm+2UDaJmXSOXlasHKfNBs8oo3M0aT50fDEWfKZjXg==",
+      "version": "2.6.4",
+      "resolved": "https://registry.npmjs.org/async/-/async-2.6.4.tgz",
+      "integrity": "sha512-mzo5dfJYwAn29PeiJ0zvwTo04zj8HDJj0Mn8TD7sno7q12prdbnasKJHhkm2c1LgrhlJ0teaea8860oxi51mGA==",
       "dev": true,
       "requires": {
         "lodash": "^4.17.14"
@@ -16974,9 +16974,9 @@
       }
     },
     "minimist": {
-      "version": "1.2.5",
-      "resolved": "https://registry.npmjs.org/minimist/-/minimist-1.2.5.tgz",
-      "integrity": "sha512-FM9nNUYrRBAELZQT3xeZQ7fmMOBg6nWNmJKTcgsJeaLstP/UODVpGsr5OhXhhXg6f+qtJ8uiZ+PUxkDWcgIXLw==",
+      "version": "1.2.6",
+      "resolved": "https://registry.npmjs.org/minimist/-/minimist-1.2.6.tgz",
+      "integrity": "sha512-Jsjnk4bw3YJqYzbdyBiNsPWHPfO++UGG749Cxs6peCu5Xg4nrena6OVxOYxrQTqww0Jmwt+Ref8rggumkTLz9Q==",
       "dev": true
     },
     "minipass": {
@@ -17151,9 +17151,9 @@
       "optional": true
     },
     "node-forge": {
-      "version": "1.2.1",
-      "resolved": "https://registry.npmjs.org/node-forge/-/node-forge-1.2.1.tgz",
-      "integrity": "sha512-Fcvtbb+zBcZXbTTVwqGA5W+MKBj56UjVRevvchv5XrcyXbmNdesfZL37nlcWOfpgHhgmxApw3tQbTr4CqNmX4w==",
+      "version": "1.3.1",
+      "resolved": "https://registry.npmjs.org/node-forge/-/node-forge-1.3.1.tgz",
+      "integrity": "sha512-dPEtOeMvF9VMcYV/1Wb8CPoVAXtp6MKMlcbAt4ddqmGqUJ6fQZFXkNZNkNlfevtNkGtaSoXf/vNNNSvgrdXwtA==",
       "dev": true
     },
     "node-gyp": {


Test patch:
diff --git a/.github/workflows/test-coverage.yaml b/.github/workflows/test-coverage.yaml
new file mode 100644
index 0000000..0c37ae7
--- /dev/null
+++ b/.github/workflows/test-coverage.yaml
@@ -0,0 +1,49 @@
+# 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.
+
+on:
+  push:
+    branches:
+      - master
+  pull_request:
+name: code-coverage-check  
+jobs:
+  coverage:
+    runs-on: ubuntu-latest
+    steps:
+    - uses: actions/checkout@v2
+    - uses: actions/setup-go@v2
+      with:
+        go-version: '1.16'
+
+    - name: Unit tests
+      run: |
+            go test ./... -coverprofile coverage.out -covermode count
+            go tool cover -func coverage.out
+
+    - name: Quality Gate - Test coverage should be above threshold
+      env:
+          TESTCOVERAGE_THRESHOLD: 50
+      run: |
+            echo "Quality Gate: checking if test coverage is above threshold ..."
+            echo "Threshold             : $TESTCOVERAGE_THRESHOLD %"
+            totalCoverage=`go tool cover -func=coverage.out | grep total | grep -Eo '[0-9]+\.[0-9]+'`
+            echo "Current test coverage : $totalCoverage %"
+            if (( $(echo "$totalCoverage $TESTCOVERAGE_THRESHOLD" | awk '{print ($1 > $2)}') )); then
+                echo "OK"
+            else
+                echo "Current test coverage is below threshold. Please add more unit tests or adjust threshold to a lower value."
+                echo "Failed"
+                exit 1
+            fi
diff --git a/profiles/source_profile_test.go b/profiles/source_profile_test.go
index e3a488c..a06dc39 100644
--- a/profiles/source_profile_test.go
+++ b/profiles/source_profile_test.go
@@ -77,12 +77,12 @@ func TestNewSourceProfileConnectionSQL(t *testing.T) {
 	}{
 		{
 			name:          "mandatory params provided",
-			params:        map[string]string{"host": "a", "user": "b", "db_name": "c", "password": "e"},
+			params:        map[string]string{"host": "a", "user": "b", "dbName": "c", "password": "e"},
 			errorExpected: false,
 		},
 		{
 			name:          "partial mandatory params provided",
-			params:        map[string]string{"user": "b", "db_name": "c"},
+			params:        map[string]string{"user": "b", "dbName": "c"},
 			errorExpected: true,
 		},
 		{
@@ -97,17 +97,17 @@ func TestNewSourceProfileConnectionSQL(t *testing.T) {
 		},
 		{
 			name:          "all params provided",
-			params:        map[string]string{"host": "a", "user": "b", "db_name": "c", "port": "d", "password": "e"},
+			params:        map[string]string{"host": "a", "user": "b", "dbName": "c", "port": "d", "password": "e"},
 			errorExpected: false,
 		},
 		{
 			name:          "empty mandatory param",
-			params:        map[string]string{"host": "", "user": "b", "db_name": "c"},
+			params:        map[string]string{"host": "", "user": "b", "dbName": "c"},
 			errorExpected: true,
 		},
 		{
 			name:          "empty port",
-			params:        map[string]string{"host": "a", "user": "b", "db_name": "c", "password": "e"},
+			params:        map[string]string{"host": "a", "user": "b", "dbName": "c", "password": "e"},
 			errorExpected: false,
 		},
 	}
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/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/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"}},
diff --git a/testing/csv/integration_test.go b/testing/csv/integration_test.go
index 7f43667..4603cdc 100644
--- a/testing/csv/integration_test.go
+++ b/testing/csv/integration_test.go
@@ -165,7 +165,7 @@ func TestIntegration_CSV_Command(t *testing.T) {
 	writeCSVs(t)
 	defer cleanupCSVs()
 	createSpannerSchema(t, projectID, instanceID, dbName)
-	args := fmt.Sprintf("data -source=csv -target-profile='instance=%s,dbname=%s'", instanceID, dbName)
+	args := fmt.Sprintf("data -source=csv -target-profile='instance=%s,dbName=%s'", instanceID, dbName)
 	err := common.RunCommand(args, projectID)
 	if err != nil {
 		t.Fatal(err)
diff --git a/testing/mysql/integration_test.go b/testing/mysql/integration_test.go
index c1979b6..1c1539a 100644
--- a/testing/mysql/integration_test.go
+++ b/testing/mysql/integration_test.go
@@ -140,9 +140,9 @@ func TestIntegration_MYSQL_SchemaAndDataSubcommand(t *testing.T) {
 	dbURI := fmt.Sprintf("projects/%s/instances/%s/databases/%s", projectID, instanceID, dbName)
 	filePrefix := filepath.Join(tmpdir, dbName+".")
 
-	host, user, db_name, password := os.Getenv("MYSQLHOST"), os.Getenv("MYSQLUSER"), os.Getenv("MYSQLDATABASE"), os.Getenv("MYSQLPWD")
+	host, user, srcDb, password := os.Getenv("MYSQLHOST"), os.Getenv("MYSQLUSER"), os.Getenv("MYSQLDATABASE"), os.Getenv("MYSQLPWD")
 	envVars := common.ClearEnvVariables([]string{"MYSQLHOST", "MYSQLUSER", "MYSQLDATABASE", "MYSQLPWD"})
-	args := fmt.Sprintf("schema-and-data -source=%s -prefix=%s -source-profile='host=%s,user=%s,db_name=%s,password=%s' -target-profile='instance=%s,dbname=%s'", constants.MYSQL, filePrefix, host, user, db_name, password, instanceID, dbName)
+	args := fmt.Sprintf("schema-and-data -source=%s -prefix=%s -source-profile='host=%s,user=%s,dbName=%s,password=%s' -target-profile='instance=%s,dbName=%s'", constants.MYSQL, filePrefix, host, user, srcDb, password, instanceID, dbName)
 	err := common.RunCommand(args, projectID)
 	common.RestoreEnvVariables(envVars)
 	if err != nil {
@@ -248,7 +248,7 @@ func TestIntegration_MySQLDUMP_DataOnly(t *testing.T) {
 }
 
 func runSchemaSubcommand(t *testing.T, dbName, filePrefix, sessionFile, dumpFilePath string) {
-	args := fmt.Sprintf("schema -prefix %s -source=mysql -target-profile='dbname=%s' < %s", filePrefix, dbName, dumpFilePath)
+	args := fmt.Sprintf("schema -prefix %s -source=mysql -target-profile='dbName=%s' < %s", filePrefix, dbName, dumpFilePath)
 	err := common.RunCommand(args, projectID)
 	if err != nil {
 		t.Fatal(err)
@@ -256,7 +256,7 @@ func runSchemaSubcommand(t *testing.T, dbName, filePrefix, sessionFile, dumpFile
 }
 
 func runDataSubcommand(t *testing.T, dbName, dbURI, filePrefix, sessionFile, dumpFilePath string) {
-	args := fmt.Sprintf("data -source=mysql -prefix %s -session %s -target-profile='instance=%s,dbname=%s' < %s", filePrefix, sessionFile, instanceID, dbName, dumpFilePath)
+	args := fmt.Sprintf("data -source=mysql -prefix %s -session %s -target-profile='instance=%s,dbName=%s' < %s", filePrefix, sessionFile, instanceID, dbName, dumpFilePath)
 	err := common.RunCommand(args, projectID)
 	if err != nil {
 		t.Fatal(err)
@@ -264,7 +264,7 @@ func runDataSubcommand(t *testing.T, dbName, dbURI, filePrefix, sessionFile, dum
 }
 
 func runSchemaAndDataSubcommand(t *testing.T, dbName, dbURI, filePrefix, dumpFilePath string) {
-	args := fmt.Sprintf("schema-and-data -source=mysql -prefix %s -target-profile='instance=%s,dbname=%s' < %s", filePrefix, instanceID, dbName, dumpFilePath)
+	args := fmt.Sprintf("schema-and-data -source=mysql -prefix %s -target-profile='instance=%s,dbName=%s' < %s", filePrefix, instanceID, dbName, dumpFilePath)
 	err := common.RunCommand(args, projectID)
 	if err != nil {
 		t.Fatal(err)
@@ -273,7 +273,7 @@ func runSchemaAndDataSubcommand(t *testing.T, dbName, dbURI, filePrefix, dumpFil
 
 func runDataOnlySubcommandForSessionFile(t *testing.T, dbName, dbURI, sessionFile string) {
 	host, user, password := os.Getenv("MYSQLHOST"), os.Getenv("MYSQLUSER"), os.Getenv("MYSQLPWD")
-	args := fmt.Sprintf("data -source=mysql -session %s -source-profile='host=%s,user=%s,db_name=%s,password=%s' -target-profile='instance=%s,dbname=%s' ", sessionFile, host, user, dbName, password, instanceID, dbName)
+	args := fmt.Sprintf("data -source=mysql -session %s -source-profile='host=%s,user=%s,dbName=%s,password=%s' -target-profile='instance=%s,dbName=%s' ", sessionFile, host, user, dbName, password, instanceID, dbName)
 	err := common.RunCommand(args, projectID)
 	if err != nil {
 		t.Fatal(err)
diff --git a/testing/oracle/integration_test.go b/testing/oracle/integration_test.go
index ee5c344..21a77bb 100644
--- a/testing/oracle/integration_test.go
+++ b/testing/oracle/integration_test.go
@@ -123,7 +123,7 @@ func TestIntegration_ORACLE_SchemaSubcommand(t *testing.T) {
 	defer os.RemoveAll(tmpdir)
 	filePrefix := filepath.Join(tmpdir, "Oracle_IntTest.")
 
-	args := fmt.Sprintf("schema -prefix %s -source=%s -source-profile='host=localhost,user=STI,db_name=xe,password=test1,port=1521'", filePrefix, constants.ORACLE)
+	args := fmt.Sprintf("schema -prefix %s -source=%s -source-profile='host=localhost,user=STI,dbName=xe,password=test1,port=1521'", filePrefix, constants.ORACLE)
 	err := common.RunCommand(args, projectID)
 	if err != nil {
 		t.Fatal(err)
@@ -138,7 +138,7 @@ func TestIntegration_ORACLE_SchemaAndDataSubcommand(t *testing.T) {
 	dbURI := fmt.Sprintf("projects/%s/instances/%s/databases/%s", projectID, instanceID, dbName)
 	filePrefix := filepath.Join(tmpdir, "Oracle_IntTest.")
 
-	args := fmt.Sprintf("schema-and-data -prefix %s -source=%s  -source-profile='host=localhost,user=STI,db_name=xe,password=test1,port=1521' -target-profile='instance=%s,dbname=%s'", filePrefix, constants.ORACLE, instanceID, dbName)
+	args := fmt.Sprintf("schema-and-data -prefix %s -source=%s  -source-profile='host=localhost,user=STI,dbName=xe,password=test1,port=1521' -target-profile='instance=%s,dbName=%s'", filePrefix, constants.ORACLE, instanceID, dbName)
 	err := common.RunCommand(args, projectID)
 	if err != nil {
 		t.Fatal(err)
diff --git a/testing/postgres/integration_test.go b/testing/postgres/integration_test.go
index 6696a15..52e8daf 100644
--- a/testing/postgres/integration_test.go
+++ b/testing/postgres/integration_test.go
@@ -142,7 +142,7 @@ func TestIntegration_PGDUMP_SchemaAndDataSubcommand(t *testing.T) {
 	dataFilepath := "../../test_data/pg_dump.test.out"
 	filePrefix := filepath.Join(tmpdir, dbName+".")
 
-	args := fmt.Sprintf("schema-and-data -prefix %s -source=postgres -target-profile='instance=%s,dbname=%s' < %s", filePrefix, instanceID, dbName, dataFilepath)
+	args := fmt.Sprintf("schema-and-data -prefix %s -source=postgres -target-profile='instance=%s,dbName=%s' < %s", filePrefix, instanceID, dbName, dataFilepath)
 	err := common.RunCommand(args, projectID)
 	if err != nil {
 		t.Fatal(err)
@@ -203,7 +203,7 @@ func TestIntegration_POSTGRES_SchemaAndDataSubcommand(t *testing.T) {
 	dbURI := fmt.Sprintf("projects/%s/instances/%s/databases/%s", projectID, instanceID, dbName)
 	filePrefix := filepath.Join(tmpdir, dbName+".")
 
-	args := fmt.Sprintf("schema-and-data -prefix %s -source=postgres -target-profile='instance=%s,dbname=%s'", filePrefix, instanceID, dbName)
+	args := fmt.Sprintf("schema-and-data -prefix %s -source=postgres -target-profile='instance=%s,dbName=%s'", filePrefix, instanceID, dbName)
 	err := common.RunCommand(args, projectID)
 	if err != nil {
 		t.Fatal(err)
diff --git a/testing/sqlserver/integration_test.go b/testing/sqlserver/integration_test.go
index 3f26f10..096608b 100644
--- a/testing/sqlserver/integration_test.go
+++ b/testing/sqlserver/integration_test.go
@@ -111,7 +111,7 @@ func TestIntegration_SQLserver_SchemaSubcommand(t *testing.T) {
 	defer os.RemoveAll(tmpdir)
 	filePrefix := filepath.Join(tmpdir, "SqlServer_IntTest.")
 
-	args := fmt.Sprintf("schema -prefix %s -source=sqlserver -source-profile='host=localhost,user=sa,db_name=SqlServer_IntTest'", filePrefix)
+	args := fmt.Sprintf("schema -prefix %s -source=sqlserver -source-profile='host=localhost,user=sa,dbName=SqlServer_IntTest'", filePrefix)
 	err := common.RunCommand(args, projectID)
 	if err != nil {
 		t.Fatal(err)
@@ -126,7 +126,7 @@ func TestIntegration_SQLserver_SchemaAndDataSubcommand(t *testing.T) {
 	dbURI := fmt.Sprintf("projects/%s/instances/%s/databases/%s", projectID, instanceID, dbName)
 	filePrefix := filepath.Join(tmpdir, "SqlServer_IntTest.")
 
-	args := fmt.Sprintf("schema-and-data -prefix %s -source=%s  -source-profile='host=localhost,user=sa,db_name=SqlServer_IntTest' -target-profile='instance=%s,dbname=%s'", filePrefix, constants.SQLSERVER, instanceID, dbName)
+	args := fmt.Sprintf("schema-and-data -prefix %s -source=%s  -source-profile='host=localhost,user=sa,dbName=SqlServer_IntTest' -target-profile='instance=%s,dbName=%s'", filePrefix, constants.SQLSERVER, instanceID, dbName)
 	err := common.RunCommand(args, projectID)
 	if err != nil {
 		t.Fatal(err)