Repository: cloudspannerecosystem/harbourbridge
Pull request #286 (2022-04-25 08:20:07)

Support for streaming migration via HarbourBridge
Support for streaming migration by automatic triggering of Dataflow and Datastream jobs. It takes care of when to launch the respective jobs when performing a streaming migration. It takes a streaming config file as input via the source profile parameter: _streamingCfg_. 
Sample streamingCfg:
```
{ 
	"datastreamCfg": {
		"streamId": "deep-hb-stream-1", //optional field
		"streamLocation": "us-central1",
		"streamDisplayName": "deep-hb-stream-1",  //optional field
		"sourceConnectionConfig": {
			"name": "deep-conn",
			"location": "us-central1"
		},
		"destinationConnectionConfig": {
			"name": "deep-conn-dst",
			"location": "us-central1"
		}
	},
	"dataflowCfg": {
		"JobName": "deep-df-hb-1",  //optional field
		"Location": "us-central1"
	}
}
```

**Things not handled:**
- Support for connection profile creation. Stream is created via existing profiles that is taken as an input to HarbourBridge. 
- Cleanup of the dataflow and datastream jobs.



- [ ] Appropriate changes to README are included in PR


Patch:
diff --git a/Makefile b/Makefile
new file mode 100644
index 0000000..ad832ec
--- /dev/null
+++ b/Makefile
@@ -0,0 +1,13 @@
+# Build the default binary
+build:
+	go build
+# Build a static binary
+build-static:
+	go build -a -tags osusergo,netgo -ldflags '-w -extldflags "-static"' -o harbourbridge main.go
+# Run unit tests
+test:
+	go test -v ./...
+# Run code coverage with unit tests
+test-coverage:
+	go test ./... -coverprofile coverage.out -covermode count
+	go tool cover -func coverage.out
diff --git a/cmd/common.go b/cmd/common.go
new file mode 100644
index 0000000..32ab9e1
--- /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.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..79a848f 100644
--- a/cmd/data.go
+++ b/cmd/data.go
@@ -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_and_data.go b/cmd/schema_and_data.go
index 056c71b..caa958f 100644
--- a/cmd/schema_and_data.go
+++ b/cmd/schema_and_data.go
@@ -125,6 +125,8 @@ func (cmd *SchemaAndDataCmd) Execute(ctx context.Context, f *flag.FlagSet, _ ...
 	if err != nil {
 		panic(err)
 	}
+	schemaCoversionEndTime := time.Now()
+	conv.SchemaConversionDuration = schemaCoversionEndTime.Sub(schemaConversionStartTime)
 
 	// Populate migration request id and migration type in conv object
 	conv.MigrationRequestId = "HB-" + uuid.New().String()
@@ -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/utils/utils.go b/common/utils/utils.go
index 97185a9..4081b2d 100644
--- a/common/utils/utils.go
+++ b/common/utils/utils.go
@@ -249,10 +249,10 @@ func GetPassword() string {
 
 // GetDatabaseName generates database name with driver_date prefix.
 func GetDatabaseName(driver string, now time.Time) (string, error) {
-	return generateName(fmt.Sprintf("%s_%s", driver, now.Format("2006-01-02")))
+	return GenerateName(fmt.Sprintf("%s_%s", driver, now.Format("2006-01-02")))
 }
 
-func generateName(prefix string) (string, error) {
+func GenerateName(prefix string) (string, error) {
 	b := make([]byte, 4)
 	_, err := rand.Read(b)
 	if err != nil {
diff --git a/conversion/conversion.go b/conversion/conversion.go
index 2f2dd7b..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,
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/report.go b/internal/report.go
index 2b27a23..34d29ac 100644
--- a/internal/report.go
+++ b/internal/report.go
@@ -325,6 +325,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 +333,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 +368,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 {
diff --git a/profiles/source_profile.go b/profiles/source_profile.go
index 7d85ae3..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["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) {
@@ -283,11 +292,12 @@ 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) {
@@ -298,6 +308,12 @@ func NewSourceProfileConnectionOracle(params map[string]string) (SourceProfileCo
 	port, _ := params["port"]
 	pwd, _ := params["password"]
 
+	streamingConfig, cfgOk := params["streamingCfg"]
+	if cfgOk && streamingConfig == "" {
+		return ss, fmt.Errorf("specify a non-empty streaming config file path")
+	}
+	ss.StreamingConfig = streamingConfig
+
 	if hostOk && userOk && dbOk {
 		// All connection params provided through source-profile. Port and password handled later.
 		ss.Host, ss.User, ss.Db, ss.Port, ss.Pwd = host, user, db, port, pwd
@@ -323,6 +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/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/toddl.go b/sources/oracle/toddl.go
index dfb7f66..e6807e0 100644
--- a/sources/oracle/toddl.go
+++ b/sources/oracle/toddl.go
@@ -158,9 +158,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/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/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
+}


Test patch:
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..0f6eb9b 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
@@ -115,7 +115,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 +125,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..93c54b2 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,
@@ -96,7 +96,7 @@ See github.com/pganalyze/pg_query_go for definitions of statement types
 ----------------------------
 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 +114,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 +124,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"}},